Interconnecting FlinkServer with HBase

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 in a directory, for example, /opt/client.
Log in to the HBase client by referring to Using an HBase Client and run the create'dim_province', "f1" command to create the dim_province table.

Procedure

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/.

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
- 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.
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 in 1 in Value.

If the node where a FlinkServer instance resides 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.
After the parameters are configured, click Save. After confirming the modification, click OK.
Click Instance, select all FlinkServer instances, choose More > Restart Instance, enter the password, and click OK to restart the instances.
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 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;
```
- 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'
- Separator of a composite RowKey:
  If data is imported using HBase BulkLoad or written by concatenating multiple fields with separators, add the following parameters when Flink reads HBase data as a source table or dimension table:
  
  'rowkey.delimiter'='...'
  
  If data is written through Flink and service applications read data through HBase, ensure that all values of the preceding parameters are strings.
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 instances reside:Kafka port number --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 instance is located:Kafka port number --topic ksource1 --producer.config /opt/client/Kafka/kafka/config/producer.properties
Enter the message content.
```
{"user_id": "3","item_id":"333333"}
{"user_id": "4","item_id":"44444444"}
```
Press Enter to send the message.
Log in to the HBase client and view the table data. For details, see Using an HBase Client.

hbase shell

scan 'dim_province'

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.

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'
   );