Result Tables

Syntax

Different Flink environments may have slightly varying SQL syntax formats. Check the event environment format for more details. The parameter names and values that come after with are specific to this document.

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

create table dwsSink (
  attr_name attr_type 
  (',' attr_name attr_type)* 
  (','PRIMARY KEY (attr_name, ...) NOT ENFORCED)
)
with (
  'connector' = 'dws',
  'url' = '',
  'tableName' = '',
  'username' = '',
  'password' = ''
);

Flink SQL Configuration Parameters

Primary keys set in Flink SQL are automatically mapped to unique keys in DWS client. The parameters are released with the client version. The parameter functions are the same as those on the client. The following parameters are the latest parameters.

When dws-client is of version 2.x, all parameters can be configured in Flink SQL statements using keys. The parameters in the following table are compatible with version 1.x. If parameters are configured for both version 2.x and version 1.x are configured, the parameters for version 2.x take effect.

Connecting to DNs using Flink SQL

This capability depends on the Flink SQL DISTRIBUTE BY capability. MRS provides this capability. For details, see Flink SQL Syntax Enhancement.

connector provides the UDF function to calculate the downstream concurrency based on the distribution column value and use the Flink SQL DISTRIBUTE BY capability to partition data by DN. The following is an example:

1. Introduce the UDF to the SQL statement.

   CREATE temporary  FUNCTION dn_hash AS 'com.huaweicloud.dws.connectors.flink.partition.DnHashFunction';

2. Write the source SQL statement (no special operation or modification is required).

   CREATE TABLE users
   (
       id         BIGINT,
       name       STRING,
       age        INT,
       text       STRING,
       created_at TIMESTAMP(3),
       updated_at TIMESTAMP(3)
   ) WITH (
         'connector' = 'datagen',
         'fields.id.kind' = 'sequence',
         'fields.id.start' = '1',
         'fields.id.end' = '1000',
         'fields.name.length' = '10',
         'fields.age.min' = '18',
         'fields.age.max' = '60',
         'fields.text.length' = '5'
         )

3. Add a field to the sink table definition SQL statement and ensure that the value of the int type is used to receive the UDF calculation result. In the example, the field name is dn_hash.

   create table dws_users
   (
       dn_hash int,
       id         BIGINT,
       name       STRING,
       age        INT,
       text       STRING,
       created_at TIMESTAMP(3),
       updated_at TIMESTAMP(3),
       PRIMARY KEY (id) NOT ENFORCED
   ) WITH (
         'connector' = 'dws',
         'url' = '%s',
         'tableName' = 'test.users',
         'username' = '%s',
         'autoFlushBatchSize' = '50000',
         'password' = '%s'
         )

4. The INSERT INTO SQL statement uses a UDF to retrieve information about the downstream operator and employs DISTRIBUTE BY to partition the result data. The data is then distributed to the downstream based on the parallelism degree specified by the UDF.

   insert into dws_users select 
   /*+ DISTRIBUTEBY('dn_hash') */ 
   dn_hash('test.users',10,1024, id) as dn_hash, * from users

DnHashFunction Parameter Description

Parameter format

dn_hash ('DWS table name', Sink parallelism degree, Maximum parallelism degree, Field name of the DWS partitioned column data in the source data{1,})

Parameter description

1. The upstream parallelism degree must be less than or equal to the sink parallelism degree. The DnHashFunction function retrieves the table metadata from the DWS client instance, which is initialized by the sink operator during processing. If the current process does not include a sink operator, the client instance cannot be obtained.
2. After the hash operator is applied, a new hash operator is introduced. If multiple operators are used to process services in a chain, no operator can alter the data partitioning after the hash operator has been executed. Otherwise, the data will be repartitioned, and the specified sink operator will not be reached.
3. By default, Flink automatically adjusts the maximum parallelism degree, which is required by the algorithm. Since automatic adjustment cannot be used in this case, the parameter must be manually set to a fixed value, which will then be used as the UDF parameter. You can set this parameter either using the pipeline.max-parallelism parameter or by utilizing the API in JAR files.

   StreamExecutionEnvironment evn = StreamExecutionEnvironment.getExecutionEnvironment();
   evn.setParallelism(1);
   evn.setMaxParallelism(1024);

