Creating a FlinkServer Job to Write Data to an HBase Table

Updated on 2024-10-25 GMT+08:00

View PDF

This section applies to MRS 3.1.2 or later clusters.

Scenario

FlinkServer can be interconnected with HBase. The details are as follows:

It can be interconnected with dimension tables and sink tables.
When HBase and Flink are in the same cluster or clusters with mutual trust, FlinkServer can be interconnected with HBase.
If HBase and Flink are in different clusters without mutual trust, Flink in a normal cluster can be interconnected with HBase in a normal cluster.

Prerequisites

The HDFS, Yarn, Flink, and HBase services have been installed in a cluster.
The client that contains the HBase service has been installed, for example, in the /opt/client directory.

Creating a Job

Scenario 1: HBase Functioning as a Sink Table

Create a table on the HBase client.

Log in to the HBase client by referring to Using the HBase Client and run the create'dim_province', "f1" command to create the dim_province table.
Copy the HBase configuration file to the node where FlinkServer is deployed.
1. Log in to the node where the client is installed as the client installation user and copy all configuration files in the /opt/client/HBase/hbase/conf/ directory of HBase to an empty directory of all nodes where FlinkServer is deployed, for example, /tmp/client/HBase/hbase/conf/.
2. Change the owner of the configuration file directory and its upper-layer directory on the FlinkServer node to omm.
  chown omm: /tmp/client/HBase/ -R
  NOTE:
  FlinkServer nodes:
  Log in to Manager, choose Cluster > Services > Flink > Instance, and check the Service IP Address of FlinkServer.
  
  If the node where a FlinkServer instance is located is the node where the HBase client is installed, skip this step on this node.
Add the local path for FlinkServer to access the HBase cluster.

Log in to Manager and choose Cluster > Services > Flink. Click Configurations then All Configurations, search for the HBASE_CONF_DIR parameter, and enter the FlinkServer directory (for example, /tmp/client/HBase/hbase/conf/) to which the HBase configuration files are copied from Value. After the parameters are configured, click Save. After confirming the modification, click OK.
NOTE:

If the node where a FlinkServer instance is deployed is the node where the HBase client is installed, enter the /opt/client/HBase/hbase/conf/ directory of HBase in Value of the HBASE_CONF_DIR parameter.
Restart the affected FlinkServer instance.

Click Instance, select all FlinkServer instances, choose More > Restart Instance, enter the password, and click OK to restart the instances.
Access FlinkServer and create a Flink SQL job.
1. Log in to Manager and choose Cluster > Services > Flink. In the Basic Information area, click the link on the right of Flink WebUI to access the Flink web UI.
2. Create a Flink SQL job and set Task Type to Stream job. For details, see Creating a FlinkServer Job. On the job development page, configure the job parameters as follows and start the job.
  In Basic Parameter, select Enable CheckPoint, set Time Interval(ms) to 60000, and retain the default value for Mode.
  If the cluster is in security mode and the HBase authentication setting is hbase.rpc.protection=authentication, create a Flink SQL job by referring to the following example:
