Using a Flink Job of DLI to Synchronize Kafka Data to a GaussDB(DWS) Cluster in Real Time

Scenario Description

Assume that the sample data of the data source Kafka is a user information table, as shown in Table 1, which contains the id, name, and age fields. The id field is unique and fixed, which is shared by multiple service systems. Generally, the id field does not need to be modified. Only the name and age fields need to be modified.

Use Kafka to generate the following three groups of data and use DLI Flink jobs to synchronize the data to GaussDB(DWS): Change the users whose IDs are 2 and 3 to jim and tom, and use DLI Flink jobs to update data and synchronize the data to GaussDB(DWS).

Constraints

• Ensure that VPC, ECS, OBS, Kafka, DLI, and GaussDB(DWS) are in the same region, for example, China-Hong Kong.
• Ensure that Kafka, DLI, and GaussDB(DWS) can communicate with each other. In this practice, Kafka and GaussDB(DWS) are created in the same region and VPC, and the security groups of Kafka and GaussDB(DWS) allow the network segment of the DLI queues.
• To ensure that the link between DLI and GaussDB(DWS) is stable, bind the ELB service to the created GaussDB(DWS) cluster.

Preparations

• You have registered a Huawei account and enabled Huawei Cloud services.. Before using GaussDB(DWS), check the account status. The account cannot be in arrears or frozen.
• You have created a VPC and subnet. For details, see Creating a VPC.

Step 1: Creating a Kafka Instance

1. Log in to the Huawei Cloud management console and choose Middleware > Distributed Message Service (for Kafka) from the service list. The Kafka management console is displayed.
2. Click Kafka Instances on the left and click Buy Instance in the upper right corner.
3. Configure the following parameters as instructed. Retain default settings for other parameters.

   Table 2 Kafka instance parameters

   Parameter	Value
   Billing Mode	Pay-per-use
   Region	CN-Hong Kong
   AZ	AZ 1 (If not available, select another AZ.)
   Bundle	Starter
   VPC	Select a created VPC. If no VPC is available, create one.
   Security Group	Select a created security group. If no security group is available, create one.
   Other parameters	Retain the default value.

4. Click Confirm, confirm the information, and click Submit. Wait until the creation is successful.
5. In the Kafka instance list, click the name of the created Kafka instance. The Basic Information page is displayed.
6. Choose Topics on the left and click Create Topic.

   Set Topic Name to topic-demo and retain the default values for other parameters.

   Figure 2 Creating a topic
   

7. Click OK. In the topic list, you can see that topic-demo is successfully created.
8. Choose Consumer Groups on the left and click Create Consumer Group.
9. Enter kafka01 for Consumer Group Name and click OK.

Step 2: Creating a GaussDB(DWS) Cluster and Target Table

1. Create a dedicated load balancer, set Network Type to IPv4 private network. Set Region and VPC to the same values as those of the Kafka instance. In this example, set Region to China-Hong Kong.
2. Creating a Cluster. To ensure network connectivity, the region and VPC of the GaussDB(DWS) cluster must be the same as those of the Kafka instance. In this practice, the region and VPC are China-Hong Kong. The VPC must be the same as that created for Kafka.
3. Log in to the GaussDB(DWS) console, choose Dedicated Clusters > Clusters, locate the row that contains the target cluster, and click Login in the Operation column.
   

   This practice uses version 8.1.3.x as an example. 8.1.2 and earlier versions do not support this login mode. You can use Data Studio to connect to a cluster. For details, see Using Data Studio to Connect to a Cluster.

4. After the login is successful, the SQL editor is displayed.
5. Copy the following SQL statement. In the SQL window, click Execute SQL to create the target table user_dws.

   1 2 3 4 5 6

   CREATE TABLE user_dws ( id int, name varchar(50), age int, PRIMARY KEY (id) );

Step 3: Creating a DLI Queue

1. Log in to the Huawei Cloud management console and choose Analytics > Data Lake Insight from the service list. The DLI management console is displayed.
2. In the navigation pane on the left, choose Resources > Resource Pool.
3. Click Buy Resource Pool in the upper right corner, configure the following parameters as instructed, and retain default settings for other parameters.

   Table 3 Parameters

   Parameter	Value
   Billing Mode	Pay-per-use
   Region	CN-Hong Kong
   Name	dli_dws
   Specifications	Standard
   CIDR Block	172.16.0.0/18. It must be in a different network segment from Kafka and GaussDB(DWS). For example, if Kafka and GaussDB(DWS) are in the 192.168.x.x network segment, select 172.16.x.x for DLI.

