Interconnecting FlinkServer with GaussDB(DWS)

Scenario

FlinkServer can interconnect with GaussDB(DWS) 8.1.x or later. This section describes the DDL definitions when GaussDB(DWS) serves as source tables, sink tables, and dimension tables, as well as the WITH parameter and code examples used during table creation, and describes how to perform operations on the FlinkServer job management page.

In this example, FlinkServer and Kafka in security mode are used to interconnect with GaussDB(DWS) in security mode.

When "FlinkSQL" is displayed in the command output on the FlinkServer web UI, the password field in the SQL statement is left blank. Before you submit a job, manually enter the password.

Mappings between FlinkSQL and ClickHouse data types

Prerequisites

• Cluster where FlinkServer resides (security mode):
  • HDFS, YARN, Kafka, ZooKeeper, and Flink have been installed in the cluster.
  • The client that contains the Kafka service has been installed in a directory, for example, /opt/client.
  • You have created a user with the FlinkServer Admin Privilege (for example, flinkuser) for accessing the Flink web UI by referring to Creating a FlinkServer Role.
• Cluster of the GaussDB(DWS) to be interconnected (security mode):
  • Whitelist all service IP addresses of the nodes where FlinkServer resides.
    

    To whitelist the IP addresses, perform the following steps:

    1. Log in to the node where GaussDB(DWS) resides as user root and run the following command to switch to user omm:

       su - omm

    2. Run the following command to load environment variables:

       source ${BIGDATA_HOME}/mppdb/.mppdbgs_profile

    3. Run the following command to whitelist one IP address. To whitelist multiple IP addresses, run the command for multiple times.

       gs_guc set -Z coordinator -N all -I all -h "host all all IP address of the FlinkServer node/32 sha256"

  • An empty table for receiving data has been created, for example, customer_t1.
    

    To create a GaussDB(DWS) data table, perform the following steps:

    1. Log in to the node where GaussDB(DWS) resides as user root and run the following command to switch to user omm:

       su - omm

    2. Run the following command to load environment variables:

       source ${BIGDATA_HOME}/mppdb/.mppdbgs_profile

    3. Run the following command to connect to the database:
       gsql -d postgres -U username –p 25308-W password –r

       • postgres indicates the name of the database to be connected.
       • username and password indicate the username and password for connecting to the database. Commands carrying authentication passwords pose security risks. Disable historical command recording before running such commands to prevent information leakage.
       • 25308 indicates the port number of the Coordinator. Replace it with the actual port number. You can run the gs_om -t status --detail command to query the Coordinator data path and view the port number in the postgresql.conf file in the path.
    4. Run the following command to create a data table:

       CREATE TABLE customer_t1(c_customer_sk INTEGER, c_customer_name VARCHAR(32));

GaussDB as a Sink Table

1. Log in to Manager as user flinkuser and choose Cluster > Services > Flink. In the Basic Information area, click the link next to Flink WebUI to access the Flink web UI.
2. Create a FlinkSQL job by referring to Creating a Job. On the job development page, configure the job parameters and start the job.

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

   CREATE TABLE MyUserTable(
     c_customer_sk INTEGER,
     c_customer_name VARCHAR(32)
   ) WITH(
     'connector' = 'jdbc',
     'url' = 'jdbc:gaussdb://IP address of the GaussDB server:25308/postgres',
   'table-name' = 'customer_t1',--If table customer_t1 is created in schema base, the configuration rule is 'table-name' = 'base"."customer_t1'.
     'username' = 'username',--Username for logging in to the GaussDB(DWS) database
     'password' = 'password',--Password for logging in to the GaussDB(DWS) database
     'write.mode' = 'upsert',--When data is written in upsert mode, you can set whether to ignore null values. 
     'ignoreNullWhenUpsert' = 'false'--true indicates that null values are ignored, and false indicates that null values are not ignored and written to the database.
   );
   CREATE TABLE KafkaSource (
     c_customer_sk INTEGER,
     c_customer_name VARCHAR(32)
   ) WITH (
     'connector' = 'kafka',
     'topic' = 'customer_source',
     'properties.bootstrap.servers' = 'Service IP address of the Kafka Broker instance:Kafka port number',
     'properties.group.id' = 'testGroup',
     'scan.startup.mode' = 'latest-offset',
     'value.format' = 'csv',
     'properties.sasl.kerberos.service.name' = 'kafka',  --Delete this parameter if the cluster where FlinkServer resides is in non-security mode.
     'properties.security.protocol' = 'SASL_PLAINTEXT',  --Delete this parameter if the cluster where FlinkServer resides is in non-security mode.
     'properties.kerberos.domain.name' = 'hadoop.System domain name'  --Delete this parameter if the cluster where FlinkServer resides is in non-security mode.
   );
   Insert into
     MyUserTable
   select
     *
   from
     KafkaSource;

   

   • 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.
   • properties.group.id indicates the Kafka user group ID. This parameter is mandatory when Kafka functions as the source.
   • properties.kerberos.domain.name: Set it to hadoop.System domain name. You can log in to FusionInsight Manager and choose System > Permission > Domain and Mutual Trust to view the actual domain name of the cluster.

