Flink CDC 监听 Postgresql表的变化（）

　　本篇文章为你整理了Flink CDC 监听 Postgresql表的变化（）的详细内容，包含有 Flink CDC 监听 Postgresql表的变化，希望能帮助你了解 Flink CDC 监听 Postgresql表的变化。

　　最近看文章说如何把Postgresql的数据同步给别的数据源，可以利用它的WAL，具体怎么操作没有说，我自己找到一篇文章https:///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:///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();

　　 .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怎么消费，里面的记录怎么显示出来，它里面实现的原理是什么，都需要花时间去研究，先开个头在这里。

　　以上就是Flink CDC 监听 Postgresql表的变化（）的详细内容，想要了解更多 Flink CDC 监听 Postgresql表的变化的内容，请持续关注盛行IT软件开发工作室。

郑重声明：本文由网友发布，不代表盛行IT的观点，版权归原作者所有，仅为传播更多信息之目的，如有侵权请联系，我们将第一时间修改或删除，多谢。

相关文章阅读