Interconnecting FlinkServer with HBase

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 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, choose Cluster > Services > Flink > Configurations > 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.
The following example shows how to create a Flink SQL job for a security cluster:
```
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',
'properties.security.protocol' = 'SASL_PLAINTEXT',
'properties.kerberos.domain.name' = 'hadoop.System domain name'
);

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;
```
- Kafka port number
  
  Value of sasl.port when Authentication Mode of the cluster is Security Mode, 21007 by default.
  
  Value of port when Authentication Mode of the cluster is Normal Mode, 9092 by default. 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. On the page that is displayed, click the Configurations tab then the All Configurations sub-tab. On the displayed page, search for allow.everyone.if.no.acl.found, set it to true, and click Save.
- 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.
- System domain name: You can log in to FusionInsight Manager, choose System > Permission > Domain and Mutual Trust, and check the value of Local Domain.
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'
   );