PostgreSQL逻辑复制到kafka-实践

kafka 安装

　　wget http://mirror.bit.edu.cn/apache/kafka/2.3.0/kafka_2.12-2.3.0.tgz

? cp kafka_2.12-2.0.1.tgz? kafka.tgz

? sudo tar xzvf kafka.tgz --directory=/opt/java/kafka --strip 1

?

启动 kafka，需要先启动本地的 zookeeper，注意修改配置文件中zk的连接地址

?

? /opt/kafka/bin/kafka-server-start.sh ? /opt/kafka/config/server.properties

?

kafkacat?是一个C语言编写的 kafka 生产者、消费者程序。安装过程需要一些库可能需要手动下载。

postgres 逻辑解码,程序?jsoncdc

jsoncdc 依赖于rust，可能需要先安装 rust 或者可以使用?wal2json替代

编译好之后本地目录下有 jsoncdc.so 或者 wal2json.so

?

postgres 安装解码插件：

vim? $PGDATA/postgresql.conf

shared_preload_libraries = ‘jsoncdc.so‘

安装完成插件

postgres 插入数据

?

生产数据到本地 kafka

/opt/bin/pgsql/pg_10/bin/pg_recvlogical -d postgres -S jsoncdc --start -f - | ./kafkacat/kafkacat -b 127.0.0.1:9092 -t pg

?Auto-selecting Producer mode (use -P or -C to override)

?

消费数据测试:

./kafkacat/kafkacat -b 127.0.0.1:9092 -t pg