3. On the job management page, check whether the job is in the Running status.
4. Execute the following commands to view the topic and write data to Kafka. For details, see Managing Messages in Kafka Topics.

   ./kafka-topics.sh --list --zookeeper Service IP address of the ZooKeeper quorumpeer instance:Port number of the ZooKeeper client/kafka

   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

   In this example, the topic name is customer_source.

   sh kafka-console-producer.sh --broker-list IP address of the node where the Kafka instance is located:Kafka port number --topic customer_source --producer.config /opt/client/Kafka/kafka/config/producer.properties

   Enter the message content.

   3,zhangsan
   4,wangwu
   8,zhaosi

   Press Enter to send the message.

   

   • Service IP address of the ZooKeeper quorumpeer instance:

     Log in to FusionInsight Manager and choose Cluster > Services > ZooKeeper. On the page that is displayed, click the Instance tab and view the service IP addresses of all nodes where the quorumpeer instances reside.

   • Port number of the ZooKeeper client:

     Log in to FusionInsight Manager and choose Cluster > Services > ZooKeeper. On the page that is displayed, click the Configurations tab and check the value of clientPort.

5. Log in to the GaussDB client and run the following command to check whether data has been written to the sink table:

   Select * from customer_t1;

   

GaussDB as a Source Table

1. Log in to Manager as user flinkuser and choose Cluster > Services > Flink. In the Basic Information area, click the link next to Flink WebUI to access the Flink web UI.
2. Create a FlinkSQL job by referring to Creating a Job. On the job development page, configure the job parameters and start the job.

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

   CREATE TABLE MyUserTable(
     --GuassDB functions as a source table.
     c_customer_sk INTEGER,
     c_customer_name VARCHAR(32)
   ) WITH(
     'connector' = 'jdbc',
     'url' = 'jdbc:gaussdb://IP address of the GaussDB server:25308/postgres ',
     'table-name' = 'customer_t1',
     'username' = 'username ',
     'password' = 'password '
   );
   CREATE TABLE KafkaSink (
     -- Kafka functions as a sink table.
     c_customer_sk INTEGER,
     c_customer_name VARCHAR(32)
   ) WITH (
     'connector' = 'kafka',
     'topic' = 'customer_sink',
     'properties.bootstrap.servers' = 'Service IP address of the Kafka Broker instance:Kafka port number',
     'properties.group.id' = 'testGroup',
     'scan.startup.mode' = 'latest-offset',
     'value.format' = 'csv',
     'properties.sasl.kerberos.service.name' = 'kafka',  --Delete this parameter if the cluster where FlinkServer resides is in non-security mode.
     'properties.security.protocol' = 'SASL_PLAINTEXT',  --Delete this parameter if the cluster where FlinkServer resides is in non-security mode.
     'properties.kerberos.domain.name' = 'hadoop.System domain name'  --Delete this parameter if the cluster where FlinkServer resides is in non-security mode.
   );
   Insert into
     KafkaSink
   select
     *
   from
     MyUserTable;

   

   • 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.
   • properties.group.id indicates the Kafka user group ID. This parameter is mandatory when Kafka functions as the source.
   • properties.kerberos.domain.name: Set it to hadoop.System domain name. You can log in to FusionInsight Manager and choose System > Permission > Domain and Mutual Trust to view the actual domain name of the cluster.

3. On the job management page, check whether the job is in the Running status.
4. Run the following command to check whether data is received in the sink table, that is, check whether data is properly written to the Kafka topic. For details, see Managing Messages in Kafka Topics.

   sh kafka-console-consumer.sh --topic customer_sink --bootstrap-server IP address of the node where the Kafka instance is located:Kafka port number --consumer.config /opt/client/Kafka/kafka/config/ consumer.properties

   

GaussDB as a Dimension Table

kafkaSource is used as the fact table, customer_t2 is used as the dimension table, and the result is written to kafkaSink.

1. Create dimension table customer_t2 on the GaussDB client by referring to Creating an Empty Data Table in the GaussDB(DWS) Cluster in Security Mode. An example of the table creation statement is as follows:

   CREATE TABLE customer_t2(
   c_customer_sk INTEGER PRIMARY KEY,
   c_customer_age INTEGER,
   c_customer_address VARCHAR(32)
   )DISTRIBUTE BY HASH(c_customer_sk);
    
   INSERT INTO customer_t2 VALUES(1,18,'city a');
   INSERT INTO customer_t2 VALUES(2,14,'city b');
   INSERT INTO customer_t2 VALUES(3,16,'city c');
   INSERT INTO customer_t2 VALUES(4,24,'city d');
   INSERT INTO customer_t2 VALUES(5,32,'city e');
   INSERT INTO customer_t2 VALUES(6,27,'city f');
   INSERT INTO customer_t2 VALUES(7,41,'city a');
   INSERT INTO customer_t2 VALUES(8,35,'city h');
   INSERT INTO customer_t2 VALUES(9,16,'city j');

