Creating a FlinkServer Job to Interconnect with DWS

Scenarios

Data Warehouse Service (DWS) is an online data analysis and processing database. This section explains how to configure FlinkServer to interconnect with DWS via the job management page. It uses FlinkServer and Kafka in security mode, with a DWS table as the sink table, and includes examples of WITH clause parameters and corresponding code samples for table creation.

The password field in Flink SQL statements is left blank when displayed in the command output on the FlinkServer web UI. Before submitting a job, you must manually re-enter the password.

Notes and Constraints

This section applies only to MRS 3.3.1 or later.

Prerequisites

Procedure

1. Create an empty table for receiving data in a DWS database, for example, dws_test.
2. Log in to FusionInsight Manager as user flinkuser and choose Cluster > Services > Flink. In the Basic Information area, click the link next to Flink Web UI to access the Flink web UI.
3. Create a Flink SQL stream job by referring to 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.

   CREATE TABLE dws_test(
     c_customer_sk INTEGER,
     c_customer_name VARCHAR(32)
   ) WITH(
     'connector' = 'dws',
     'url' = 'jdbc:postgresql://DWS connection address/gaussdb',
     'table-name' = 'dws_test',
     'username' = 'DWS username',
     'password' = 'DWS user password'
    
   
   );
   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',
     '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 is deployed is in non-security mode.
     'properties.security.protocol' = 'SASL_PLAINTEXT',  --Delete this parameter if the cluster where FlinkServer is deployed is in non-security mode.
     'properties.kerberos.domain.name' = 'hadoop.System domain name'  --Delete this parameter if the cluster where FlinkServer is deployed is in non-security mode.
   );
   Insert into
     dws_test
   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 Instances, 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. On the displayed page, click Configurations and then All Configurations. 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.
   • 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.
   • The DWS sink supports the configuration of batch parameters. When one of the conditions is met, the sink is triggered.
     • autoFlushBatchSize: size of the batch for automatically refreshing the database. The default value is 5000.
     • autoFlushMaxInterval: maximum interval for automatically refreshing the database (duration to form a batch). The default value is 5s.

4. On the job management page, check whether the job status is Running.
5. Check the topic and write data to Kafka. For details, see Managing Messages in Kafka Topics.

   ../kafka-topics.sh --list --bootstrap-server Service IP address of the Kafka Broker instance:Kafka port --command-config Client directory/Kafka/kafka/config/client.properties

   sh kafka-console-producer.sh --broker-list IP address of the node where the Kafka Broker instance is deployed:Kafka port --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 Broker instance is deployed:Kafka port --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.

6. Log in to the DWS client and run the following command to check whether the sink table received data:

   Select * from dws_test;