實驗環境

• Debezium 版本 1.9 (2022-04-05)
• Debezium Tested Versions
  
• PostgreSQL 版本是單機的 14.2
• 本測試參考文檔：https://debezium.io/documentation/reference/1.9/
• 基于 Debezium 的變更數據捕獲的架構:
  
• https://github.com/debezium/docker-images

啟動 Zookeeper

# 后臺運行
docker run -d --name zookeeper -p 2181:2181 -p 2888:2888 -p 3888:3888 quay.io/debezium/zookeeper

# 實時查看 zookeeper 的日志信息
docker logs -f -t --tail 10 zookeeper

啟動 Kafka

# 后臺運行
docker run -d --name kafka -p 9092:9092 --link zookeeper:zookeeper quay.io/debezium/kafka

# 實時查看 kafka 的日志信息
docker logs -f -t --tail 10 kafka

啟動 PostgreSQL 14.2，使用 debezium 提供的示例鏡像，里面自帶了一個測試的 schemas inventory

# 創建一個數據持久化目錄
mkdir -p /docker_data/postgres
chmod -R a+rwx /docker_data/postgres/
# 后臺運行 14.2 版本的 PostgreSQL 數據庫
docker run -d --name postgres \
 -p 5432:5432 \
 -e POSTGRES_PASSWORD=postgres \
 -e PGDATA=/var/lib/pgdata \
 -v /docker_data/postgres:/var/lib/pgdata \
 quay.io/debezium/example-postgres

# 運行 psql 容器
[root@docker ~]# alias psql='docker run -it --rm --name psql debezium/example-postgres psql -h 192.168.0.40 -U postgres -p 5432'
[root@docker ~]# psql
Password for user postgres:
psql (14.2 (Debian 14.2-1.pgdg110+1))
Type "help" for help.

postgres=# select version();
                                                           version
-----------------------------------------------------------------------------------------------------------------------------
 PostgreSQL 14.2 (Debian 14.2-1.pgdg110+1) on x86_64-pc-linux-gnu, compiled by gcc (Debian 10.2.1-6) 10.2.1 20210110, 64-bit
(1 row)

postgres=# \l
                                 List of databases
   Name    |  Owner   | Encoding |  Collate   |   Ctype    |   Access privileges
-----------+----------+----------+------------+------------+-----------------------
 postgres  | postgres | UTF8     | en_US.utf8 | en_US.utf8 |
 template0 | postgres | UTF8     | en_US.utf8 | en_US.utf8 | =c/postgres          +
           |          |          |            |            | postgres=CTc/postgres
 template1 | postgres | UTF8     | en_US.utf8 | en_US.utf8 | =c/postgres          +
           |          |          |            |            | postgres=CTc/postgres
(3 rows)

postgres=# \dn
   List of schemas
   Name    |  Owner
-----------+----------
 inventory | postgres
 public    | postgres
(2 rows)

postgres=# \dt inventory.*
                List of relations
  Schema   |       Name       | Type  |  Owner
-----------+------------------+-------+----------
 inventory | customers        | table | postgres
 inventory | geom             | table | postgres
 inventory | orders           | table | postgres
 inventory | products         | table | postgres
 inventory | products_on_hand | table | postgres
 inventory | spatial_ref_sys  | table | postgres
(6 rows)

postgres=# select schemaname,relname,n_live_tup from pg_stat_user_tables;
 schemaname |     relname      | n_live_tup
------------+------------------+------------
 inventory  | customers        |          4
 inventory  | products_on_hand |          9
 inventory  | orders           |          4
 inventory  | products         |          9
 inventory  | spatial_ref_sys  |       8500
 inventory  | geom             |          3
(6 rows)

看看這個 debezium 提供的 PostgreSQL 鏡像中都做了哪些配置

• pg_hba.conf

[root@docker ~]# cd /docker_data/postgres/
[root@docker postgres]# cat pg_hba.conf
host all all all scram-sha-256
host replication postgres 0.0.0.0/0 trust

• postgresql.conf

listen_addresses = '*'
shared_preload_libraries = 'decoderbufs,wal2json'
wal_level = logical
max_wal_senders = 4
#wal_keep_segments = 4
#wal_sender_timeout = 60s
max_replication_slots = 4

啟動 Kafka Connect

# 后臺運行
docker run -d --name connect \
-p 8083:8083 \
-e GROUP_ID=1 \
-e CONFIG_STORAGE_TOPIC=my_connect_configs \
-e OFFSET_STORAGE_TOPIC=my_connect_offsets \
-e STATUS_STORAGE_TOPIC=my_connect_statuses \
--link zookeeper:zookeeper \
--link kafka:kafka \
--link postgres:postgres \
quay.io/debezium/connect

# 實時查看 Kafka Connect 的日志信息
docker logs -f -t --tail 10 connect 

Debezium PostgreSQL connector

• 準備 Debezium PostgreSQL connector 配置文件
  將配置文件創建在 docker 宿主機上即可，connect 容器開放了 REST API 來管理 Debezium 的連接器

[root@docker ~]# vi pgsql-inventory-connector.json

{
    "name": "pgsql-inventory-connector",
    "config": {
	    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
	    "database.hostname": "postgres",
	    "database.port": "5432",
	    "database.user": "postgres",
	    "database.password": "postgres",
	    "database.dbname": "postgres",
	    "database.server.name": "pgsql",
	    "slot.name": "inventory_slot",
	    "table.include.list": "inventory.orders,inventory.products",
	    "publication.name": "dbz_inventory_connector",
	    "publication.autocreate.mode": "filtered",
	    "plugin.name": "pgoutput"
    }
}

