On this page

Show all

Help Center/ MapReduce Service/ Component Operation Guide (LTS)/ Using Flink/ Creating a FlinkServer Job/ Creating a FlinkServer Job to Interconnect with a GaussDB(DWS) Table

Creating a FlinkServer Job to Interconnect with a GaussDB(DWS) Table

Updated on 2024-12-13 GMT+08:00

View PDF

This section applies to MRS 3.2.0 and to MRS 3.3.1.

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. The following table lists the mapping between Flink SQL and GaussDB(DWS) data types.

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

NOTICE:

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

**Table 1** Mappings Between Flink SQL and GaussDB(DWS) Data Types
Flink SQL Data Type	GaussDB(DWS) Data Type
BOOLEAN	BOOLEAN
TINYINT	-
SMALLINT	SMALLINT(INT2)
SMALLINT	SMALLSERIAL(SERIAL2)
INTEGER	INTEGER
INTEGER	SERIAL
BIGINT	BIGINT
BIGINT	BIGSERIAL
FLOAT	REAL
FLOAT	FLOAT4
DOUBLE	DOUBLE
DOUBLE	FLOAT8
CHAR	CHAR(n)
VARCHAR	VARCHAR(n)
DATE	DATE
TIMESTAMP	TIMESTAMP[(p)] [WITHOUT TIME ZONE]
DECIMAL	NUMERIC[(p[,s])]
DECIMAL	DECIMAL[(p[,s])]

Prerequisites

The cluster where FlinkServer is deployed must be able to communicate with the cluster where GaussDB(DWS) is deployed. The AZ, VPC, and security group configurations of the two clusters must be the same.
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, for example, in the /opt/client directory.
- 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):
Run the following commands to connect to the database and create a table for receiving data:

gsql -d postgres -h IP -U username –p port -W password –r
NOTE:
- postgres indicates the name of the database to be connected.
- IP: indicates the IP address of the GaussDB(DWS) cluster. If a public network address is used for connection, set this parameter to the public network domain name. If a private network address is used for connection, set this parameter to the private network domain name. If an ELB is used for connection, set this parameter to the ELB address.
- username and password indicate the username and password for connecting to the database. Commands containing authentication passwords pose security risks. Disable the command recording function (history) before running such commands to prevent information leakage.
- port: 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.
Create an empty table for receiving data, for example, customer_t1.
```
CREATE TABLE customer_t1
(
    c_customer_sk             INTEGER,
    c_customer_name           VARCHAR(32)
)
with (orientation = column,compression=middle)
distribute by hash (c_customer_name);
```

GaussDB as a Sink Table

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.

Create a Flink SQL job by referring to Creating a FlinkServer 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 MyUserTable(
  c_customer_sk INTEGER,
  c_customer_name VARCHAR(32)
) WITH(
  'connector' = 'jdbc',
  'url'='jdbc:gaussdb://IP address of the GaussDB server:Database port number/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 connecting to the GaussDB(DWS) database. You need to specify the password when submitting the job.
  'write.mode' = 'upsert',--When data is written in upsert mode, you can set whether to ignore null values. (applicable to MRS 3.3.0 and later versions)
  '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;

NOTE:

The Kafka port can be:
- 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.

properties.group.id indicates the Kafka user group ID. This parameter is mandatory when Kafka functions as the source.
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 is in the Running status.
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 deployed: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.
NOTE:
- 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.
Log in to the GaussDB client and run the following command to check whether data has been sent to the sink table:

Select * from customer_t1;

GaussDB as a Source Table

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.

Create a Flink SQL job by referring to Creating a FlinkServer 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(
  --GaussDB 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:Database port number/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;

NOTE:

The Kafka port can be:
- 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.

properties.group.id indicates the Kafka user group ID. This parameter is mandatory when Kafka functions as the source.
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 is in the Running status.
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 deployed: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.

Create dimension table customer_t2 on the GaussDB client. 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');

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.

Create a Flink SQL job by referring to Creating a FlinkServer 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:Database port number/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;

NOTE:

The Kafka port can be:
- 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.

properties.group.id indicates the Kafka user group ID. This parameter is mandatory when Kafka functions as the source.
System domain name: You can log in to FusionInsight Manager, choose System > Permission > Domain and Mutual Trust, and check the value of Local Domain.

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 deployed:Kafka port number --consumer.config /opt/client/Kafka/kafka/config/ consumer.properties
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
NOTE:
- IP address of the ZooKeeper quorumpeer instance
  To obtain the 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.
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 deployed: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
```

Parent topic: Creating a FlinkServer Job

Previous topic: Creating a FlinkServer Job to Interconnect with a Doris Table

Next topic: Creating a FlinkServer Job to Interconnect with JDBC

Feedback

Was this page helpful?

Helpful Not helpful

Provide feedback

Thank you very much for your feedback. We will continue working to improve the documentation.See the reply and handling status in My Cloud VOC.

The system is busy. Please try again later.

Which of the following issues have you encountered?

Content is inconsistent with the product UI

Unclear descriptions

Lack of examples or code

Incorrect steps

Can't find what I need

Lack of best practices

Feedback (optional)

0/500

Select at least one type of issue, and enter your comments or suggestions.

Enter a maximum of 500 characters.

Submit Cancel

For any further questions, feel free to contact us through the chatbot.

Chatbot