```
CREATE TABLE ksource1 (
user_id STRING,
item_id STRING,
proctime as PROCTIME()
) WITH (
'connector' = 'kafka',
'topic' = 'ksource1',
'properties.group.id' = 'group1',
'properties.bootstrap.servers' = 'IP address of the Kafka broker instance 1:Kafka port number,IP address of the Kafka broker instance 2:Kafka port number',
'format' = 'json',
'properties.sasl.kerberos.service.name' = 'kafka',--This parameter is not required for clusters in normal mode.
'properties.security.protocol' = 'SASL_PLAINTEXT',--This parameter is not required for clusters in normal mode.
'properties.kerberos.domain.name' = 'hadoop.System domain name'--This parameter is not required for clusters in normal mode.
);

CREATE TABLE hsink1 (
rowkey STRING,
f1 ROW < item_id STRING >,
PRIMARY KEY (rowkey) NOT ENFORCED
) WITH (
'connector' = 'hbase-2.2',
'table-name' = 'dim_province',
'zookeeper.quorum' = 'IP address of the ZooKeeper quorumpeer instance 1:ZooKeeper port number,'IP address of the ZooKeeper quorumpeer instance 2:ZooKeeper port number'
);

INSERT INTO
hsink1
SELECT
user_id as rowkey,
ROW(item_id) as f1
FROM
ksource1;
```
  NOTE:
  The IP address and port number of the Kafka broker instance are as follows:
  
  To obtain the instance IP address, log in to FusionInsight Manager, choose Cluster > Services > Kafka, click Instance, and query the instance IP address on the instance list page.
  
  If Kerberos authentication is enabled for the cluster (the cluster is in security mode), the Broker port number is the value of sasl.port. The default value is 21007.
  
  If Kerberos authentication is disabled for the cluster (the cluster is in normal mode), the broker port number is the value of port. The default value is 9092. If the port number is set to 9092, set allow.everyone.if.no.acl.found to true. The procedure is as follows:
  Log in to FusionInsight Manager and choose Cluster > Services > Kafka. Click Configurations then All Configurations. On the page that is displayed, search for allow.everyone.if.no.acl.found, set it to true, and click Save.
  
  System domain name: You can log in to FusionInsight Manager, choose System > Permission > Domain and Mutual Trust, and check the value of Local Domain.
  
  IP address of the ZooKeeper quorumpeer instance
  To obtain IP addresses of all ZooKeeper quorumpeer instances, log in to FusionInsight Manager and choose Cluster > Services > ZooKeeper. On the displayed page, click Instance and view the IP addresses of all the hosts where the quorumpeer instances locate.
  
  Port number of the ZooKeeper client
  Log in to FusionInsight Manager and choose Cluster > Service > ZooKeeper. On the displayed page, click Configurations and check the value of clientPort.
  
  HBase authentication
  Log in to FusionInsight Manager, choose Cluster > Services > HBase, click Configuration and then All Configurations, search for hbase.rpc.protection, and check the HBase authentication mode. If the authentication mode is integrity or privacy, add the following parameters:
  
  'properties.hbase.rpc.protection' = 'HBase authentication mode'
  
  'properties.zookeeper.znode.parent' = '/hbase'
  
  'properties.hbase.security.authorization' = 'true'
  
  'properties.hbase.security.authentication' = 'kerberos'
3. On the job management page, check whether the job status is Running.
Execute the following script to write data to Kafka. For details, see Managing Messages in Kafka Topics.

sh kafka-console-producer.sh --broker-list IP address of the node where Kafka Broker instance is deployed:Kafka port --topic Topic name --producer.config Client directory/Kafka/kafka/config/producer.properties

For example, if the topic name is ksource1, the script is sh kafka-console-producer.sh --broker-list IP address of the node where the Kafka Broker instance is located:Kafka port --topic ksource1 --producer.config /opt/client/Kafka/kafka/config/producer.properties
Enter the message content and press Enter to send the message.
```
{"user_id": "3","item_id":"333333"}
{"user_id": "4","item_id":"44444444"}
```
Log in to the HBase client and view the table data.

hbase shell

scan 'dim_province'

Scenario 2: HBase Functioning as a Dimension Table

Create a table on the HBase client and write data into the table.
Log in to the HBase client, use create 'hbase_dim_table',"f1" to create the hbase_dim_table table, and write data into the table. For details, see Using the HBase Client.
```
put 'hbase_dim_table','1','f1:address','city1'
put 'hbase_dim_table','2','f1:address','city2'
put 'hbase_dim_table','3','f1:address','city3'
```

Copy the HBase configuration file to the node where FlinkServer is deployed.
1. Log in to the node where the client is installed as the client installation user and copy all configuration files in the /opt/client/HBase/hbase/conf/ directory of HBase to an empty directory of all nodes where FlinkServer is deployed, for example, /tmp1/client/HBase/hbase/conf/.
2. Change the owner of the configuration file directory and its upper-layer directory on the FlinkServer node to omm.
  chown omm: /tmp1/client/HBase/ -R
  NOTE:
  FlinkServer nodes:
  Log in to Manager, choose Cluster > Services > Flink > Instance, and check the Service IP Address of FlinkServer.
  
  If the node where a FlinkServer instance is located is the node where the HBase client is installed, skip this step on this node.
Add the local path for FlinkServer to access the HBase cluster.

Log in to Manager and choose Cluster > Services > Flink. Click Configurations then All Configurations, search for the HBASE_CONF_DIR parameter, and enter the FlinkServer directory (for example, /tmp1/client/HBase/hbase/conf/) to which the HBase configuration files are copied from Value. After the parameters are configured, click Save. After confirming the modification, click OK.
NOTE:

If the node where a FlinkServer instance is deployed is the node where the HBase client is installed, enter the /opt/client/HBase/hbase/conf/ directory of HBase in Value of the HBASE_CONF_DIR parameter.

Restart the affected FlinkServer instance.

Click Instance, select all FlinkServer instances, choose More > Restart Instance, enter the password, and click OK to restart the instances.

Access FlinkServer and create a Flink SQL job.

Log in to Manager and choose Cluster > Services > Flink. In the Basic Information area, click the link on the right of Flink WebUI to access the Flink web UI.

Create a Flink SQL job and set Task Type to Stream job. For details, see Creating a FlinkServer Job. On the job development page, configure the job parameters as follows and start the job.