4. Click Buy and click Submit.

   After the resource pool is created, go to the next step.

5. On the elastic resource pool page, locate the row that contains the created resource pool, click Add Queue in the Operation column, and set the following parameters. Retain the default values for other parameters that are not described in the table.

   Table 4 Adding a queue

   Parameter	Value
   Name	dli_dws
   Type	General purpose queue

6. Click Next and click OK. The queue is created.

Step 4: Creating an Enhanced Datasource Connection for Kafka and GaussDB(DWS)

1. In the security group of Kafka, allow the network segment where the DLI queue is located.
   1. Return to the Kafka console and click the Kafka instance name to go to the Basic Information page. View the value of Instance Address (Private Network) in connection information and record the address for future use.
      Figure 3 Kafka private network address
      
   2. Click the security group name.
      Figure 4 Kafka security group
      
   3. Choose Inbound Rules > Add Rule, as shown in the following figure. Add the network segment of the DLI queue. In this example, the network segment is 172.16.0.0/18. Ensure that the network segment is the same as that entered during Step 3: Creating a DLI Queue.
      Figure 5 Adding rules to the Kafka security group
      
   4. Click OK.

2. Return to the DLI management console, click Datasource Connections on the left, select Enhanced, and click Create.
3. Set the following parameters. Retain the default values for other parameters that are not described in the table.

   Table 5 Connection from DLI to Kafka

   Parameter	Value
   Connection Name	dli_kafka
   Resource Pool	Select the created DLI queue dli_dws.
   VPC	Select the VPC of Kafka.
   Subnet	Select the subnet where Kafka is located.
   Other parameters	Retain the default value.

   Figure 6 Creating an enhanced connection
   

4. Click OK. Wait until the Kafka connection is successfully created.
   

   If you do not select a resource pool when creating an enhanced datasource connection, you can manually bind one afterwards.

   1. Locate the row that contains the target datasource connection and choose More > Bind Resource Pool in the Operation column.
   2. Select a resource pool and click OK.

      

5. Choose Resources > Queue Management on the left, and choose More > Test Address Connectivity on the right of dli_dws.
6. In the address box, enter the private IP address and port number of the Kafka instance obtained in 1.a. (There are three Kafka addresses. Enter only one of them.)
   Figure 7 Testing Kafka connectivity
   

7. Click Test to verify that DLI is successfully connected to Kafka.
8. Log in to the GaussDB(DWS) console, choose Dedicated Clusters > Clusters on the left, and click the cluster name to go to the details page.
9. Record the private network domain name, port number, and ELB address of the GaussDB(DWS) cluster for future use.
   Figure 8 Private domain name and ELB address
   

10. Click the security group name.
    Figure 9 GaussDB(DWS) security group
    

11. Choose Inbound Rules > Add Rule, as shown in the following figure. Add the network segment of the DLI queue. In this example, the network segment is 172.16.0.0/18. Ensure that the network segment is the same as that entered during Step 3: Creating a DLI Queue.
    Figure 10 Adding a rule to the GaussDB(DWS) security group
    

12. Click OK.
13. Switch to the DLI console, choose Resources > Queue Management on the left, and click More > Test Address Connectivity on the right of dli_dws.
14. In the address box, enter the ELB address and port number of the GaussDB(DWS) cluster obtained in 9.
    Figure 11 Testing GaussDB(DWS) connectivity
    

15. Click Test to verify that DLI is successfully connected to GaussDB(DWS).

Step 5: Preparing the dws-connector-flink Tool for Interconnecting GaussDB(DWS) with Flink

dws-connector-flink is a tool for interconnecting with Flink based on GaussDB(DWS) JDBC APIs. During DLI job configuration, this tool and its dependencies are stored in the Flink class loading directory to improve the capability of importing Flink jobs to GaussDB(DWS).

1. Go to https://mvnrepository.com/artifact/com.huaweicloud.dws using a browser.
2. In the software list, select the latest version of GaussDB(DWS) Connectors Flink. In this practice, select DWS Connector Flink 2 12 1 12.

   

3. Click the 1.0.4 branch. (Click the newest branch in actual scenarios).

   

4. Click View ALL.

   

5. Click dws-connector-flink_2.12_1.12-1.0.4-jar-with-dependencies.jar to download it to the local host.

   

