做结果表

格式语法

SQL语法格式可能在不同Flink环境下有细微差异，具体以事件环境格式为准，with后面的参数名称及参数值以此文档为准。

 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配置参数

Flink SQL中设置的PRIMARY KEY将自动映射到dws-client中的uniqueKeys。参数跟随client版本发布，参数功能与client一致，以下参数说明表示为最新参数。

dws client参数全量支持在flink sql通过key方式配置，下表参数为兼容1.x版本参数，当同时配置2.x和1.x参数时生效2.x版本参数值：

使用flink SQL直连DN入库

该能力依赖flink sql DISTRIBUTEBY能力，mrs有提供此能力，具体请参见Flink SQL语法增强。

connector提供udf函数可根据分布列值计算出下游并发结合flink sql DISTRIBUTEBY能力实现将数据按DN分区能力，示例：

1. 需要在SQL中引入UDF。

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

2. 正常写Source SQL。

   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. Sink表定义SQL中需要新增一个字段并且要求int类型值用于接收UDF计算的结果，示例中叫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. Insert into sql使用 udf获取数据下游算子信息，同时使用DISTRIBUTEBY对返回结果做数据分区，数据就会按照udf返回信息到下游指定并行度。

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

UDF函数DnHashFunction参数说明

参数格式

dn_hash（'dws表名',sink并行度,最大并行度,dws作为分布列的数据在源数据的字段名称{1,}）

参数说明

1. 使用时上游并行度必须不多于sink并行度，DnHashFunction同样是通过进程内获取sink 算子初始化的dws client实例获取到的表元数据，如果当前进程无sink算子就会导致无法获取client实例。
2. 使用后会增加一个hash算子，如果链路有多个算子处理业务，当执行hash算子后不可以再有改变数据分区的算子，否则数据会被再次分区就不能到达指定sink算子。
3. 最大并行度默认flink自动调整的，算法中需要使用，因此自动调整的无法使用，必须通过参数设置固定并把设置额值作为UDF的参数，可以通过参数pipeline.max-parallelism设置或者jar方式通过API设置：

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

4. 如果分布列包含多个字段，分布列的字段顺序需要保持和DWS一致，分布列支持的字段类型和dws client一致参考参数WRITE_PARTITION_POLICY，使用功能同样需要额外配置，不可自行使用。

示例

该示例是从kafka数据源中读取数据，写入DWS结果表中，并指定攒批时间不超过10秒，每批数据最大30000条，其具体步骤如下：

1. 在DWS数据库中创建表public.dws_order：

   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. 消费Kafka中order_test topic中的数据作为数据源，public.dws_order作为结果表，Kafka数据为JSON格式，并且字段名称和数据库字段名称一一对应：

   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' = 'DWSPassword', 'autoFlushMaxInterval' = '10s', 'autoFlushBatchSize' = '30000' ); insert into dwsSink select * from kafkaSource;

3. 给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. 等10秒后在DWS表中查询结果：

   1	select * from dws_order

   结果如下：

   

常见问题

• Q：writeMode参数设置什么值比较合适？

  A：根据业务场景分update（只更新存在的数据）和upsert（对于同一主键数据如果存在就更新，不存在就新增一条数据）两个类型，推荐直接使用auto方式即可，该方式下会根据数据量的大小自动选择，如果数据量较大会增大攒批参数autoFlushBatchSize，即可提升入库性能。

• Q：autoFlushBatchSize和autoFlushMaxInterval怎么设置比较合适？
  A：autoFlushBatchSize参数用于设置最大攒批条数，autoFlushMaxInterval参数用于设置最大攒批间隔，两个参数分别从时间和空间维度管控攒批。

  • 通过autoFlushMaxInterval可保证数据量较小时的时效性，如对时效性无强制要求通常不建议设置的太小，建议不低于3s走默认值即可。
  • 通过autoFlushBatchSize可控制一批数据的最大条数，一般来说攒批量越大，对于整体入库性能会更好，对性能来说通常该参数的设置推荐越大越好，参数的设置根据业务数据的大小以及flink运行内存来设置，保证不内存溢出。

    对于大多业务来说无需设置autoFlushMaxInterval，将autoFlushBatchSize设置为50000即可。

• Q: 遇到数据库死锁了怎么办？

  A：通常出现死锁大致分为行锁死锁和分布式死锁。

  • 行锁：该场景通常为同一主键数据的并发更新造成行锁，该情况可以通过对数据做key by解决，key by必须根据数据库主键做，保证同一个主键数据会在同一个并发中，破坏掉并发更新的条件，无法造成死锁。Flink SQL做key by需要Flink本身支持，对于DLI/MRS均能实现，如MRS flink通过增加参数“key-by-before-sink=true”可实现key by。具体怎么使用以实现方为准，对于无法使用的建议使用API方式入库。
  • 分布式死锁：该场景通常为列存表的并发更新造成分布式死锁，暂无法解决，建议使用行存或者hstore。