kafka connect +debezium实时采集postgresql

使用pgoutput的方式设置postgresql修改postgresql.conf添加或者修改以下字段wal_level = logicalmax_wal_senders = 1max_replication_slots = 1修改pg_hba.conflocal replication <youruser> trusthost...

wang972779876

972人浏览 · 2021-09-06 16:28:23

wang972779876 · 2021-09-06 16:28:23 发布

使用pgoutput的方式

设置postgresql

修改postgresql.conf添加或者修改以下字段

wal_level = logical

max_wal_senders = 1

max_replication_slots = 1

修改pg_hba.conf

local replication <youruser> trust

host replication <youruser> 127.0.0.1/32 trust

host replication <youruser> ::1/128 trust

需要权限

表Select权限 GRANT SELECT on public.[table] to [user];

Publication 权限 ALTER USER test_dbz REPLICATION；

表的PUBLICATION 权限 CREATE PUBLICATION [publication_name] FOR TABLE [table1,table2]（需要表的owner或者管理员）

配置kafka connect

1.将kafka connect的jar报放在plugn 目录（config/connect-distributed.properties的plugin.path）目录下

2.重启kafka connect

3.编写connect json

{
  "name": "postgres-connect",  
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector", 
    "database.hostname": "ip", 
    "database.port": "5432", 
    "database.user": "user", 
    "database.password": "pass", 
    "database.dbname" : "test_owner", 
    "database.server.name": "postgres-test-owner", 
    "table.include.list": "public.test1,public.test2,public.test3",
    "publication.autocreate.mode":"filtered",
    "plugin.name":"pgoutput",
  }
}

4.post 提交json 到http://ip:8083/connectors

5.查看kafka topic postgres-test-owner.public.test1