現(xiàn)有需求，需要使用pg的數(shù)據(jù)實時統(tǒng)計一些指標(biāo)，經(jīng)過調(diào)研，決定使用kafkaCat或debezium將pg的操作日志同步到kafka中，現(xiàn)將簡單測試debezium的過程總結(jié)一下。
debezium簡介
1.debezium默認(rèn)行為是連接器將所有生成的事件流式傳輸?shù)皆摫淼膯为?dú)kafka主題。
2.PG需要安裝一個邏輯解碼輸出插件，插件可以是decoderbufs、wal2json、pgoutput，我使用的是wal2json
3.連接器為捕獲的每個行級插入、更新和刪除操作生成一個更改事件，并未單獨(dú)的kafka主題中的每個表發(fā)送更改事件記錄
4.邏輯解碼不支持DDL更改，連接器無法將DDL更改事件報告回消費(fèi)者
5.只有primary的表支持邏輯解碼復(fù)制槽
6.目前僅支持使用UTF-8字符編碼的數(shù)據(jù)庫。使用單字節(jié)字符編碼，無法正確處理包含擴(kuò)展ASCII代碼字符的字符串
安裝wal2json插件
1.配置環(huán)境變量
#不是必須是~/.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參數(shù)
因為postgresql.conf在data目錄下，正常是沒有訪問權(quán)限的，需要先使用管理員權(quán)限修改權(quán)限
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參數(shù)，運(yùn)行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目錄權(quán)限修改為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.啟動消費(fèi)者，查看topic是否有數(shù)據(jù)
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test.public.test_table
4.在pg的test庫下面對test_table進(jìn)行操作
4.1 test_table表結(jié)構(gòu)(必須要設(shè)置主鍵)
create table test_table( id varchar, node varchar, primary key (id) );
4.1 Insert操作
insert into test_table(id,code) values(3,'c');
消費(fèi)者消費(fèi)到信息：
{"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';
消費(fèi)者消費(fèi)到信息：
{"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';
消費(fèi)者消費(fèi)到信息：如果是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操作不會產(chǎn)生Topic數(shù)據(jù)
參考：
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
