• 創(chuàng)建3.6.0及以上版本實(shí)時計算集群。

• 創(chuàng)建6.0版本云原生數(shù)據(jù)倉庫AnalyticDB PostgreSQL版實(shí)例（實(shí)時計算集群和云原生數(shù)據(jù)倉庫AnalyticDB PostgreSQL版實(shí)例需要位于同一VPC下，且云原生數(shù)據(jù)倉庫AnalyticDB PostgreSQL版實(shí)例的白名單規(guī)則允許Flink集群網(wǎng)段訪問）。

CREATE TABLE dim_adbpg(
        id int,
        username varchar,
        INDEX(id)
) with(
        type='custom',
        tableFactoryClass='com.alibaba.blink.customersink.ADBPGCustomSourceFactory',
        url='jdbc:postgresql://內(nèi)網(wǎng)連接串/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時需要指定在代碼中加入維表標(biāo)識 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. 登錄實(shí)時計算控制臺，在頁面頂部菜單欄上，鼠標(biāo)懸停在用戶頭像上，單擊項目管理。在項目管理>項目列表頁面，單擊項目名進(jìn)入自己創(chuàng)建的項目。

2. 單擊開發(fā)>新建作業(yè)，創(chuàng)建數(shù)據(jù)寫入的Flink SQL作業(yè)。

3. 目前采用Flink自定義維表的方式支持讀取云原生數(shù)據(jù)倉庫AnalyticDB PostgreSQL版目標(biāo)表數(shù)據(jù)，使用自定義維表功能上線前需要在資源引用界面上傳及引用.jar包，編寫完作業(yè)后點(diǎn)擊資源引用>新建資源>上傳JAR包>更多>引用。下載JAR包。

4. 完成作業(yè)開發(fā)后，依次點(diǎn)擊保存、上線，即可上線該任務(wù)。

5. 繼續(xù)點(diǎn)擊運(yùn)維，啟動對應(yīng)項目即可啟動任務(wù)。

這里給出讀取云原生數(shù)據(jù)倉庫AnalyticDB PostgreSQL版數(shù)據(jù)打印到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://內(nèi)網(wǎng)連接串/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;