PostgreSQL到Greenplum的逻辑复制及Debezium配置方法问询

一、Debezium配置实现同步步骤 #

你的场景只有INSERT操作，配置可以简化不少，具体步骤如下：

1. PostgreSQL端前置配置 #

• 修改postgresql.conf开启逻辑复制：

  wal_level = logical
  max_replication_slots = 5  # 至少大于1，按需调整
  

  重启PostgreSQL服务生效。
• 创建具备复制权限的专用用户：

  CREATE ROLE debezium REPLICATION LOGIN PASSWORD 'your_secure_password';
  GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium;
  

• 创建逻辑复制槽：

  SELECT pg_create_logical_replication_slot('debezium_gp_slot', 'pgoutput');
  

2. Debezium连接器（Kafka Connect）配置 #

Debezium依赖Kafka Connect运行，先确保Kafka、Kafka Connect部署完成，且Debezium PostgreSQL连接器插件已安装到Connect的插件目录。然后创建连接器配置文件（比如postgres-source.json）：

{
  "name": "postgres-gp-source",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "2",
    "database.hostname": "你的PostgreSQL地址",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "your_secure_password",
    "database.dbname": "目标数据库名",
    "database.server.name": "postgres_source",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_gp_slot",
    "table.include.list": "public.需要同步的表名1,public.需要同步的表名2",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "true",  // 仅INSERT场景，墓碑记录直接丢弃
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter.schemas.enable": "false"
  }
}

提交配置到Kafka Connect：

curl -X POST -H "Content-Type: application/json" --data @postgres-source.json http://你的KafkaConnect地址:8083/connectors

3. Greenplum端Sink连接器配置 #

用Kafka Connect的JDBC Sink连接器将Kafka中的数据写入Greenplum，创建配置文件（比如gp-sink.json）：

{
  "name": "gp-sink-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "2",
    "connection.url": "jdbc:postgresql://你的Greenplum地址:5432/GP数据库名",
    "connection.user": "GP用户名",
    "connection.password": "GP密码",
    "topics": "postgres_source.public.需要同步的表名1,postgres_source.public.需要同步的表名2",
    "auto.create": "false",  // 建议提前在Greenplum建好结构匹配的表，避免自动创建的结构不符合预期
    "insert.mode": "insert",  // 仅插入场景，用insert模式即可
    "batch.size": "2000",  // 批量插入大小，根据数据量调整
    "table.name.format": "public.目标表名",  // 映射到Greenplum的具体表
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter.schemas.enable": "false"
  }
}

同样提交到Kafka Connect：

curl -X POST -H "Content-Type: application/json" --data @gp-sink.json http://你的KafkaConnect地址:8083/connectors

二、其他近实时数据复制方案 #

1. PostgreSQL逻辑复制直接同步到Greenplum #

Greenplum 6.2及以上版本支持PostgreSQL的逻辑复制协议，无需中间件，直接点对点同步：

• PostgreSQL端创建发布：

  CREATE PUBLICATION gp_publication FOR TABLE public.需要同步的表名;
  

• Greenplum端创建订阅：

  CREATE SUBSCRIPTION gp_subscription 
  CONNECTION 'host=PostgreSQL地址 port=5432 dbname=源库名 user=debezium password=your_secure_password' 
  PUBLICATION gp_publication;
  

  优势：架构极简，延迟低；缺点：故障排查不如带中间件的方案直观，仅支持标准逻辑复制操作（你的场景刚好适配）。

2. Flink CDC直接同步 #

用Flink CDC跳过Kafka，直接捕获PostgreSQL逻辑日志并写入Greenplum，适合不需要消息队列中转的场景：

• 编写Flink作业，使用PostgreSQL CDC Source读取增量数据，通过JDBC Sink批量写入Greenplum，配置合适的并行度和批量参数，可实现秒级延迟。
• 优势：减少组件依赖，端到端延迟更可控；缺点：需要编写Flink代码或使用Flink SQL。

3. 定时增量导出导入脚本 #

如果能接受5-10分钟的延迟，用简单的脚本就能搞定：

• 先通过pg_dump全量同步初始数据到Greenplum
• 定时执行脚本，查询PostgreSQL中最近N分钟新增的数据（依赖表中有创建时间字段），用psql或gpload导入Greenplum
• 优势：零额外组件，运维成本极低；缺点：延迟取决于脚本执行频率，无法做到真正的近实时。

内容的提问来源于stack exchange，提问作者Vladimir Shadrin