In Basic Parameter, select Enable CheckPoint, set Time Interval(ms) to 60000, and retain the default value for Mode.

If the cluster is in security mode and the HBase authentication setting is hbase.rpc.protection=authentication, create a Flink SQL job by referring to the following example:

CREATE TABLE KafkaSource (
  `user_id` STRING,
  `user_name` STRING,
  proctime as proctime()
) WITH (
  'connector' = 'kafka',
  'topic' = 'user_source',
  'properties.bootstrap.servers' = 'IP address of the Kafka broker instance:Kafka port',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'latest-offset',
  'value.format' = 'csv',
  'properties.sasl.kerberos.service.name' = 'kafka',--This parameter is not required for clusters in normal mode.
'properties.security.protocol' = 'SASL_PLAINTEXT',--This parameter is not required for clusters in normal mode.
'properties.kerberos.domain.name' = 'hadoop.System domain name'--This parameter is not required for clusters in normal mode.
);
CREATE TABLE KafkaSink (
  -- Kafka functions as a sink table.
  `user_id` VARCHAR,
  `user_name` VARCHAR,
  `address` VARCHAR
) WITH (
  'connector' = 'kafka',
  'topic' = 'user_sink',
  'properties.bootstrap.servers' = 'IP address of the Kafka broker instance:Kafka port',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'latest-offset',
  'value.format' = 'csv',
  'properties.sasl.kerberos.service.name' = 'kafka',--This parameter is not required for clusters in normal mode.
'properties.security.protocol' = 'SASL_PLAINTEXT',--This parameter is not required for clusters in normal mode.
'properties.kerberos.domain.name' = 'hadoop.System domain name'--This parameter is not required for clusters in normal mode.
);
CREATE TABLE hbaseTable (
  -- HBase dimension table
  user_id STRING,
  f1 ROW < address STRING >,
  PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
  'connector' = 'hbase-2.2',
  'table-name' = 'hbase_dim_table',
  'zookeeper.quorum' = 'IP address of the ZooKeeper quorumpeer instance 1:ZooKeeper port number,'IP address of the ZooKeeper quorumpeer instance 2:ZooKeeper port number'
);
INSERT INTO
  KafkaSink
SELECT
  t.user_id,
  t.user_name,
  d.address
FROM
  KafkaSource as t
  JOIN hbaseTable FOR SYSTEM_TIME AS OF t.proctime as d ON t.user_id = d.user_id;

On the job management page, check whether the job status is Running.

Execute the following script to write data to Kafka. For details, see Managing Messages in Kafka Topics.

sh kafka-console-producer.sh --broker-list IP address of the node where Kafka Broker instance is deployed:Kafka port --topic Topic name --producer.config Client directory/Kafka/kafka/config/producer.properties
Enter the message content and press Enter to send the message.
```
1,name1
2,name2
3,name3
```
Run the following commands to check whether data is written from the Kafka topic to the sink table:

sh kafka-console-consumer.sh --topic Topic name --bootstrap-server Service IP address of the Kafka broker instance:Kafka port number --consumer.config Client directory/Kafka/kafka/config/consumer.properties

The result is as follows:
```
1,name1,city1
2,name2,city2
3,name3,city3
```

Submitting a Job Using the Application

If the Flink run mode is used, you are advised to use the export HBASE_CONF_DIR= HBase configuration directory, for example, export HBASE_CONF_DIR=/opt/hbaseconf.

If the Flink run-application mode is used, you can use either of the following methods to submit jobs:

(Recommended) Add the following configurations to a table creation statement.

**Table 1** Related configurations
Parameter	Description
'properties.hbase.rpc.protection' = 'authentication'	This parameter must be consistent with that on the HBase server.
'properties.zookeeper.znode.parent' = '/hbase'	If there are multiple services, hbase1 and hbase2 coexist. You must clarify the cluster to be accessed.
'properties.hbase.security.authorization' = 'true'	Authentication is enabled.
'properties.hbase.security.authentication' = 'kerberos'	Kerberos authentication is enabled.

Example:

CREATE TABLE hsink1 (
     rowkey STRING,
     f1 ROW < q1 STRING >,
     PRIMARY KEY (rowkey) NOT ENFORCED
    ) WITH (
      'connector' = 'hbase-2.2',
      'table-name' = 'cc',
      'zookeeper.quorum' = 'x.x.x.x:clientPort',
      'properties.hbase.rpc.protection' = 'authentication',
      'properties.zookeeper.znode.parent' = '/hbase',
      'properties.hbase.security.authorization' = 'true',
      'properties.hbase.security.authentication' = 'kerberos'
   );