Interconnecting FlinkServer with ClickHouse

Scenario

Flink interconnects with the ClickHouseBalancer instance of ClickHouse to read and write data, preventing ClickHouse traffic distribution problems.

When "FlinkSQL" is displayed in the command output on the FlinkServer web UI in MRS 3.2.0 or later clusters, the password field in the SQL statement is left blank to meet security requirements. Before you submit a job, manually enter the password.

Prerequisites

• Services such as ClickHouse, HDFS, YARN, Flink, and Kafka have been installed in the cluster.
• The client has been installed in a directory, for example, /opt/client.

Mapping Between Flink SQL and ClickHouse Data Types

Procedure

1. Log in to the node where the client is installed as user root.
2. Run the following command to go to the client installation directory:

   cd /opt/client

3. Run the following command to configure environment variables:

   source bigdata_env

4. If Kerberos authentication is enabled for the current cluster, run the following command to authenticate the current user. The current user must have the permission to create ClickHouse tables. If Kerberos authentication is disabled for the current cluster, skip this step:

   kinit Component service user

   Example: kinit clickhouseuser

5. Connect to the ClickHouse client. For details, see Using ClickHouse from Scratch. There can be security risks if a command contains the authentication password. You are advised to disable the command recording function (history) before running the command.
   • Normal mode:

     clickhouse client --host IP address of the ClickHouse instance --user Username --password 'Password' --port ClickHouse port number --multiline

   • Security mode:

     clickhouse client --host IP address of the ClickHouse instance --user Username --password 'Password'--port ClickHouse port number --secure --multiline

6. Run the following statements to create a replication table and a distributed table.
   1. Create a replication table default.test1.

      CREATE TABLE default.test1 on cluster default_cluster
      (
      `pid` Int8,
      `uid` UInt8,
      `Int_16` Int16,
      `Int_32` Int32,
      `Int_64` Int64,
      `String_x` String,
      `String_y` String,
      `float_32` Float32,
      `float_64` Float64,
      `Decimal_x` Decimal32(2),
      `Date_x` Date,
      `DateTime_x` DateTime
      )
      ENGINE = ReplicatedReplacingMergeTree('/clickhouse/tables/{shard}/test1','{replica}')
      PARTITION BY pid
      ORDER BY (pid, DateTime_x);

   2. Create a distributed table test1_all.

      CREATE TABLE test1_all ON CLUSTER default_cluster
      (
      `pid` Int8,
      `uid` UInt8,
      `Int_16` Int16,
      `Int_32` Int32,
      `Int_64` Int64,
      `String_x` String,
      `String_y` String,
      `float_32` Float32,
      `float_64` Float64,
      `Decimal_x` Decimal32(2),
      `Date_x` Date,
      `DateTime_x` DateTime
      )
      ENGINE = Distributed(default_cluster, default, test1, rand());

