Flink CDC 监听 Postgresql表的变化
前言
最近看文章说如何把Postgresql的数据同步给别的数据源,可以利用它的WAL,具体怎么操作没有说,我自己找到一篇文章https://www.cnblogs.com/xiongmozhou/p/14817641.html 可以利用Flink CDC。 我自己正好前段时间也看过Flink,把这个知识串起来也很有意义,于是开始动手试了一下,期间也遇到些困难,也尝试解决了,有些原理不是很清晰,记录下来,后面看能不能解决。
Postgresql配置
我们使用上篇文章搭建的Postgresql数据库,要让Postgresql支持同步给其它数据源,一个最关键的配置是更改wal日志方式为logical, 这个配置在postgresql.conf, 而我们docker里面的postgresql.conf这个配置又在哪个目录呢? 网上找到了答案:https://stackoverflow.com/questions/30848670/how-to-customize-the-configuration-file-of-the-official-postgresql-docker-image
进入psql后,使用如下命令
SHOW config_file;
得到如下的结果
/var/lib/postgresql/data/postgresql.conf
得到路径后, 我打算像平时一样用vi去修改,发现不行,这个postgresql的Image并没有安装vim。
如何修改呢,继续网上找答案 https://stackoverflow.com/questions/30848670/how-to-customize-the-configuration-file-of-the-official-postgresql-docker-image
方法很多,我们用个简单的,使用sed命令来修改
sed -i -e"s/^#wal_level = replica.*$/wal_level = logical/" /var/lib/postgresql/data/postgresql.conf
就是查找到“#wal_level = replica“,把它替换为“wal_level = logical”
修改后需要重启postgresql,执行如下命令
su - postgres -c "PGDATA=$PGDATA /usr/lib/postgresql/15/bin/pg_ctl -w restart"
执行后会退出docker,需要重新进入
新建用户和授予权限参考https://www.cnblogs.com/xiongmozhou/p/14817641.html
注意文档中使用CREATE USER user它建的用户是user,我用的这个用户名是不成功的,提示语法错误
感觉是把user当作保留命令参数了,用户名改为user1可以成功。
使用flink-connector-postgres-cdc
我们参考官方文档https://ververica.github.io/flink-cdc-connectors/master/content/connectors/postgres-cdc.html#connector-options
首先在已有的Flink项目中加入如下的pom
<dependency>
<groupId>com.ververica</groupId>
<artifactId>flink-connector-postgres-cdc</artifactId>
<version>2.3.0</version>
<scope>provided</scope>
</dependency>
这里代码参考文档
SourceFunction postgreSQLSource = PostgreSQLSource.<String>builder()
.hostname("localhost")
.port(5432)
.database("postgres") // set captured database
.tableList("postgres.market_price") // set captured table
.username("user1")
.password("pwd")
.decodingPluginName("pgoutput")
.deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
.build();
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env
.addSource(postgreSQLSource)
.print().setParallelism(1); // use parallelism 1 for sink to keep message ordering
env.execute("Print Postgres Snapshot + WAL");
有一点需要注意,官方文档中没有.decodingPluginName(“pgoutput”),使用默认的decoderbufs,运行程序会提示
“PSQLException: ERROR: could not access file “decoderbufs”: No such file or directory”, 修改成pgoutput,才能成功。 这里应该是要安装插件decoderbufs在Postgresql里面。这里暂时留下这个疑问,后面还有wal2json,看怎么把wal的值转成json格式显示出来。
程序运行起来后我们往表里插入和删除数据,可以在控制台中打印出变化来。
这里直接贴图
这里也有个疑问,我对表操作了三次,结果控制台打印出超过3条的信息,这里应该和是否commit有关
暂时也没有细究。
程序运行后,我们可以使用这个命令查看这个slot,
SELECT * FROM pg_replication_slots;
如果我们直接修改配置,比如把pgoutput改为别的,会提示slot flink已经存在,我们需要在postgresql里面把它先删除掉。
总结
总体上这个流程是打通了,但是对于里面的细节没有深入,比如flink怎么消费,里面的记录怎么显示出来,它里面实现的原理是什么,都需要花时间去研究,先开个头在这里。