ADB PG版基于Flink 自定义conenctor支持读取（维表）和写入（结果表）。通过Flink SQL即可实现对ADB PG版的访问。

前提条件

版本要求

Flink 1.11及以上版本

ADBPG 6.0版本；

网络要求

ADBPG实例与Flink实例在同一VPC下；

ADBPG设置白名单，开放对Flink实例的网络访问。

操作步骤

设置ADBPG实例

1、购买6.0版本ADBPG实例，创建账号，并设置白名单：

2、连接数据库，创建待写入目标表、和待查询源数据表：

create table test6(a int,b text,c text,d int,e int, f int, g bigint, h float, i double precision, j boolean);

insert into test6 values(0 ,'b0', 'c0', 40,  50,    60,    70,       80.1,    90.1,     'true');

insert into test6 values(0 ,'b0', 'c0', 40,  50,    60,    70,       80.2,    90.2,     'false');

insert into test6 values(1 ,'b1', 'c1', 41,  50,    60,    70,       80.1,    90.1,     'true');

insert into test6 values(1 ,'b1', 'c1', 41,  50,    60,    70,       80.2,    90.2,     'false');

insert into test6 values(2 ,'b2', 'c2', 40,  50,    60,    70,       80.1,    90.1,     'true');

insert into test6 values(2 ,'b2', 'c2', 40,  50,    60,    70,       80.2,    90.2,     'false');

insert into test6 values(3 ,'b3', 'c3', 40,  50,    60,    70,       80.1,    90.1,     'true');

insert into test6 values(3 ,'b3', 'c3', 40,  50,    60,    70,       80.2,    90.2,     'false');

insert into test6 values(4 ,'b4', 'c4', 40,  50,    60,    70,       80.1,    90.1,     'true');

insert into test6 values(4 ,'b4', 'c4', 40,  50,    60,    70,       80.2,    90.2,     'false');

insert into test6 values(5 ,'b5', 'c5', 40,  50,    60,    70,       80.1,    90.1,     'true');

insert into test6 values(5 ,'b5', 'c5', 40,  50,    60,    70,       80.2,    90.2,     'false');

insert into test6 values(6 ,'b6', 'c6', 40,  50,    60,    70,       80.1,    90.1,     'true');

insert into test6 values(6 ,'b6', 'c6', 40,  50,    60,    70,       80.2,    90.2,     'false');

insert into test6 values(7 ,'b7', 'c7', 40,  50,    60,    70,       80.1,    90.1,     'true');

insert into test6 values(7 ,'b7', 'c7', 40,  50,    60,    70,       80.2,    90.2,     'false');

insert into test6 values(8 ,'b8', 'c8', 40,  50,    60,    70,       80.1,    90.1,     'true');

insert into test6 values(8 ,'b8', 'c8', 40,  50,    60,    70,       80.2,    90.2,     'false');

insert into test6 values(9 ,'b9', 'c9', 40,  50,    60,    70,       80.1,    90.1,     'true');

insert into test6 values(9 ,'b9', 'c9', 40,  50,    60,    70,       80.2,    90.2,     'false');

create table test7(a int,b text,c text,d int,e int, f int, g bigint, h float, i double precision, j boolean, k int,l text,m text,n int,o int, p int, q bigint, r float, s double precision, t boolean);

Flink作业开发

1、创建Flink vvp版实例，要保证Flink实例与ADBPG实例处于同一个VPC下；

2、创建SQL作业

3、作业开发

代码参考：

CREATE TEMPORARY TABLE datagen_source2(

 a INT,

 b VARCHAR,

 c CHAR(15),

 d TINYINT,

 e SMALLINT,

 f INT,

 g BIGINT,

 h FLOAT,

 i DOUBLE,

 j BOOLEAN,

 `proctime` AS PROCTIME()

) with (

 'connector' = 'datagen'

);

CREATE TEMPORARY TABLE adbpg_dim2 (

 a INT,

 b VARCHAR,

 c CHAR(15),

 d TINYINT,

 e SMALLINT,

 f INT,

 g BIGINT,

 h FLOAT,

 i DOUBLE,

 j BOOLEAN

) with (

  'connector' = 'adbpg',

  'password' = 'password',

  'tablename' = 'tablename',

  'username' = 'username',

  'url' = 'jdbc:postgresql://url:port/databasename',

  'maxretrytimes' = '2',

  'connectionmaxactive' = '5',

  'targetschema' = 'public',

  'casesensitive' = '0',

  'retrywaittime' = '200',

  'cache' = 'lru',

  'cacheSize'= '1000000',

  'cacheTTLMs' = '2000000000');

CREATE TEMPORARY TABLE adbpg_sink2(

 a INT,

 b VARCHAR,

 c CHAR(15),

 d TINYINT,

 e SMALLINT,

 f INT,

 g BIGINT,

 h FLOAT,

 i DOUBLE,

 j BOOLEAN,

 k INT,

 l VARCHAR,

 m CHAR(15),

 n TINYINT,

 o SMALLINT,

 p INT,

 q BIGINT,

 r FLOAT,

 s DOUBLE,

 t BOOLEAN

) with (

  'connector' = 'adbpg',

  'password' = 'password',

  'tablename' = 'tablename',

  'username' = 'username',

  'url' = 'jdbc:postgresql://url:port/databasename',

  'maxretrytimes' = '2',

  'batchsize' = '100',

  'connectionmaxactive' = '5',

  'conflictmode' = 'ignore',

  'usecopy' = '0',

  'targetschema' = 'public',

  'exceptionmode' = 'ignore',

  'casesensitive' = '0',

  'writemode' = '0',

  'retrywaittime' = '200'

);

insert into adbpg_sink2 select T.a, T.b, T.c, T.d, T.e, T.f, T.g, T.h, T.i, T.j, H.a, H.b, H.c, H.d, H.e, H.f, H.g, H.h, H.i, H.j FROM datagen_source2 AS T JOIN adbpg_dim2 FOR SYSTEM_TIME AS OF T.proctime AS H ON MOD(T.a, 10) = H.a;

4、上传jar包：

https://adbpg-public.oss-cn-beijing.aliyuncs.com/flink-connector-adbpg-1.11.1-jar-with-dependencies.jar

5、运行上线：

点击验证、运行、上线，观察日志和数据库判断是否有异常，是否成功写入数据库。

维表参数说明

结果表参数说明

/