4. If the distributed column contains multiple fields, the field sequence of the distribution column must be the same as that of DWS. The field type supported by the distributed column is the same as that of DWS client. For details, see the WRITE_PARTITION_POLICY parameter. If you want to use other functions, you will need to make additional configurations.

Examples

Here is an example of how to read data from the Kafka data source and store it in the DWS result table. Each batch can hold a maximum of 30,000 data records and must be stored within 10 seconds.

1. Create the public.dws_order table in the DWS database.

   1 2 3 4 5 6 7 8 9 10 11

   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 );

2. The data in the order_test topic of the Kafka is used as the data source, the public.dws_order is used as the result table, the Kafka data is in JSON format, and the field names correspond to the database field names.

   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 32 33 34 35 36 37 38 39 40

   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' = 'order_test', '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' = 'dws', 'url' = 'jdbc:gaussdb://DWSAddress:DWSPort/DWSdbName', 'tableName' = 'dws_order', 'username' = 'DWSUserName', 'password' = 'password', 'autoFlushMaxInterval' = '10s', 'autoFlushBatchSize' = '30000' ); insert into dwsSink select * from kafkaSource;

3. Insert test data to Kafka.

   1	{"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"}

4. Wait for 10 seconds and query the result in the DWS table.

   1	select * from dws_order

   The result is shown in the following figure.

   

FAQs

• Q: What is the proper value of the writeMode parameter?

  A: There are two types of service scenarios: update and upsert. update only modifies existing data, while upsert updates data if the primary key already exists and adds a new record if it does not. You are advised to use the auto mode, where the system selects a value based on the data volume. If the data volume is large, increasing the autoFlushBatchSize value can improve the performance of importing data to the database.

  Specifically, when auto is writeMode (The default value of copyWriteBatchSize is 1000), upsert is used once the batch size is less than copyWriteBatchSize. When the batch size is greater than copyWriteBatchSize, there are two cases: If the primary key exists, copy the data to the temporary table and upsert the data. If the primary key does not exist, copy the data to the target table.

• Q: How to set autoFlushBatchSize and autoFlushMaxInterval properly?
  A: The autoFlushBatchSize parameter sets the maximum number of batches that can be stored, while the autoFlushMaxInterval parameter sets the maximum time interval for storing batches. These two parameters control the number of batches that can be stored in terms of both time and space.

  • When dealing with small data volumes, setting a value for autoFlushMaxInterval can ensure timely updates. But if timeliness is not a top priority, it is better to avoid setting a small value. It is advisable to either stick with the default value or set it to a value greater than or equal to three seconds.
  • The autoFlushBatchSize parameter lets you limit the number of data records in a batch. Generally, a higher number of records in a batch results in better data import performance. It is best to set this parameter to a larger value, while considering the service data size and Flink running memory to prevent memory overflow problems.

    For most services, you do not need to set autoFlushMaxInterval. Set autoFlushBatchSize to 50000.

• Q: What can I do if a database deadlock occurs?

  A: Deadlocks are classified into row locks and distributed deadlocks.

  • Row lock occurs when multiple updates are made to data with the same primary key. To solve this issue, you can perform key by on the data, based on the primary key of the database. This ensures that data with the same primary key is in the same concurrency, eliminating the possibility of concurrent updates and preventing deadlocks. You need to set connectionSize to 1 on the DWS-connector side to ensure that only one connection is used to import data to the database. This prevents concurrent deadlock on the tool side. To perform key by using Flink SQL, Flink must support it. Both DLI and MRS can implement key by. For instance, you can set key-by-before-sink to true in MRS Flink to implement key by. To learn how to use the API, refer to the implementation party for more details. If the API is not usable, it is recommended to import it to the database using the API.
  • Distributed deadlock occurs when column-store tables are concurrently updated, and it cannot be resolved at present. To avoid this, use row-store or hstore.