• 向 Kafka 連接器注冊 Debezium PostgreSQL connector

[root@docker ~]# curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" 192.168.0.40:8083/connectors/ -d @pgsql-inventory-connector.json

HTTP/1.1 201 Created
Date: Fri, 22 Apr 2022 00:08:47 GMT
Location: http://192.168.0.40:8083/connectors/pgsql-inventory-connector
Content-Type: application/json
Content-Length: 551
Server: Jetty(9.4.43.v20210629)

{"name":"pgsql-inventory-connector","config":{"connector.class":"io.debezium.connector.postgresql.PostgresConnector","database.hostname":"postgres","database.port":"5432","database.user":"postgres","database.password":"postgres","database.dbname":"postgres","database.server.name":"pgsql","slot.name":"inventory_slot","table.include.list":"inventory.orders,inventory.products","publication.name":"dbz_inventory_connector","publication.autocreate.mode":"filtered","plugin.name":"pgoutput","name":"pgsql-inventory-connector"},"tasks":[],"type":"source"}

使用 kafka-ui 核對捕獲到的數據

kafka-ui：Open-Source Web GUI for Apache Kafka Management：https://github.com/provectus/kafka-ui

docker run -p 8811:8080 \
 -e KAFKA_CLUSTERS_0_NAME=oracle-scott-connector \
 -e KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=192.168.0.40:9092 \
 -d provectuslabs/kafka-ui:latest 

網頁登錄：http://192.168.0.40:8811/

模擬業務

• INSERT

postgres=# select * from inventory.orders;
  id   | order_date | purchaser | quantity | product_id
-------+------------+-----------+----------+------------
 10001 | 2016-01-16 |      1001 |        1 |        102
 10002 | 2016-01-17 |      1002 |        2 |        105
 10003 | 2016-02-19 |      1002 |        2 |        106
 10004 | 2016-02-21 |      1003 |        1 |        107
(4 rows)

postgres=# insert into inventory.orders values (11001,now(),1003,1,102);
INSERT 0 1

postgres=# select * from inventory.orders;
  id   | order_date | purchaser | quantity | product_id
-------+------------+-----------+----------+------------
 10001 | 2016-01-16 |      1001 |        1 |        102
 10002 | 2016-01-17 |      1002 |        2 |        105
 10003 | 2016-02-19 |      1002 |        2 |        106
 10004 | 2016-02-21 |      1003 |        1 |        107
 11001 | 2022-04-22 |      1003 |        1 |        102
(5 rows)

• UPDATE

postgres=# select * from inventory.orders;
  id   | order_date | purchaser | quantity | product_id
-------+------------+-----------+----------+------------
 10001 | 2016-01-16 |      1001 |        1 |        102
 10002 | 2016-01-17 |      1002 |        2 |        105
 10003 | 2016-02-19 |      1002 |        2 |        106
 10004 | 2016-02-21 |      1003 |        1 |        107
 11001 | 2022-04-22 |      1003 |        1 |        102
(5 rows)

postgres=# update inventory.orders set quantity=2 where id=11001;
UPDATE 1

postgres=# select * from inventory.orders;
  id   | order_date | purchaser | quantity | product_id
-------+------------+-----------+----------+------------
 10001 | 2016-01-16 |      1001 |        1 |        102
 10002 | 2016-01-17 |      1002 |        2 |        105
 10003 | 2016-02-19 |      1002 |        2 |        106
 10004 | 2016-02-21 |      1003 |        1 |        107
 11001 | 2022-04-22 |      1003 |        2 |        102
(5 rows)

• DELETE

postgres=# select * from inventory.orders;
  id   | order_date | purchaser | quantity | product_id
-------+------------+-----------+----------+------------
 10001 | 2016-01-16 |      1001 |        1 |        102
 10002 | 2016-01-17 |      1002 |        2 |        105
 10003 | 2016-02-19 |      1002 |        2 |        106
 10004 | 2016-02-21 |      1003 |        1 |        107
 11001 | 2022-04-22 |      1003 |        2 |        102
(5 rows)

postgres=# delete from inventory.orders where id = 11001;
DELETE 1

postgres=# select * from inventory.orders;
  id   | order_date | purchaser | quantity | product_id
-------+------------+-----------+----------+------------
 10001 | 2016-01-16 |      1001 |        1 |        102
 10002 | 2016-01-17 |      1002 |        2 |        105
 10003 | 2016-02-19 |      1002 |        2 |        106
 10004 | 2016-02-21 |      1003 |        1 |        107
(4 rows)

一個問題，“schema.include.list” 捕獲的表不全

當連接器屬性配置 “schema.include.list”: “inventory”，正常來說會捕獲 schema inventory 里面的所有表，但是測試發現少捕獲一張 spatial_ref_sys 表，沒整明白啥情況。

{
    "name": "snapshot-mode-initial",
    "config": {
        "connector.class" : "io.debezium.connector.postgresql.PostgresConnector",
        "database.hostname" : "postgres",
        "database.port" : "5432",
        "database.user" : "postgres",
        "database.password" : "postgres",
        "database.dbname" : "postgres",
        "database.server.name" : "initial",
        "slot.name": "initial_slot",
        "schema.include.list": "inventory",
        "snapshot.mode": "initial",
        "plugin.name": "pgoutput",
        "publication.name": "dbz_inventory_connector",
        "publication.autocreate.mode": "filtered"
    }
}

監控 PostgreSQL 的復制槽

postgres=# select * from pg_replication_slots;