HBase Result Table

Function

DLI writes job output data to HBase. HBase is a reliable, high-performance, scalable, column-oriented distributed cloud storage system suitable for mass data storage and distributed computing scenarios. You can use HBase to establish a storage system capable of handling terabytes to petabytes of data, easily filtering and analyzing data with millisecond response times, swiftly uncovering data insights. HBase supports structured and semi-structured key-value data storage, including message data, report data, recommendation data, risk management data, log data, and order data. Using DLI, you can efficiently write vast quantities of data into HBase with high speed and minimal latency.

Prerequisites

• An enhanced datasource connection has been created for DLI to connect to HBase, so that jobs can run on the dedicated queue of DLI and you can set the security group rules as required.
  • For details about how to set up an enhanced datasource connection, see Enhanced Datasource Connections.
  • For details about how to configure security group rules, see Security Group Overview.
• If MRS HBase is used, IP addresses of all hosts in the MRS cluster have been added to host information of the enhanced datasource connection.

  For details, see Modifying Host Information.

• In Flink cross-source development scenarios, there is a risk of password leakage if datasource authentication information is directly configured. You are advised to use the datasource authentication provided by DLI.

  For details about datasource authentication, see Datasource Authentication Overview.

Precautions

• When creating a Flink OpenSource SQL job, you need to set Flink Version to 1.12 on the Running Parameters tab of the job editing page, select Save Job Log, and set the OBS bucket for saving job logs.
• The column families in created HBase result table must be declared as the ROW type, the field names map the column family names, and the nested field names map the column qualifier names. There is no need to declare all the families and qualifiers in the schema. Users can declare what is used in the query. Except the ROW type fields, the single atomic type field (for example, STRING or BIGINT) will be recognized as the HBase rowkey. The rowkey field can be an arbitrary name, but should be quoted using backticks if it is a reserved keyword.

Syntax

create table hbaseSink (
  attr_name attr_type 
  (',' attr_name attr_type)* 
  ','PRIMARY KEY (attr_name, ...) NOT ENFORCED)
) with (
  'connector' = 'hbase-2.2',
  'table-name' = '',
  'zookeeper.quorum' = ''
);

Parameters

Data Type Mapping

HBase stores all data as byte arrays. The data needs to be serialized and deserialized during read and write operations.

When serializing and de-serializing, Flink HBase connector uses utility class org.apache.hadoop.hbase.util.Bytes provided by HBase (Hadoop) to convert Flink data types to and from byte arrays.

Flink HBase connector encodes null values to empty bytes, and decodes empty bytes to null values for all data types except the string type. For string type, the null literal is determined by the null-string-literal option.

Example

In this example, data is read from the Kafka data source and written to the HBase result table. The procedure is as follows (the HBase versions used in this example are 1.3.1 and 2.2.3):

1. Create an enhanced datasource connection in the VPC and subnet where HBase and Kafka locate, and bind the connection to the required Flink elastic resource pool. For details, see Enhanced Datasource Connections. Add MRS host information in the enhanced datasource connection by referring to Modifying Host Information.
2. Set HBase and Kafka security groups and add inbound rules to allow access from the Flink queue. Test the connectivity using the HBase and Kafka address by referring to Testing Address Connectivity. If the connection is successful, the datasource is bound to the queue. Otherwise, the binding fails.
3. Use the HBase shell to create HBase table order that has only one column family detail. For details, see Using HBase from Scratch.

   create 'order', {NAME => 'detail'}

4. Create a Flink OpenSource SQL job. Enter the following job script and submit the job. The job script uses Kafka as the data source and HBase as the result table (the Rowkey is order_id and the column family name is detail).
   When you create a job, set Flink Version to 1.12 on the Running Parameters tab. Select Save Job Log, and specify the OBS bucket for saving job logs. Change the values of the parameters in bold as needed in the following script.

   CREATE TABLE orders (
     order_id string,
     order_channel string,
     order_time string,
     pay_amount double,
     real_pay double,
     pay_time string,
     user_id string,
     user_name string,
     area_id string
   ) WITH (
     'connector' = 'kafka',
     'topic' = 'KafkaTopic',
     'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort',
     'properties.group.id' = 'GroupId',
     'scan.startup.mode' = 'latest-offset',
     'format' = 'json'
   );
   
   create table hbaseSink(
     order_id string,
     detail Row(
       order_channel string,
       order_time string,
       pay_amount double,
       real_pay double,
       pay_time string,
       user_id string,
       user_name string,
       area_id string)
   ) with (
     'connector' = 'hbase-2.2',
     'table-name' = 'order',
     'zookeeper.quorum' = 'ZookeeperAddress:ZookeeperPort',
     'sink.buffer-flush.max-rows' = '1'
   );
   
   insert into hbaseSink select order_id, Row(order_channel,order_time,pay_amount,real_pay,pay_time,user_id,user_name,area_id) from orders;