6. Create an OBS bucket. In this practice, set the bucket name to obs-flink-dws and upload the file to the OBS bucket. Ensure that the bucket is in the same region as DLI, which in this practice is China-Hong Kong.
   Figure 12 Uploading the JAR package to the OBS bucket
   

Step 6: Creating and Editing a DLI Flink Job

1. Update the DLI agency permissions.
   1. Return to the DLI console and choose Global Configuration > Service Authorization on the left.
   2. Select DLI Datasource Connections Agency Access and DLI Notification Agency Access.
   3. Click Update. Click OK.
   Figure 13 Updating DLI agency permissions
   

2. Create an OBS agency policy.
   1. Hover over the account name in the upper right corner of the console, and click Identity and Access Management.
   2. In the navigation pane on the left, choose Agencies and then click Create Agency in the upper right corner.
      • Agency Name: dli_ac_obs
      • Agency Type: Cloud service
      • Cloud Service: Data Lake Insight (DLI)
      • Validity Period: Unlimited
      Figure 14 Creating an OBS agency
      

   3. Click Next.
   4. Click the agency name. On the displayed page, click the Permissions tab. Click Authorize. On the displayed page, click Create Policy.
   5. Configure policy information. Enter a policy name, for example, dli_ac_obs, and select JSON.
   6. In the Policy Content area, paste a custom policy. Replace OBS bucket name with the actual bucket name.

       {
          "Version": "1.1",
          "Statement": [
              {
                  "Effect": "Allow",
                  "Action": [
                      "obs:object:GetObject",
                      "obs:object:DeleteObjectVersion",
                      "obs:bucket:GetBucketLocation",
                      "obs:bucket:GetLifecycleConfiguration",
                      "obs:object:AbortMultipartUpload",
                      "obs:object:DeleteObject",
                      "obs:bucket:GetBucketLogging",
                      "obs:bucket:HeadBucket",
                      "obs:object:PutObject",
                      "obs:object:GetObjectVersionAcl",
                      "obs:bucket:GetBucketAcl",
                      "obs:bucket:GetBucketVersioning",
                      "obs:bucket:GetBucketStoragePolicy",
                      "obs:bucket:ListBucketMultipartUploads",
                      "obs:object:ListMultipartUploadParts",
                      "obs:bucket:ListBucketVersions",
                      "obs:bucket:ListBucket",
                      "obs:object:GetObjectVersion",
                      "obs:object:GetObjectAcl",
                      "obs:bucket:GetBucketPolicy",
                      "obs:bucket:GetBucketStorage"
                  ],
                  "Resource": [
                      "OBS:*:*:object:*",
                      "OBS:*:*:bucket:OBS bucket name"
                  ]
              },
              {
                  "Effect": "Allow",
                  "Action": [
                      "obs:bucket:ListAllMyBuckets"
                  ]
              }
          ]
      }

   7. Click Next.
   8. Select the created custom policy.
   9. Click Next. Select All resources.
   10. Click OK.

       It takes 15 to 30 minutes for the authorization to take effect.

3. Return to the DLI management console, choose Job Management > Flink Jobs on the left, and click Create Job in the upper right corner.
4. Set Type to Flink OpenSource SQL and Name to kafka-dws.
   Figure 15 Creating a job
   

5. Click OK. The page for editing the job is displayed.
6. Set the following parameters on the right of the page. Retain the default values for other parameters that are not described in the table.

   Table 6 Flink job parameters

   Parameter	Value
   Queue	dli_dws
   Flink Version	1.15
   UDF Jar	Select the JAR file in the OBS bucket created in Step 5: Preparing the dws-connector-flink Tool for Interconnecting GaussDB(DWS) with Flink.
   Agency	Select the agency created in 2.
   OBS Bucket	Select the bucket created in Step 5: Preparing the dws-connector-flink Tool for Interconnecting GaussDB(DWS) with Flink.
   Enable Checkpointing	Check the box.
   Other parameters	Retain the default value.

7. Copy the following SQL code to the SQL code window on the left.
   Obtain the private IP address and port number of the Kafka instance from 1.a, and obtain the private domain name from 9.

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31

   CREATE TABLE user_kafka ( id string, name string, age int ) WITH ( 'connector' = 'kafka', 'topic' = 'topic-demo', 'properties.bootstrap.servers' ='Private IP address and port number of the Kafka instance', 'properties.group.id' = 'kafka01', 'scan.startup.mode' = 'latest-offset', "format" = "json" ); CREATE TABLE user_dws ( id string, name string, age int, PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'dws', 'url'='jdbc:postgresql://GaussDB(DWS) private network domain name:8000/gaussdb', 'tableName' = 'public.user_dws', 'username' = 'dbadmin', 'password' ='Password of database user dbdamin' 'writeMode' = 'auto', 'autoFlushBatchSize'='50000', 'autoFlushMaxInterval'='5s', 'key-by-before-sink'='true' ); INSERT INTO user_dws select * from user_kafka;

