本文为您介绍如何使用云数据库HBase连接器。
背景信息
云数据库HBase是低成本、高扩展、云智能的大数据NoSQL,兼容标准HBase访问协议,提供低成本存储、高扩展吞吐、智能数据处理等核心优势,是为淘宝推荐、花呗风控、广告投放、监控大屏、菜鸟物流轨迹、支付宝账单、手淘消息等众多阿里巴巴核心服务提供支撑的数据库,具备PB规模、高并发、秒级伸缩、毫秒响应、跨机房高可用、全托管、全球分布等企业能力。
HBase连接器支持的信息如下:
类别 | 详情 |
支持类型 | 维表和结果表 |
运行模式 | 流模式 |
数据格式 | 暂不支持 |
特有监控指标 |
|
API种类 | SQL |
是否支持更新或删除结果表数据 | 是 |
前提条件
注意事项
使用前,请确认已创建数据库实例类型,并选择正确的连接器,使用不当的连接器可能会导致不可预期的问题:
云数据库HBase实例,使用本文的HBase连接器。
Lindorm实例兼容HBase模式,使用Lindorm连接器,详情请参见云原生多模数据库Lindorm。
如果连接开源HBase,则无法保证数据的正确性。
语法结构
CREATE TABLE hbase_table(
rowkey INT,
family1 ROW<q1 INT>,
family2 ROW<q2 STRING, q3 BIGINT>,
family3 ROW<q4 DOUBLE, q5 BOOLEAN, q6 STRING>
) WITH (
'connector'='cloudhbase',
'table-name'='<yourTableName>',
'zookeeper.quorum'='<yourZookeeperQuorum>'
);
HBase的列族(Column Family)必须声明为ROW类型,列族名即该ROW的字段名。例如,DDL定义中声明了family1、family2和family3三个列族。
HBase列族中的列(Column)与对应ROW中嵌套的每个字段对应,列名即字段名。例如,DDL定义中列族family2声明了q2和q3两列。
除了类型为ROW的字段外,只能有一个原始类型(Atomic Type)的字段(例如STRING或BIGINT),该字段将被视作HBase的行键(Row Key),例如DDL定义中的Rowkey。
必须将HBase的行键定义为结果表的主键(Primary Key),如果没有显示定义主键,默认使用行键作为主键。
结果表中不需要将HBase表的所有列族和列都进行声明,只声明需要的即可。
WITH参数
通用
参数
说明
数据类型
是否必填
默认值
备注
connector
表类型。
String
是
无
固定值为
cloudhbase
。table-name
HBase表名。
String
是
无
无。
zookeeper.znode.quorum
HBase的zookeeper住址。
String
是
无
无。
zookeeper.znode.parent
HBase在zookeeper中的根目录。
String
否
/hbase
仅在HBase标准版中生效。
userName
用户名。
String
否
无
仅在HBase增强版中生效。
password
密码。
String
否
无
仅在HBase增强版中生效。
haclient.cluster.id
HBase高可用实例ID。
String
否
无
只有访问同城主备实例时才需要配置仅在HBase增强版中生效。
retires.number
HBase客户端的重试次数。
Integer
否
31
无。
null-string-literal
HBase字段类型为字符串时,如果Flink字段数据为null,则将该字段赋值为
null-string-literal
,并写入HBase。String
否
null
无。
结果表独有
参数
说明
数据类型
是否必填
默认值
备注
sink.buffer-flush.max-size
写入HBase前,内存中缓存的数据量(字节)大小。调大该值有利于提高HBase写入性能,但会增加写入延迟和内存使用。
String
否
2MB
支持字节单位B、KB、MB和GB,不区分大小写。设置为0表示不进行缓存。
sink.buffer-flush.max-rows
写入HBase前,内存中缓存的数据条数。调大该值有利于提高HBase写入性能,但会增加写入延迟和内存使用。
Integer
否
1000
设置为0表示不进行缓存。
sink.buffer-flush.interval
将缓存数据周期性写入到HBase的间隔,可以控制写入HBase的延迟。
Duration
否
1s
支持时间单位ms、s、min、h和d。设置为0表示关闭定期写入。
dynamic.table
是否使用支持动态列的HBase表。
Boolean
否
false
参数取值如下:
true:使用支持动态列的HBase表。
false:不使用支持动态列的HBase表。
sink.ignore-delete
是否忽略撤回消息。
Boolean
否
false
参数取值如下:
true:忽略撤回消息。
false:不忽略撤回消息。
说明仅实时计算引擎VVR 4.0.10及以上版本支持该参数。
sink.sync-write
是否同步写入HBase。
Boolean
否
true
参数取值如下:
true:同步写,保证顺序,会牺牲一定性能。
false:异步写,不保证顺序,性能更好。
说明仅实时计算引擎VVR 4.0.13及以上版本支持该参数。
sink.buffer-flush.batch-rows
同步写入HBase时内存中缓存的数据条数,调大该值有利于提高HBase写入性能,但会增加写入延迟和内存使用。
Integer
否
100
仅当sink.sync-write为true时生效。
说明仅实时计算引擎VVR 4.0.13及以上版本支持该参数。
sink.ignore-null
是否忽略写入null值。
Boolean
否
false
说明设置成true时,参数
null-string-literal
将不再生效。仅实时计算引擎VVR 8.0.9及以上版本支持该参数。
维表独有(比如Cache参数)
参数
说明
数据类型
是否必填
默认值
备注
cache
缓存策略。
String
否
ALL
目前云数据库HBase版维表支持以下三种缓存策略:
None:无缓存。
LRU:缓存维表里的部分数据。源表的每条数据都会触发系统先在Cache中查找数据,如果没有找到,则去物理维表中查找。
说明需要配置相关参数:缓存大小(cacheSize)和缓存更新时间间隔(cacheTTLMs)。
ALL(默认值):缓存维表里的所有数据。在Job运行前,系统会将维表中所有数据加载到Cache中,之后所有的维表查找数据都会通过Cache进行。如果在Cache中无法找到数据,则KEY不存在,并在Cache过期后重新加载一遍全量Cache。
说明适用于远程表数据量小且MISS KEY(源表数据和维表JOIN时,ON条件无法关联)特别多的场景。需要配置相关参数:缓存更新时间间隔cacheTTLMs,更新时间黑名单cacheReloadTimeBlackList。
维表中所有数据加载到缓存中,可能会导致作业启动变慢,您可以根据业务需求灵活配置缓存策略。
因为系统会异步加载维表数据,所以在使用CACHE ALL时,需要增加维表JOIN节点的内存,增加的内存大小为远程表数据量的两倍。
cacheSize
缓存大小。
Long
否
10000
当缓存策略选择LRU时,可以设置缓存大小。
cacheTTLMs
缓存失效时间,单位为毫秒。
Long
否
无
cacheTTLMs配置和cache有关:
如果cache配置为None,则cacheTTLMs可以不配置,表示缓存不超时。
如果cache配置为LRU,则cacheTTLMs为缓存超时时间。默认不过期。
如果cache配置为ALL,则cacheTTLMs为缓存加载时间。默认不重新加载。
cacheEmpty
是否缓存空结果。
Boolean
否
true
无。
cacheReloadTimeBlackList
更新时间黑名单。在缓存策略选择为ALL时,启用更新时间黑名单,防止在此时间内做Cache更新(例如双11场景)。
String
否
无
格式为2017-10-24 14:00 -> 2017-10-24 15:00,2017-11-10 23:30 -> 2017-11-11 08:00。分隔符的使用情况如下所示:
用英文逗号(,)来分隔多个黑名单。
用箭头(->)来分割黑名单的起始结束时间。
cacheScanLimit
读取全量HBase数据,RPC(Remote Procedure Call Protocol)服务端一次返回给客户端的行数。
Integer
否
100
缓存策略选择ALL时启用。
类型映射
Flink中的数据类型在HBase中通过org.apache.hadoop.hbase.util.Bytes
转换成字节数组,解码过程有以下两种情况:
对于Flink的非字符串类型,如果HBase中的值为空字节数组,则解码为null。
对于Flink的字符串类型,如果HBase中的值为
null-string-literal
字节数组,则解码为null。
Flink SQL类型 | 写入Bytes时CloudHBase转换函数 | 从CloudHBase读取Bytes的转换函数 |
CHAR | byte[] toBytes(String s) | String toString(byte[] b) |
VARCHAR | ||
STRING | ||
BOOLEAN | byte[] toBytes(boolean b) | boolean toBoolean(byte[] b) |
BINARY | byte[] | byte[] |
VARBINARY | ||
DECIMAL | byte[] toBytes(BigDecimal v) | BigDecimal toBigDecimal(byte[] b) |
TINYINT | new byte[] { val } | bytes[0] |
SMALLINT | byte[] toBytes(short val) | short toShort(byte[] bytes) |
INT | byte[] toBytes(int val) | int toInt(byte[] bytes) |
BIGINT | byte[] toBytes(long val) | long toLong(byte[] bytes) |
FLOAT | byte[] toBytes(float val) | float toFloat(byte[] bytes) |
DOUBLE | byte[] toBytes(double val) | double toDouble(byte[] bytes) |
DATE | 将日期转换成自1970.01.01以来的天数,用int表示,并通过 | HBase字节数组通过 |
TIME | 将时间转换成自00:00:00以来的毫秒数,用int表示,并通过 | HBase字节数组通过 |
TIMESTAMP | 将时间戳转换成自1970-01-01 00:00:00以来的毫秒数,用long表示,并通过 | HBase字节数组通过 |
代码示例
维表示例。
CREATE TEMPORARY TABLE datagen_source ( a INT, b BIGINT, c STRING, `proc_time` AS PROCTIME() ) WITH ( 'connector'='datagen' ); CREATE TEMPORARY TABLE hbase_dim ( rowkey INT, family1 ROW<col1 INT>, family2 ROW<col1 STRING, col2 BIGINT>, family3 ROW<col1 DOUBLE, col2 BOOLEAN, col3 STRING> ) WITH ( 'connector' = 'cloudhbase', 'table-name' = '<yourTableName>', 'zookeeper.quorum' = '<yourZookeeperQuorum>' ); CREATE TEMPORARY TABLE blackhole_sink( a INT, f1c1 INT, f3c3 STRING ) WITH ( 'connector' = 'blackhole' ); INSERT INTO blackhole_sink SELECT a, family1.col1 as f1c1, family3.col3 as f3c3 FROM datagen_source JOIN hbase_dim FOR SYSTEM_TIME AS OF datagen_source.`proc_time` as h ON datagen_source.a = h.rowkey;
结果表示例。
CREATE TEMPORARY TABLE datagen_source ( rowkey INT, f1q1 INT, f2q1 STRING, f2q2 BIGINT, f3q1 DOUBLE, f3q2 BOOLEAN, f3q3 STRING ) WITH ( 'connector'='datagen' ); CREATE TEMPORARY TABLE hbase_sink ( rowkey INT, family1 ROW<q1 INT>, family2 ROW<q1 STRING, q2 BIGINT>, family3 ROW<q1 DOUBLE, q2 BOOLEAN, q3 STRING>, PRIMARY KEY (rowkey) NOT ENFORCED ) WITH ( 'connector'='cloudhbase', 'table-name'='<yourTableName>', 'zookeeper.quorum'='<yourZookeeperQuorum>' ); INSERT INTO hbase_sink SELECT rowkey, ROW(f1q1), ROW(f2q1, f2q2), ROW(f3q1, f3q2, f3q3) FROM datagen_source;
结果动态表示例。
CREATE TEMPORARY TABLE datagen_source ( id INT, f1hour STRING, f1deal BIGINT, f2day STRING, f2deal BIGINT ) WITH ( 'connector'='datagen' ); CREATE TEMPORARY TABLE hbase_sink ( rowkey INT, f1 ROW<`hour` STRING, deal BIGINT>, f2 ROW<`day` STRING, deal BIGINT> ) WITH ( 'connector'='cloudhbase', 'table-name'='<yourTableName>', 'zookeeper.quorum'='<yourZookeeperQuorum>', 'dynamic.table'='true' ); INSERT INTO hbase_sink SELECT id, ROW(f1hour, f1deal), ROW(f2day, f2deal) FROM datagen_source;
当dynamic.table参数值为true时,表示使用支持动态列的HBase表。
每个列族对应的ROW中必须声明两个字段:第1个字段的值表示动态列,第2个字段的值表示动态列的值。
如果datagen_source表存在一条数据,代表ID为1的商品,在10:00-11:00点之间的成交额是100,在2020年7月26日当天的成交额是10000,则HBase中将插入行键为1的行,其中f1:10为100,f2:2020-7-26为10000。