HBase

HBase 连接器提供了对分布式 HBase 数据库表的读写数据能力，支持做数据源表、结果表和维表。

使用限制

Flink 目前提供了 HBase-1.4 和 HBase-2.2 两种连接器，请根据实际情况选择：

• 在 Flink 1.11-volcano 引擎版本中仅支持使用 HBase-1.4 连接器。
• 在 Flink 1.16-volcano 引擎版本中支持使用 HBase-1.4 和 HBase-2.2 两种连接器。

注意事项

在公网环境中连接火山 HBase 时，您需要添加以下两个参数：

• 'properties.zookeeper.znode.metaserver' = 'public-meta-region-server'
• 'properties.zookeeper.znode.master' = 'public-master'

DDL 定义

CREATE TABLE hbase_source (
    rowkey INT,               --行键
    family1 ROW<q1 INT>,      --声明列族、列和列名
    family2 ROW<q2 STRING, q3 BIGINT>,
    family3 ROW<q4 DOUBLE, q5 BOOLEAN, q6 VARCHAR>,
    PRIMARY KEY (rowkey) NOT ENFORCED   --行键定义为主键
 ) WITH (
     'connector' = 'hbase-1.4',
     'table-name'='<yourTableName>',
     'zookeeper.quorum'='<yourZookeeperQuorum>'
 );

DDL 定义语句介绍：

• 需要声明 HBase 的行键（Row Key）。
• HBase 的行键需要定义为表的主键（Primary Key），如果没有定义，默认也是行键作为主键。
• HBase 的列族（Column Family）必须声明为 ROW 类型，列族名即该 ROW 的字段名。例如，定义中声明了 family1、family2、family3 三个列族。
• HBase 列族中的列（Cloumn）与对应 ROW 中嵌套的每个字段对应，列名即字段名。例如，列族 family2 中的 q2 和 q3 表示两列的字段名。
• 除了类型为 ROW 的字段外，只能有一个原始类型字段，该字段将被视作 HBase 的行键（Row Key），例如，定义中的 rowkey。

WITH 参数

通用参数

结果表参数

维表参数

数据类型映射

HBase 连接器通过org.apache.hadoop.hbase.util.Bytes提供的实用程序类将 Flink 数据类型转换为字节数组，然后在 Hbase 存储。读取时，把字节数组转为具体类型的对象。
转换方式如下：

示例代码

• 结果表

  CREATE TABLE datagen_source (
       order_id bigint,
       order_product_id bigint,
       order_customer_id bigint,
       order_status varchar,
       order_update_time as localtimestamp
      )
  WITH (
      'connector' = 'datagen',
      'rows-per-second' = '1',
      'fields.order_status.length' = '3',
      'fields.order_id.min' = '1',
      'fields.order_id.max' = '1000',
      'fields.order_product_id.min' = '1',
      'fields.order_product_id.max' = '100',
      'fields.order_customer_id.min' = '1',
      'fields.order_customer_id.max' = '1000'        
  );
  
  CREATE TABLE hbase_sink (
      order_id bigint,
      order_info ROW (
        order_product_id bigint,
        order_customer_id bigint,
        order_status varchar,
        order_update_time TIMESTAMP(0)
          )
      )
  WITH (
    'connector' = 'hbase-1.4',
    'table-name' = 'orders',
    'zookeeper.quorum' = 'hb-a***929-zk.config.ivolces.com:2181'
  );
  
  insert into hbase_sink
  select
    order_id,
    ROW (
      order_product_id,
      order_customer_id,
      order_status,
      order_update_time
    ) as order_info
  from
    datagen_source;
  

• 维表

  CREATE TABLE orders (
      rowkey bigint,
      order_product_id bigint,
      order_customer_id bigint,
      order_status varchar,
      order_update_time as localtimestamp,
      proc_time as PROCTIME ()
    )
  WITH
    (
      'connector' = 'datagen',
      'rows-per-second' = '1',
      'fields.order_status.length' = '3',
      'fields.order_product_id.min' = '1',
      'fields.order_product_id.max' = '100',
      'fields.order_customer_id.min' = '1',
      'fields.order_customer_id.max' = '1000'
    );
  
  
  CREATE TABLE hbase_sink (
      rowkey bigint,
      cf1 ROW (
        order_product_id1 bigint,
        order_customer_id1 bigint,
        order_status1 varchar,
        order_update_time1 TIMESTAMP(0)
      )
    )
  WITH
    (
      'connector' = 'hbase-1.4',
      'table-name' = 'orders',
      'zookeeper.quorum' = 'hb-a***929-zk.config.ivolces.com:2181'
    );
  
  
  create table print_sink (
      order_product_id bigint,
      order_customer_id bigint,
      order_status varchar,
      order_product_id1 bigint,
      order_customer_id1 bigint,
      order_status1 varchar
    )
  with
    ('connector' = 'print');
  
  
  insert into print_sink
  SELECT
    orders.order_product_id,
    orders.order_customer_id,
    orders.order_status,
    hbase_sink.order_product_id1,
    hbase_sink.order_customer_id1,
    hbase_sink.order_status1
  FROM
    orders
    LEFT JOIN hbase_sink FOR SYSTEM_TIME AS OF orders.proc_time ON orders.rowkey = hbase_sink.rowkey;