通过实时计算Flink版读取云原生数据仓库AnalyticDB PostgreSQL版

• 创建3.6.0及以上版本实时计算集群。

• 创建6.0版本云原生数据仓库AnalyticDB PostgreSQL版实例（实时计算集群和云原生数据仓库AnalyticDB PostgreSQL版实例需要位于同一VPC下，且云原生数据仓库AnalyticDB PostgreSQL版实例的白名单规则允许Flink集群网段访问）。

CREATE TABLE dim_adbpg(
        id int,
        username varchar,
        INDEX(id)
) with(
        type='custom',
        tableFactoryClass='com.alibaba.blink.customersink.ADBPGCustomSourceFactory',
        url='jdbc:postgresql://内网连接串/databasename',
        tableName='tablename',
        userName='username',
        password='password',
        joinMaxRows='100',
        maxRetryTimes='1',
        connectionMaxActive='5',
        retryWaitTime='100',
        targetSchema='public',
        caseSensitive='0',
        cache='LRU',
        cacheSize='1000',
        cacheTTLMs='10000',
        cacheReloadTimeBlackList='2017-10-24 14:00 -> 2017-10-24 15:00',
        partitionedJoin='true'
);

-- join时需要指定在代码中加入维表标识 FOR SYSTEM_TIME AS OF PROCTIME()
INSERT INTO print_sink
SELECT R.c1, R.a2, R.a3, R.a4, R.a5, R.a6, R.a6, R.a8, R.a9, R.a10, R.a11, R.a13, T.username
FROM s_member_cart_view AS R
left join
dim_adbpg FOR SYSTEM_TIME AS OF PROCTIME() AS T
on R.c1 = T.id;

1. 登录实时计算控制台，在页面顶部菜单栏上，鼠标悬停在用户头像上，单击项目管理。在项目管理>项目列表页面，单击项目名进入自己创建的项目。

2. 单击开发>新建作业，创建数据写入的Flink SQL作业。

3. 目前采用Flink自定义维表的方式支持读取云原生数据仓库AnalyticDB PostgreSQL版目标表数据，使用自定义维表功能上线前需要在资源引用界面上传及引用.jar包，编写完作业后点击资源引用>新建资源>上传JAR包>更多>引用。下载JAR包。

4. 完成作业开发后，依次点击保存、上线，即可上线该任务。

5. 继续点击运维，启动对应项目即可启动任务。

这里给出读取云原生数据仓库AnalyticDB PostgreSQL版数据打印到Flink日志中的Flink SQL示例：

--SQL
--********************************************************************--
--Author: zihua
--CreateTime: 2019-09-07 10:34:34
--********************************************************************--

CREATE TABLE s_member_cart
(
        a1 int,
        a2 tinyint    ,
        a3 smallint  ,
        a4 int,
        a5 boolean,
        a6 FLOAT     ,
        a7 DECIMAL   ,
        a8 double,
        a9 date   ,
        a10 time      ,
        a11 timestamp ,
        a12 tinyint
) WITH (
        type='random'
);

CREATE VIEW s_member_cart_view AS
SELECT MOD(a1, 10) c1, a2, a3, a4, a5, a6, a6, a8, a9, a10, a11, case when a12 >0 then 'test1' else 'test5' end as b12,'{ "customer": "中文56", "items": {"product": "Beer","qty": 6}}' a13
FROM s_member_cart;

--adbpg dim index
CREATE TABLE dim_adbpg(
        id int,
        username varchar,
        INDEX(id)
) with(
        type='custom',
        tableFactoryClass='com.alibaba.blink.customersink.ADBPGCustomSourceFactory',
        url='jdbc:postgresql://内网连接串/databasename',
        tableName='tablename',
        userName='username',
        password='password',
        joinMaxRows='100',
        maxRetryTimes='1',
        connectionMaxActive='5',
        retryWaitTime='100',
        targetSchema='public',
        caseSensitive='0',
        cache='LRU',
        cacheSize='1000',
        cacheTTLMs='10000',
        cacheReloadTimeBlackList='2017-10-24 14:00 -> 2017-10-24 15:00',
        partitionedJoin='true'
);

-- ads sink.
CREATE TABLE print_sink (
        B1 int,
        B2 tinyint  ,
        B3 smallint  ,
        B4 int,
        B5 boolean,
        B6 FLOAT     ,
        B7 FLOAT   ,
        B8 double,
        B9 date   ,
        B10 time      ,
        B11 timestamp ,
        B12 varchar,
        B15 varchar,
        PRIMARY KEY(B1)
) with (
        type='print'
);


INSERT INTO print_sink
SELECT R.c1, R.a2, R.a3, R.a4, R.a5, R.a6, R.a6, R.a8, R.a9, R.a10, R.a11, R.a13, T.username
FROM s_member_cart_view AS R
left join
dim_adbpg FOR SYSTEM_TIME AS OF PROCTIME() AS T
on R.c1 = T.id;