GaussDB(DWS) Result Table

Function

DLI outputs the Flink job output data to GaussDB(DWS). GaussDB(DWS) database kernel is compliant with PostgreSQL. The PostgreSQL database can store data of more complex types and deliver space information services, multi-version concurrent control (MVCC), and high concurrency. It applies to location applications, financial insurance, and e-Commerce.

GaussDB(DWS) is an online data processing database based on the cloud infrastructure and platform and helps you mine and analyze massive sets of data.

Prerequisites

• When creating a Flink OpenSource SQL job, you need to set Flink Version to 1.12 on the Running Parameters tab of the job editing page, select Save Job Log, and set the OBS bucket for saving job logs.
• You have created a GaussDB(DWS) cluster. For details about how to create a GaussDB(DWS) cluster, see Creating a Cluster in the Data Warehouse Service Management Guide.
• You have created a GaussDB(DWS) database table.
• An enhanced datasource connection has been created for DLI to connect to GaussDB(DWS) clusters, so that jobs can run on the dedicated queue of DLI and you can set the security group rules as required.

Precautions

• To use the upsert mode, you must define the primary key for both the GaussDB(DWS) result table and the GaussDB(DWS) table connected to the result table.
• If tables with the same name exist in different GaussDB(DWS) schemas, you need to specify the schemas in the Flink open source SQL statements.
• Before submitting a Flink job, you are advised to select Save Job Log and set the OBS bucket for saving job logs. This helps you view logs and locate faults when the job fails to be submitted or runs abnormally.
• If you use the gsjdbc4 driver for connection, set driver to org.postgresql.Driver. You can omit this parameter because the gsjdbc4 driver is the default one.
  For example, run the following statements to use the gsjdbc4 driver to write data to GaussDB(DWS) in upsert mode:

  1 2 3 4 5 6 7 8 9 10 11 12 13

  create table dwsSink( car_id STRING, car_owner STRING, car_brand STRING, car_speed INT ) with ( 'connector' = 'gaussdb', 'url' = 'jdbc:postgresql://DwsAddress:DwsPort/DwsDatabase', 'table-name' = 'car_info', 'username' = 'DwsUserName', 'password' = 'DwsPasswrod', 'write.mode' = 'upsert' );

Syntax

Do not set all attributes in a GaussDB(DWS) result table to PRIMARY KEY.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

create table dwsSink (
  attr_name attr_type 
  (',' attr_name attr_type)* 
  (','PRIMARY KEY (attr_name, ...) NOT ENFORCED)
)
with (
  'connector' = 'gaussdb',
  'url' = '',
  'table-name' = '',
  'driver' = '',
  'username' = '',
  'password' = ''
);

Parameters

Example

In this example, data is read from the Kafka data source and written to the GaussDB(DWS) result table in insert mode. The procedure is as follows:

1. Create an enhanced datasource connection in the VPC and subnet where GaussDB(DWS) and Kafka locate, and bind the connection to the required Flink elastic resource pool.
2. Set GaussDB(DWS) and Kafka security groups and add inbound rules to allow access from the Flink queue. Test the connectivity using the GaussDB(DWS) and Kafka address. If the connection is successful, the datasource is bound to the queue. Otherwise, the binding fails.
3. Connect to the GaussDB(DWS) database and create a table named dws_order.

   create table public.dws_order(
     order_id VARCHAR,
     order_channel VARCHAR,
     order_time VARCHAR,
     pay_amount FLOAT8,
     real_pay FLOAT8,
     pay_time VARCHAR,
     user_id VARCHAR,
     user_name VARCHAR,
     area_id VARCHAR);

4. Create a Flink OpenSource SQL job. Enter the following job script and submit the job. The job script uses the Kafka data source and the GaussDB(DWS) result table.
   When you create a job, set Flink Version to 1.12 on the Running Parameters tab. Select Save Job Log, and specify the OBS bucket for saving job logs. Change the values of the parameters in bold as needed in the following script.

   CREATE TABLE kafkaSource (
     order_id string,
     order_channel string,
     order_time string,
     pay_amount double,
     real_pay double,
     pay_time string,
     user_id string,
     user_name string,
     area_id string
   ) WITH (
     'connector' = 'kafka',
     'topic' = 'KafkaTopic',
     'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort',
     'properties.group.id' = 'GroupId',
     'scan.startup.mode' = 'latest-offset',
     'format' = 'json'
   );
   
   CREATE TABLE dwsSink (
     order_id string,
     order_channel string,
     order_time string,
     pay_amount double,
     real_pay double,
     pay_time string,
     user_id string,
     user_name string,
     area_id string
   ) WITH (
     'connector' = 'gaussdb',
     'url' = 'jdbc:postgresql://DWSAddress:DWSPort/DWSdbName',
     'table-name' = 'dws_order',
     'driver' = 'org.postgresql.Driver',
     'username' = 'DWSUserName',
     'password' = 'DWSPassword',
     'write.mode' = 'insert'
   );
   
   insert into dwsSink select * from kafkaSource;

5. Connect to the Kafka cluster and enter the following test data to Kafka:

   {"order_id":"202103241000000001", "order_channel":"webShop", "order_time":"2021-03-24 10:00:00", "pay_amount":"100.00", "real_pay":"100.00", "pay_time":"2021-03-24 10:02:03", "user_id":"0001", "user_name":"Alice", "area_id":"330106"}

6. Run the following SQL statement in GaussDB(DWS) to view the data result:

    select * from dws_order

   The data result is as follows:

   202103241000000001	webShop	2021-03-24 10:00:00	100.0	100.0	2021-03-24 10:02:03	0001	Alice	330106

FAQ

• Q: What should I do if the Flink job execution fails and the log contains the following error information?

  java.io.IOException: unable to open JDBC writer
  ...
  Caused by: org.postgresql.util.PSQLException: The connection attempt failed.
  ...
  Caused by: java.net.SocketTimeoutException: connect timed out

  A: The datasource connection is not bound or the binding fails.

• Q: How can I configure a GaussDB(DWS) table that is in a schema?
  A: When GaussDB(DWS) table test is in schema ads_game_sdk_base, refer to the 'table-name' parameter setting in the following example:

  CREATE TABLE ads_rpt_game_sdk_realtime_ada_reg_user_pay_mm (
    ddate DATE,
    dmin TIMESTAMP(3),
    game_appkey VARCHAR,
    channel_id VARCHAR,
    pay_user_num_1m bigint,
    pay_amt_1m bigint,
    PRIMARY KEY (ddate, dmin, game_appkey, channel_id) NOT ENFORCED
  ) WITH (
    'connector' = 'gaussdb',
    'url' = 'jdbc:postgresql://<yourDwsAddress>:<yourDwsPort>/dws_bigdata_db',
    'table-name' = 'ads_game_sdk_base\".\"test',
    'username' = '<yourUsername>',
    'password' = '<yourPassword>',
    'write.mode' = 'upsert'
  );