5. Connect to the Kafka cluster and enter the following data to Kafka:

   {"order_id":"202103241000000001", "order_channel":"webShop", "order_time":"2021-03-24 10:00:00", "pay_amount":"100.00", "real_pay":"100.00", "pay_time":"2021-03-24 10:02:03", "user_id":"0001", "user_name":"Alice", "area_id":"330106"}
   
   {"order_id":"202103241606060001", "order_channel":"appShop", "order_time":"2021-03-24 16:06:06", "pay_amount":"200.00", "real_pay":"180.00", "pay_time":"2021-03-24 16:10:06", "user_id":"0001", "user_name":"Alice", "area_id":"330106"}
   
   {"order_id":"202103251202020001", "order_channel":"miniAppShop", "order_time":"2021-03-25 12:02:02", "pay_amount":"60.00", "real_pay":"60.00", "pay_time":"2021-03-25 12:03:00", "user_id":"0002", "user_name":"Bob", "area_id":"330110"}

6. Run the following statement on the HBase shell to view the data result:

    scan 'order'

   The data result is as follows:

   202103241000000001   column=detail:area_id, timestamp=2021-12-16T21:30:37.954, value=330106
   
   202103241000000001   column=detail:order_channel, timestamp=2021-12-16T21:30:37.954, value=webShop
   
   202103241000000001   column=detail:order_time, timestamp=2021-12-16T21:30:37.954, value=2021-03-24 10:00:00
   
   202103241000000001   column=detail:pay_amount, timestamp=2021-12-16T21:30:37.954, value=@Y\x00\x00\x00\x00\x00\x00
   
   202103241000000001   column=detail:pay_time, timestamp=2021-12-16T21:30:37.954, value=2021-03-24 10:02:03
   
   202103241000000001   column=detail:real_pay, timestamp=2021-12-16T21:30:37.954, value=@Y\x00\x00\x00\x00\x00\x00
   
   202103241000000001   column=detail:user_id, timestamp=2021-12-16T21:30:37.954, value=0001
   
   202103241000000001   column=detail:user_name, timestamp=2021-12-16T21:30:37.954, value=Alice
   
   202103241606060001   column=detail:area_id, timestamp=2021-12-16T21:30:44.842, value=330106
   
   202103241606060001   column=detail:order_channel, timestamp=2021-12-16T21:30:44.842, value=appShop
   
   202103241606060001   column=detail:order_time, timestamp=2021-12-16T21:30:44.842, value=2021-03-24 16:06:06
   
   202103241606060001   column=detail:pay_amount, timestamp=2021-12-16T21:30:44.842, value=@i\x00\x00\x00\x00\x00\x00
   
   202103241606060001   column=detail:pay_time, timestamp=2021-12-16T21:30:44.842, value=2021-03-24 16:10:06
   
   202103241606060001   column=detail:real_pay, timestamp=2021-12-16T21:30:44.842, value=@f\x80\x00\x00\x00\x00\x00
   
   202103241606060001   column=detail:user_id, timestamp=2021-12-16T21:30:44.842, value=0001
   
   202103241606060001   column=detail:user_name, timestamp=2021-12-16T21:30:44.842, value=Alice
   
   202103251202020001   column=detail:area_id, timestamp=2021-12-16T21:30:52.181, value=330110
   
   202103251202020001   column=detail:order_channel, timestamp=2021-12-16T21:30:52.181, value=miniAppShop
   
   202103251202020001   column=detail:order_time, timestamp=2021-12-16T21:30:52.181, value=2021-03-25 12:02:02
   
   202103251202020001   column=detail:pay_amount, timestamp=2021-12-16T21:30:52.181, value=@N\x00\x00\x00\x00\x00\x00
   
   202103251202020001   column=detail:pay_time, timestamp=2021-12-16T21:30:52.181, value=2021-03-25 12:03:00
   
   202103251202020001   column=detail:real_pay, timestamp=2021-12-16T21:30:52.181, value=@N\x00\x00\x00\x00\x00\x00
   
   202103251202020001   column=detail:user_id, timestamp=2021-12-16T21:30:52.181, value=0002
   
   202103251202020001   column=detail:user_name, timestamp=2021-12-16T21:30:52.181, value=Bob

FAQ

Q: What should I do if the Flink job execution fails and the log contains the following error information?

org.apache.zookeeper.ClientCnxn$SessionTimeoutException: Client session timed out, have not heard from server in 90069ms for connection id 0x0

A: The datasource connection is not bound or the binding fails. Configure the datasource connection by referring to Enhanced Datasource Connections or configure the security group of the Kafka cluster to allow access from the DLI queue.