2. Log in to Manager as user flinkuser and choose Cluster > Services > Flink. In the Basic Information area, click the link next to Flink WebUI to access the Flink web UI.
3. Create a FlinkSQL job by referring to Creating a Job. On the job development page, configure the job parameters and start the job.

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

   CREATE TABLE KafkaSource (
     -- Kafka as a source table
     c_customer_sk INTEGER,
     c_customer_name VARCHAR(32),
     proctime as proctime()
   ) WITH (
     'connector' = 'kafka',
     'topic' = 'customer_source',
     'properties.bootstrap.servers' = 'Service IP address of the Kafka Broker instance:Kafka port number',
     'properties.group.id' = 'testGroup',
     'scan.startup.mode' = 'latest-offset',
     'value.format' = 'csv',
     'properties.sasl.kerberos.service.name' = 'kafka',  --Delete this parameter if the cluster where FlinkServer resides is in non-security mode.
     'properties.security.protocol' = 'SASL_PLAINTEXT',  --Delete this parameter if the cluster where FlinkServer resides is in non-security mode.
     'properties.kerberos.domain.name' = 'hadoop.System domain name'  --Delete this parameter if the cluster where FlinkServer resides is in non-security mode.
   );
   CREATE TABLE KafkaSink (
     -- Kafka as a sink table
     c_customer_sk INTEGER,
     c_customer_name VARCHAR(32),
     c_customer_age INTEGER,
     c_customer_address VARCHAR(32)
   ) WITH (
     'connector' = 'kafka',
     'topic' = 'customer_sink',
     'properties.bootstrap.servers' = 'Service IP address of the Kafka Broker instance:Kafka port number',
     'properties.group.id' = 'testGroup',
     'scan.startup.mode' = 'latest-offset',
     'value.format' = 'csv',
     'properties.sasl.kerberos.service.name' = 'kafka',  --Delete this parameter if the cluster where FlinkServer resides is in non-security mode.
     'properties.security.protocol' = 'SASL_PLAINTEXT',  --Delete this parameter if the cluster where FlinkServer resides is in non-security mode.
     'properties.kerberos.domain.name' = 'hadoop.System domain name'  --Delete this parameter if the cluster where FlinkServer resides is in non-security mode.
   );
   CREATE TABLE MyUserTable (
     -- GaussDB as a dimension table
     c_customer_sk INTEGER PRIMARY KEY,
     c_customer_age INTEGER,
     c_customer_address VARCHAR(32)
   ) WITH (
     'connector' = 'jdbc',
     'url' = 'jdbc:gaussdb://IP address of the GaussDB server:25308/postgres ',
     'table-name' = 'customer_t2',
     'username' = 'username ',
     'password' = 'password '
   );
   INSERT INTO
     KafkaSink
   SELECT
     t.c_customer_sk,
     t.c_customer_name,
     d.c_customer_age,
     d.c_customer_address
   FROM
     KafkaSource as t
     JOIN MyUserTable FOR SYSTEM_TIME AS OF t.proctime as d ON t.c_customer_sk = d.c_customer_sk;

   

   • 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.
   • properties.group.id indicates the Kafka user group ID. This parameter is mandatory when Kafka functions as the source.
   • properties.kerberos.domain.name: Set it to hadoop.System domain name. You can log in to FusionInsight Manager and choose System > Permission > Domain and Mutual Trust to view the actual domain name of the cluster.

4. Run the following command to check whether data is received in the sink table, that is, check whether data is properly written to the Kafka topic after 5 is performed. For details, see Managing Messages in Kafka Topics.

   sh kafka-console-consumer.sh --topic customer_sink --bootstrap-server IP address of the node where the Kafka instance is located:Kafka port number --consumer.config /opt/client/Kafka/kafka/config/ consumer.properties

5. View the topic and write data to the Kafka topic by referring to Managing Messages in Kafka Topics. After the data is written, view the execution result in the window in 4.

   ./kafka-topics.sh --list --zookeeper Service IP address of the ZooKeeper quorumpeer instance:Port number of the ZooKeeper client/kafka

   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

   

   • Service IP address of the ZooKeeper quorumpeer instance:

     Log in to FusionInsight Manager and choose Cluster > Services > ZooKeeper. On the page that is displayed, click the Instance tab and view the service IP addresses of all nodes where the quorumpeer instances reside.

   • Port number of the ZooKeeper client:

     Log in to FusionInsight Manager and choose Cluster > Services > ZooKeeper. On the page that is displayed, click the Configurations tab and check the value of clientPort.

   In this example, the topic name is customer_source.

   sh kafka-console-producer.sh --broker-list IP address of the node where the Kafka instance is located:Kafka port number --topic customer_source --producer.config /opt/client/Kafka/kafka/config/producer.properties

   Enter the message content.

   3,zhangsan
   5,zhaosi
   1,xiaoming
   2,liuyang
   7,liubei
   10,guanyu
   20,zhaoyun

   Press Enter to send the message. The output in the kafka-console-consumer window in 4 is as follows:

   3,zhangsan,16,city c
   5,zhaosi,32,city e
   1,xiaoming,18,city a
   2,liuyang,14,city b
   7,liubei,41,city a