现有需求，需要使用pg的数据实时统计一些指标，经过调研，决定使用kafkaCat或debezium将pg的操作日志同步到kafka中，现将简单测试debezium的过程总结一下。
debezium简介
1.debezium默认行为是连接器将所有生成的事件流式传输到该表的单独kafka主题。
2.PG需要安装一个逻辑解码输出插件，插件可以是decoderbufs、wal2json、pgoutput，我使用的是wal2json
3.连接器为捕获的每个行级插入、更新和删除操作生成一个更改事件，并未单独的kafka主题中的每个表发送更改事件记录
4.逻辑解码不支持DDL更改，连接器无法将DDL更改事件报告回消费者
5.只有primary的表支持逻辑解码复制槽
6.目前仅支持使用UTF-8字符编码的数据库。使用单字节字符编码，无法正确处理包含扩展ASCII代码字符的字符串
安装wal2json插件
1.配置环境变量
#不是必须是~/.bash_profile ，也可以是与其他组件相同的profile文件
vim ~/.bash_profile
export PATH=/Library/PostgreSQL/10/bin:$PATH
export PGDATA=//Library/PostgreSQL/10/data
export PATH=/Library/PostgreSQL/10/bin/pg_config:$PATH
#退出
source ～/.bash_profile
2.clone wal2json项目并编译安装
git clone https://github.com/streamsets/wal2json.git
export PGDATA=//Library/PostgreSQL/10/data
make
make install
3.修改postgresql.conf参数
因为postgresql.conf在data目录下，正常是没有访问权限的，需要先使用管理员权限修改权限
sudo chmod 755 data
#postgresql.conf
wal_level = logical
#
# these parameters only need to set in versions 9.4, 9.5 and 9.6
# default values are ok in version 10 or later
#
max_replication_slots = 10
max_wal_senders = 10
wal_sender_timeout = 2000
如果不修改wal_level参数，运行kafka-connect后会报错
Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask:193)
org.apache.kafka.connect.errors.ConnectException: org.postgresql.util.PSQLException: ERROR: logical decoding requires wal_level >= logical
at io.debezium.connector.postgresql.PostgresConnectorTask.start(PostgresConnectorTask.java:114)
at io.debezium.connector.common.BaseSourceTask.start(BaseSourceTask.java:77)
at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:232)
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:186)
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:241)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
4.将wal2json.so拷贝到/Library/PostgreSQL/10/lib下
5.重启pg
./pg_ctl -l /Library/PostgreSQL/10/logs/pg_server.log -D /Library/PostgreSQL/10/data/ restart
在重启前需要将data目录权限修改为700,否则会报错
安装debezium
1.下载安装包
$ wget https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/1.7.1.Final/debezium-connector-postgres-1.7.1.Final-plugin.tar.gz
$ tar -zxvf ./debezium-connector-postgres-1.7.1.Final-plugin.tar.gz
$ cd debezium-connector-postgres/
$ cp ./*.jar /Users/yanhaowen/bigdata_tools/kafka/libs
$ cp -R ../debezium-connector-postgres /Users/yanhaowen/bigdata_tools/kafka/plugins
2.修改connect standalone
配置postgres.properties
#postgres.properties
name=dbz-pg-connector-source
connector.class=io.debezium.connector.postgresql.PostgresConnector
plugin.name=pgoutput
database.hostname=localhost
database.port=5432
database.user=postgres
database.password=123456
database.dbname=test
database.history.kafka.bootstrap.servers=localhost:9092
database.server.name=test
database.whitelist=public.test_table #白名单，指同步哪张表
3.修改connect-standalone.properties/connect-distributed.properties
bootstrap.servers=localhost:9092
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true
offset.storage.file.filename=/tmp/connect.offsets
offset.flush.interval.ms=10000
plugin.path=/Users/yanhaowen/bigdata_tools/kafka/plugins
测试
1.启动kafka-connect
./connect-standalone.sh ../config/connect-standalone.properties ../config/postgres.properties
2.查看kafka的topic
./kafka-topics.sh --list --bootstrap-server localhost:9092
3.启动消费者，查看topic是否有数据
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test.public.test_table
4.在pg的test库下面对test_table进行操作
4.1 test_table表结构(必须要设置主键)
create table test_table( id varchar, node varchar, primary key (id) );
4.1 Insert操作
insert into test_table(id,code) values(3,'c');
消费者消费到信息：
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"string","optional":true,"field":"code"}],"optional":true,"name":"test.public.test_table.Value","field":"before"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"string","optional":true,"field":"code"}],"optional":true,"name":"test.public.test_table.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"}],"optional":false,"name":"test.public.test_table.Envelope"},"payload":{"before":null,"after":{"id":"3 ","code":"c "},"source":{"version":"1.0.3.Final","connector":"postgresql","name":"test","ts_ms":1641449386360,"snapshot":"false","db":"test","schema":"public","table":"test_table","txId":574,"lsn":25157688,"xmin":null},"op":"c","ts_ms":1641449386362}}
4.2 Update操作
update test_table set code='aaa' where id='3';
消费者消费到信息：
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"string","optional":true,"field":"code"}],"optional":true,"name":"test.public.test_table.Value","field":"before"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"string","optional":true,"field":"code"}],"optional":true,"name":"test.public.test_table.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"}],"optional":false,"name":"test.public.test_table.Envelope"},"payload":{"before":null,"after":{"id":"3 ","code":"aa "},"source":{"version":"1.0.3.Final","connector":"postgresql","name":"test","ts_ms":1641450009438,"snapshot":"false","db":"test","schema":"public","table":"test_table","txId":575,"lsn":25158168,"xmin":null},"op":"u","ts_ms":1641450009444}}
4.3 Delete操作
delete from test_table where id='3';
消费者消费到信息：如果是delete操作，则会多输出一行null
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"string","optional":true,"field":"code"}],"optional":true,"name":"test.public.test_table.Value","field":"before"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"string","optional":true,"field":"code"}],"optional":true,"name":"test.public.test_table.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"}],"optional":false,"name":"test.public.test_table.Envelope"},"payload":{"before":{"id":"1","code":null},"after":null,"source":{"version":"1.0.3.Final","connector":"postgresql","name":"test","ts_ms":1641450949231,"snapshot":"false","db":"test","schema":"public","table":"test_table","txId":589,"lsn":25325152,"xmin":null},"op":"d","ts_ms":1641450949232}}
null
4.4 Create 和 Drop操作不会产生Topic数据
参考：
https://ctypyb2002.blog.csdn.net/article/details/121427957?spm=1001.2101.3001.6650.6&utm_medium=distribute.pc_relevant.none-task-blog-2%7Edefault%7EBlogCommendFromBaidu%7Edefault-6.pc_relevant_paycolumn_v2&depth_1-utm_source=distribute.pc_relevant.none-task-blog-2%7Edefault%7EBlogCommendFromBaidu%7Edefault-6.pc_relevant_paycolumn_v2&utm_relevant_index=9
https://blog.csdn.net/weixin_40898246/article/details/108327247
https://debezium.io/documentation/reference/1.2/connectors/postgresql.html#how-the-postgresql-connector-works
https://blog.csdn.net/javahelpyou/article/details/114536386
