DWS源表

功能描述

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

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

前提条件

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

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

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

语法格式

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

create table dwsSource (
  attr_name attr_type 
  (',' attr_name attr_type)* 
  (','PRIMARY KEY (attr_name, ...) NOT ENFORCED)
  (',' watermark for rowtime_column_name as watermark-strategy_expression)
)
with (
  'connector.type' = 'gaussdb',
  'connector.url' = '',
  'connector.table' = '',
  'connector.username' = '',
  'connector.password' = ''
);

参数说明

示例

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

  表car_info没有在schema下时。

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

  create table dwsSource( 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' );

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

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

  create table dwsSource( 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' = 'test_schema\".\"test', 'connector.username' = 'xx', 'connector.password' = 'xx' );

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

  create table dwsSource(
    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'
  );