8. Click Check Semantics and wait until the verification is successful.

   If the verification fails, check whether the SQL input has syntax errors.

   Figure 16 SQL statement of a job
   

9. Click Save.
10. Return to the DLI console home page and choose Job Management > Flink Jobs on the left.
11. Click Start on the right of the job name kafka-dws and click Start Now.

    Wait for about 1 minute and refresh the page. If the status is Running, the job is successfully executed.

    Figure 17 Job execution status
    

Step 7: Creating and Modifying Messages on the Kafka Client

1. Create an ECS by referring to the ECS document. Ensure that the region and VPC of the ECS are the same as those of Kafka.
2. Install JDK.
   1. Log in to the ECS, go to the /usr/local directory, and download the JDK package.

      1 2	cd /usr/local wget https://download.oracle.com/java/17/latest/jdk-17_linux-x64_bin.tar.gz

   2. Decompress the downloaded JDK package.

      1	tar -zxvf jdk-17_linux-x64_bin.tar.gz

   3. Run the following command to open the /etc/profile file:

      1	vim /etc/profile

   4. Press i to enter editing mode and add the following content to the end of the /etc/profile file:

      1 2 3 4 5

      export JAVA_HOME=/usr/local/jdk-17.0.7 #JDK installation directory export JRE_HOME=${JAVA_HOME}/jre export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib:${JAVA_HOME}/test:${JAVA_HOME}/lib/gsjdbc4.jar:${JAVA_HOME}/lib/dt.jar:${JAVA_HOME}/lib/tools.jar:$CLASSPATH export JAVA_PATH=${JAVA_HOME}/bin:${JRE_HOME}/bin export PATH=$PATH:${JAVA_PATH}

      

   5. Press Esc and enter :wq! to save the settings and exit.
   6. Run the following command for the environment variables to take effect:

      1	source /etc/profile

   7. Run the following command. If the following information is displayed, the JDK is successfully installed:

      1	java -version

      

3. Install the Kafka client.
   1. Go to the /opt directory and run the following command to obtain the Kafka client software package.

      1 2	cd /opt wget https://archive.apache.org/dist/kafka/2.7.2/kafka_2.12-2.7.2.tgz

   2. Decompress the downloaded software package.

      1	tar -zxf kafka_2.12-2.7.2.tgz

   3. Go to the Kafka client directory.

      1	cd /opt/kafka_2.12-2.7.2/bin

4. Run the following command to connect to Kafka: {Connection address} indicates the internal network connection address of Kafka. For details about how to obtain the address, see 1.a. topic indicates the name of the Kafka topic created in 6.

   1	./kafka-console-producer.sh --broker-list {connection address} --topic {Topic name}

   The following is an example:

   ./kafka-console-producer.sh --broker-list 192.168.0.136:9092,192.168.0.214:9092,192.168.0.217:9092 --topic topic-demo

   

   If > is displayed and no other error message is displayed, the connection is successful.

5. In the window of the connected Kafka client, copy the following content (one line at a time) based on the data planned in the Scenario Description and press Enter to produce messages:

   1 2 3

   {"id":"1","name":"lily","age":"16"} {"id":"2","name":"lucy","age":"17"} {"id":"3","name":"lilei","age":"15"}

   

6. Return to the GaussDB(DWS) console, choose Dedicated Clusters > Clusters on the left, and click Log In on the right of the GaussDB(DWS) cluster. The SQL page is displayed.
7. Run the following SQL statement to verify that data is successfully imported to the database in real time:

   1	SELECT * FROM user_dws ORDER BY id;

   

8. Go back to the client window for connecting to Kafka on the ECS, copy the following content (one line at a time), and press Enter to produce messages.

   1 2	{"id":"2","name":"jim","age":"17"} {"id":"3","name":"tom","age":"15"}

9. Go back to the opened SQL window of GaussDB(DWS) and run the following SQL statement. It is found that the names whose IDs are 2 and 3 have been changed to jim and tom.
   The scenario description is as expected. End of this practice.

   1	SELECT * FROM user_dws ORDER BY id;