前言
最近看文章说如何把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怎么消费，里面的记录怎么显示出来，它里面实现的原理是什么，都需要花时间去研究，先开个头在这里。
评论(0)