实时计算Flink版：云数据库HBase

背景信息

云数据库HBase是低成本、高扩展、云智能的大数据NoSQL，兼容标准HBase访问协议，提供低成本存储、高扩展吞吐、智能数据处理等核心优势，是为淘宝推荐、花呗风控、广告投放、监控大屏、菜鸟物流轨迹、支付宝账单、手淘消息等众多阿里巴巴核心服务提供支撑的数据库，具备PB规模、高并发、秒级伸缩、毫秒响应、跨机房高可用、全托管、全球分布等企业能力。

前提条件

• 已购买HBase集群并创建表，详情请参见购买集群。

• 已设置白名单，详情请参见设置白名单。

注意事项

使用前，请确认已创建数据库实例类型，并选择正确的连接器，使用不当的连接器可能会导致不可预期的问题：

• 云数据库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时启用。

类型映射

代码示例

• 维表示例。

  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。