• Q: How can I address a timeout issue when executing an SQL statement to import data into the database?

  A: If the error message canceling statement due to statement timeout is displayed, you need to:

  Analyze the top SQL statements of the DWS kernel and check whether the specific SQL statement matches the SQL statement that reports the error.

  • Solution 1: The SQL statement timeout may be caused by insufficient resources on the DWS node. You can view the DWS resource monitoring details and pay attention to the CPU, I/O, and disk metrics. If there is obvious fluctuation, the resources may be insufficient. To solve the problem, you need to upgrade the resource configuration.
  • Solution 2: The possible cause is that the default timeout interval configured on the connector tool is too short to allow the complete execution of the SQL statement (the default value is usually 5 minutes). In this case, you need to increase the timeout interval by adding the following configuration to the WITH parameter of Flink SQL.

  Table 5 Parameters for resolving SQL timeout issues in different versions

  DWS-connector	Parameter	Value (Unit: ms)	Example
  1.x	connectionTimeOut	600000	'connectionTimeOut' = '600000'
  2.x	dws.client.timeout.statement	900000	'dws.client.timeout.statement' = '900000'

• Q: What should I do if an exception occurs when data is imported to the Flink TaskManager, but the exception is not displayed on the Flink UI?

  A: This is caused by the Flink synchronization mechanism. Flink usually detects only the exceptions directly thrown by the sink operator, but does not detect the exceptions thrown by other sub-threads started by the sink operator. However, this does not affect the retry policy of Flink.

• Q: When is the ignoreNullWhenUpdate parameter used?

  A: The ignoreNullWhenUpdate parameter is usually used to address the service of updating some columns. Based on the update of some columns, the SQL statements for importing data to the database need to be customized based on data features (that is, only the columns with values are updated during upsert). Otherwise, data may be overwritten, causing data inconsistency. However, the following problems may occur after this parameter is used:

  • There are too many groups. After this parameter is configured, a batch of data usually has multiple data features. Therefore, data with different data features needs to be divided, and an independent SQL statement is executed for each group to import data to the database. In this case, there may be an extreme situation. That is, the original table is a large wide table and the data features are scattered. As a result, there are too many groups, and SQL statements are frequently executed in a batch, failing to achieve the purpose of executing a large number of SQL statements.
  • The upsert performance of some columns is poor. This is the performance baseline of the GaussDB kernel. Upsert of some columns is more than three times slower than upsert of all columns.

  If the preceding performance problem occurs, rectify the data on the service side or tolerate the impact of disabling this parameter.

• Q: What can I do if spaces before and after data need to be ignored?

  A: You can use the trim function in Flink SQL to ignore spaces before and after data. For example, run insert into sink select trim(a),trim(b) from source;

• Q: What can I do if \0 in the varchar field of the source data (such as MySQL) is lost after synchronization?

  A: Data loss is normal and is not related to the DWS-connector tool.

  \0 is generally considered as a terminator in characters and may be truncated or lost. Some CDC tools (such as MySQL-CDC) supports deserialization logic and escapes \0 to NULL. As a result, \0 is lost.

  • Workaround 1: Use other data types, such as varbinary, instead of varchar.
  • Workaround 2: Use the Stream API to start the job and add a custom deserializer to implement this logic.
  • Before importing data to the database, the service side can modify the special data to prevent the varchar field from carrying the special data.
• Q: When the sink service table lacks a primary key and writeMode is set to copy, data duplication can occur if a fault triggers a Flink retry. How can this issue be resolved?

  A: During the Flink retry, the data from the last successful checkpoint is restored. That is, the source operator rolls back the consumption point to the position recorded in the checkpoint. As a result, some data is repeatedly imported to the database (from the time when the last checkpoint is successful to the time when the fault occurs). If the service table does not have a primary key, a copy of data will be inserted again, resulting in duplicate data or data inconsistency.

  • Solution 1: Upsert semantics is used together with the primary key of the service table to ensure that the written results are consistent.
  • Solution 2: Enable transactions on the sink side. Some data sources support two-phase commit. That is, Flink pre-imports data, and a transaction is committed only after the checkpoint of the sink operator is successful. However, dws-connector does not support this solution.