如何通过管道将pg_recvlogical输出传给psql实现PostgreSQL逻辑复制？

核心问题分析 #

你遇到的问题本质是pg_recvlogical通过不同解码器输出的内容并非psql可直接执行的标准SQL语句：

• wal2json：输出JSON格式的变更元数据，不是可执行SQL，psql自然会报JSON语法错误
• pgoutput：是PostgreSQL内置逻辑复制的协议格式（供订阅端使用），不是面向psql的文本输出，所以会提示协议版本错误
• text_decoder：仅输出极简的文本化变更描述，不是标准SQL语句，无法被psql解析执行

可行解决方案 #

方案1：使用支持生成可执行SQL的解码器插件 #

AWS Aurora/RDS PostgreSQL支持sql_decoder插件（需确认实例版本兼容性），它可直接输出可执行的SQL语句。操作步骤如下：

1. 删除原复制槽（若不再需要）：

SELECT pg_drop_replication_slot('migration_goldsky');

2. 创建使用sql_decoder的复制槽：

SELECT pg_create_logical_replication_slot('migration_goldsky_sql', 'sql_decoder');

3. 通过Unix管道实时同步变更到目标RDS：

pg_recvlogical -d <源Aurora连接字符串> -S migration_goldsky_sql -f - | psql <目标RDS连接字符串>

方案2：解析wal2json的JSON输出生成可执行SQL #

若sql_decoder不可用，可通过脚本解析wal2json输出的JSON，转换为psql可执行的SQL。以下是基于jq的bash示例（需提前安装jq）：

pg_recvlogical -d <源Aurora连接字符串> -S migration_goldsky -f - -P 'format-version=2' | \
jq -r '
  .change[] |
  if .kind == "insert" then
    "INSERT INTO " + .schema + "." + .table + " (" + (.columns | map(.name) | join(",")) + ") VALUES (" + (.columns | map(.value | @sh) | join(",")) + ");"
  elif .kind == "update" then
    "UPDATE " + .schema + "." + .table + " SET " + (.columns | map(.name + "=" + (.value | @sh)) | join(",")) + " WHERE " + (.identity | map(.name + "=" + (.value | @sh)) | join(" AND ")) + ";"
  elif .kind == "delete" then
    "DELETE FROM " + .schema + "." + .table + " WHERE " + (.identity | map(.name + "=" + (.value | @sh)) | join(" AND ")) + ";"
  else
    empty
  end
' | psql <目标RDS连接字符串>

注：该脚本仅处理基础增删改操作，复杂场景（如DDL、特殊数据类型）需额外适配。

方案3：使用AWS原生CDC工具 #

如果手动处理逻辑复制成本过高，可直接使用AWS DMS（数据库迁移服务）的**变更数据捕获（CDC）**功能，它能自动同步Aurora到RDS的增量变更，无需手动处理复制槽和SQL转换，适合生产环境割接场景。

关于Unix管道的可行性 #

完全可以通过Unix管道实现实时同步，前提是确保pg_recvlogical输出的内容是psql可直接执行的格式（如方案1的原生SQL输出，或方案2转换后的SQL）。管道的优势是无需中间文件，能实现变更的实时同步，非常适合割接期间的数据一致性保障。

内容的提问来源于stack exchange，提问作者Paymahn Moghadasian