DWS结果表

功能描述

DLI将Flink作业的输出数据输出到数据仓库服务（DWS）中。DWS数据库内核兼容PostgreSQL，PostgreSQL数据库可存储更加复杂类型的数据，支持空间信息服务、多版本并发控制（MVCC）、高并发，适用场景包括位置应用、金融保险、互联网电商等。

数据仓库服务（Data Warehouse Service，简称DWS）是一种基于基础架构和平台的在线数据处理数据库，为用户提供海量数据挖掘和分析服务。DWS的更多信息，请参见《数据仓库服务管理指南》。

前提条件

• 请务必确保您的账户下已在数据仓库服务（DWS）里创建了DWS集群。

  如何创建DWS集群，请参考《数据仓库服务管理指南》中“创建集群”章节。

• 请确保已创建DWS数据库表。
• 该场景作业需要运行在DLI的独享队列上，因此要与DWS集群建立增强型跨源连接，且用户可以根据实际所需设置相应安全组规则。
• 如何建立增强型跨源连接，请参考《数据湖探索用户指南》中增强型跨源连接章节。
• 如何设置安全组规则，请参见《虚拟私有云用户指南》中“安全组”章节。

语法格式

DWS结果表中不允许指定所有属性为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.type' = 'gaussdb',
  'connector.url' = '',
  'connector.table' = '',
  'connector.driver' = '',
  'connector.username' = '',
  'connector.password' = ''
);

参数说明

注意事项

无

示例

• 使用gsjdbc4驱动连接时，加载的数据库驱动类为：org.postgresql.Driver。该驱动为默认，创建表时可以不填该驱动参数。
  • 使用upsert模式，写入数据到DWS

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

    create table dwsSink( car_id STRING, car_owner STRING, car_brand STRING, car_speed INT ) with ( 'connector.type' = 'gaussdb', 'connector.url' = 'jdbc:postgresql://xx.xx.xx.xx:8000/xx', 'connector.table' = 'car_info', 'connector.username' = 'xx', 'connector.password' = 'xx', 'connector.write.mode' = 'upsert', 'connector.write.flush.interval' = '30s' );

    当DWS表test在名为ads_game_sdk_base的schema下时，可以参考如下样例。

    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.type' = 'gaussdb',
      'connector.url' = 'jdbc:postgresql://xx.xx.xx.xx:8000/dws_bigdata_db',
      'connector.table' = 'ads_game_sdk_base\".\"test',
      'connector.username' = 'xxxx',
      'connector.password' = 'xxxxx',
      'connector.write.mode' = 'upsert',
      'connector.write.flush.interval' = '30s'
    );

• 使用gsjdbc200驱动连接时，加载的数据库驱动类为：com.huawei.gauss200.jdbc.Driver。
  当DWS表test在名为ads_game_sdk_base的schema下时，可以参考如下样例。

  create table dwsSink(
    car_id STRING,
    car_owner STRING,
    car_brand STRING,
    car_speed INT
  ) with (
    'connector.type' = 'gaussdb',
    'connector.table' = 'ads_game_sdk_base\".\"test',
    'connector.driver' = 'com.huawei.gauss200.jdbc.Driver',
    'connector.url' = 'jdbc:gaussdb://xx.xx.xx.xx:8000/xx',
    'connector.username' = 'xx',
    'connector.password' = 'xx',
    'connector.write.mode' = 'upsert',
    'connector.write.flush.interval' = '30s'
  );