通过Flink CDC实时订阅全量和增量数据（邀测中）

登录云原生数据仓库AnalyticDB PostgreSQL版控制台，找到目标实例，单击实例ID。

在基本信息页面的右侧下，单击登录数据库。

创建测试数据库和源表adbpg_source_table，并向源表写入50条数据。

-- 创建测试数据库
CREATE DATABASE testdb;
-- 切换至testdb数据库并创建schema
CREATE SCHEMA testschema;
-- 创建源表adbpg_source_table
CREATE TABLE testschema.adbpg_source_table(
  id int,
  username text,
  PRIMARY KEY(id)
);
-- 向adbpg_source_table的表中写入50条数据
INSERT INTO testschema.adbpg_source_table(id, username)
SELECT i, 'username'||i::text
FROM generate_series(1, 50) AS t(i);

创建结果表adbpg_sink_table，用于Flink写入结果数据。

CREATE TABLE testschema.adbpg_sink_table(
  id int,
  username text,
  score int
);

登录实时计算控制台，在Flink全托管页签，单击目标工作空间操作列下的控制台。

在左侧导航栏，单击连接器，创建并上传CDC连接器。

在左侧导航栏，单击数据开发 > ETL。

单击顶部菜单栏新建，选择空白的流作业草稿，单击下一步，在新建文件草稿对话框，填写作业配置信息。

单击创建。

创建数据源datagen_source（用于生成模拟数据）和source_adbpg（用于从 AnalyticDB for PostgreSQL数据库中捕获实时的数据变更），将两个数据源关联，并将结果写入结果表sink_adbpg（将处理后的数据写入AnalyticDB for PostgreSQL）。

将以下作业代码拷贝到作业文本编辑区。

---创建datagen源表，使用datagen连接器生成流式数据
CREATE TEMPORARY TABLE datagen_source (
 id INT,
 score INT
) WITH (
 'connector' = 'datagen',
 'fields.id.kind'='sequence',
 'fields.id.start'='1',
 'fields.id.end'='100',
 'fields.score.kind'='random',
 'fields.score.min'='70',
 'fields.score.max'='100'
);

--创建adbpg源表,使用adbpg-cdc连接器，通过slot.name和逻pgoutput捕获adbpg_source_table的数据变更
CREATE TEMPORARY TABLE source_adbpg(
 id int,
 username varchar,
 PRIMARY KEY(id) NOT ENFORCED
) WITH(
  'connector' = 'adbpg-cdc', 
  'hostname' = 'gp-bp16v8cgx46ns****-master.gpdb.rds.aliyuncs.com',
  'port' = '5432',
  'username' = 'account****',
  'password' = 'password****',
  'database-name' = 'testdb',
  'schema-name' = 'testschema',
  'table-name' = 'adbpg_source_table',
  'slot.name' = 'flink',
  'decoding.plugin.name' = 'pgoutput'
);

--创建adbpg结果表，将处理后的结果写入数据库的目标表adbpg_sink_table。
CREATE TEMPORARY TABLE sink_adbpg (
  id int,
  username varchar,
  score int
) WITH (
  'connector' = 'adbpg', 
  'url' = 'jdbc:postgresql://gp-bp16v8cgx46ns****-master.gpdb.rds.aliyuncs.com:5432/testdb',
  'tablename' = 'testschema.adbpg_sink_table',  
  'username' = 'account****',
  'password' = 'password****',
  'maxRetryTimes' = '2',
  'batchsize' = '5000',
  'conflictMode' = 'ignore',
  'writeMode' = 'insert',
  'retryWaitTime' = '200'
);

-- datagen_source表和source_adbpg源表关联后的结果写入adbpg结果表
INSERT INTO sink_adbpg
SELECT ts.id,ts.username,ds.score FROM datagen_source AS ds
JOIN source_adbpg AS ts ON ds.id = ts.id;

参数说明

在作业开发页面顶部，单击深度检查，进行语法检查。

单击部署，单击确定。

单击右上角前往运维，在作业运维页面，单击启动。

在测试数据库中执行如下语句，查看Flink写入的数据。

SELECT * FROM testschema.adbpg_sink_table;
SELECT COUNT(*) FROM testschema.adbpg_sink_table; 

在源表中新增50条数据，并在结果表中查看Flink写入的增量数据的总数。

-- 源表写入50条增量数据
INSERT INTO testschema.adbpg_source_table(id, username)
SELECT i, 'username'||i::text
FROM generate_series(51, 100) AS t(i);

-- 检查目标表新增数据
SELECT COUNT(*) FROM testschema.adbpg_sink_table where id > 50;

结果如下。

 count 
-------
    50
(1 row)