7. 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.
8. 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 current MRS cluster is in security mode, perform the following operations:

     create table kafkasource(
     `pid` TINYINT,
     `uid` BOOLEAN,
     `Int_16` SMALLINT,
     `Int_32` INTEGER,
     `Int_64` BIGINT,
     `String_x` CHAR,
     `String_y` VARCHAR(10),
     `float_32` FLOAT,
     `float_64` DOUBLE,
     `Decimal_x` DECIMAL(9,2),
     `Date_x` DATE,
     `DateTime_x` TIMESTAMP
     ) with(
       'connector' = 'kafka',
       'topic' = 'input',
       'properties.bootstrap.servers' = 'IP address of the Kafka broker instance:Kafka port number',
       'properties.group.id' = 'group1',
       'scan.startup.mode' = 'earliest-offset',
       'format' = 'json',
       'properties.sasl.kerberos.service.name' = 'kafka',
       'properties.security.protocol' = 'SASL_PLAINTEXT',
       'properties.kerberos.domain.name' = 'hadoop.System domain name'
     );
     CREATE TABLE cksink (
     `pid` TINYINT,
     `uid` BOOLEAN,
     `Int_16` SMALLINT,
     `Int_32` INTEGER,
     `Int_64` BIGINT,
     `String_x` CHAR,
     `String_y` VARCHAR(10),
     `float_32` FLOAT,
     `float_64` DOUBLE,
     `Decimal_x` DECIMAL(9,2),
     `Date_x` DATE,
     `DateTime_x` TIMESTAMP
     ) WITH (
     'connector' = 'jdbc',
     'url' = 'jdbc:clickhouse://IP address 1 of the ClickHouseBalancer instance:ClickHouseBalancer port number,IP address 2 of the ClickHouseBalancer instance:ClickHouseBalancer port number/default?ssl=true&sslmode=none',
     'username' = 'ClickHouse user. For details, see the note below.',
     'password' = 'ClickHouse user password. For details, see the note below.',
     'table-name' = 'test1_all',
     'driver' = 'com.clickhouse.jdbc.ClickHouseDriver',
     'sink.buffer-flush.max-rows' = '0',
     'sink.buffer-flush.interval' = '60s'
     );
     Insert into cksink 
     select
     * 
     from 
     kafkasource;

   • If the current MRS cluster is in normal mode, perform the following operations:

     create table kafkasource(
     `pid` TINYINT,
     `uid` BOOLEAN,
     `Int_16` SMALLINT,
     `Int_32` INTEGER,
     `Int_64` BIGINT,
     `String_x` CHAR,
     `String_y` VARCHAR(10),
     `float_32` FLOAT,
     `float_64` DOUBLE,
     `Decimal_x` DECIMAL(9,2),
     `Date_x` DATE,
     `DateTime_x` TIMESTAMP
     ) with(
       'connector' = 'kafka',
       'topic' = 'kinput',
       'properties.bootstrap.servers' = 'IP address of the Kafka broker instance:Kafka port number',
       'properties.group.id' = 'kafka_test',
       'scan.startup.mode' = 'earliest-offset',
       'format' = 'json'
     );
     CREATE TABLE cksink (
     `pid` TINYINT,
     `uid` BOOLEAN,
     `Int_16` SMALLINT,
     `Int_32` INTEGER,
     `Int_64` BIGINT,
     `String_x` CHAR,
     `String_y` VARCHAR(10),
     `float_32` FLOAT,
     `float_64` DOUBLE,
     `Decimal_x` DECIMAL(9,2),
     `Date_x` DATE,
     `DateTime_x` TIMESTAMP
     ) WITH (
     'connector' = 'jdbc',
     'url' = 'jdbc:clickhouse://IP address 1 of the ClickHouseBalancer instance:ClickHouseBalancer port number,IP address 2 of the ClickHouseBalancer instance:ClickHouseBalancer port number/default',
     'table-name' = 'test1_all',
     'username' = 'ClickHouse user. For details, see the note below.',
     'password' = 'ClickHouse user password. For details, see the note below.',
     'driver' = 'com.clickhouse.jdbc.ClickHouseDriver',
     'sink.buffer-flush.max-rows' = '0',
     'sink.buffer-flush.interval' = '60s'
     );
     Insert into cksink 
     select
     * 
     from 
     kafkasource;

   

   • The user in the cksink table must have related permissions on the ClickHouse tables. For details, see ClickHouse User and Permission Management.
   • 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.

   • System domain name: You can log in to FusionInsight Manager, choose System > Permission > Domain and Mutual Trust, and check the value of Local Domain.
   • ClickHouseBalancer port number:
     • If the cluster where ClickHouse resides is in security mode, the ClickHouseBalancer port number is 21428 by default.
     • If the cluster where ClickHouse resides is in normal mode, the ClickHouseBalancer port number is 21426 by default.
   • url: You can configure multiple IP addresses for ClickHouseBalancer instances to avoid single points of failure (SPOFs) of the instances.
   • DELETE messages generated during Flink computing are filtered out when data is written to ClickHouse.
   • Parameters for batch write: Flink stores data in the memory and then flushes the data to the database table when the trigger condition is met. The configurations are as follows:

     sink.buffer-flush.max-rows: Number of rows written to ClickHouse. The default value is 100.

     sink.buffer-flush.interval: Interval for batch write. The default value is 1s.

     If either of the two conditions is met, a sink operation is triggered. That is, data will be flushed to the database table.

     • Scenario 1: sink every 60s

       'sink.buffer-flush.max-rows' = '0',

       'sink.buffer-flush.interval' = '60s'

     • Scenario 2: sink every 100 rows

       'sink.buffer-flush.max-rows' = '100',

       'sink.buffer-flush.interval' = '0s'

     • Scenario 3: no sink

       'sink.buffer-flush.max-rows' = '0',

       'sink.buffer-flush.interval' = '0s'

9. On the job management page, check whether the job status is Running.
10. 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 --topicTopic name --producer.config Client directory/Kafka/kafka/config/producer.properties

    For example, if the topic name is kinput, 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 kinput --producer.config /opt/client/Kafka/kafka/config/producer.properties

    Enter the message content.

    {"pid": "3","uid":false,"Int_16": "6533","Int_32": "429496294","Int_64": "1844674407370955614","String_x": "abc1","String_y": "abc1defghi","float_32": "0.1234","float_64": "95.1","Decimal_x": "0.451236414","Date_x": "2021-05-29","DateTime_x": "2021-05-21 10:05:10"}
    {"pid": "4","uid":false,"Int_16": "6533","Int_32": "429496294","Int_64": "1844674407370955614","String_x": "abc1","String_y": "abc1defghi","float_32": "0.1234","float_64": "95.1","Decimal_x": "0.4512314","Date_x": "2021-05-29","DateTime_x": "2021-05-21 10:05:10"}

    Press Enter to send the message.

11. Interconnect with ClickHouse to query the table data.

    clickhouse client --host IP address of the ClickHouse instance --user Username --password 'Password'--port ClickHouse port number --secure --multiline

    Run the following command to check whether data is written to a specified ClickHouse table, for example, test1_all.

    select * from test1_all;