搭建PostgreSQL统一数据库，咨询多子库数据归集高效方案

一、基于PostgreSQL原生特性的方案 #

1. 外部数据包装器（FDW）+ pg_cron 实时查询+定时增量同步 #

• 先在统一库中创建外部表关联子库：

-- 启用FDW扩展
CREATE EXTENSION IF NOT EXISTS postgres_fdw;

-- 连接DB1
CREATE SERVER db1_server FOREIGN DATA WRAPPER postgres_fdw OPTIONS (host 'db1_host', dbname 'db1', port '5432');
CREATE USER MAPPING FOR current_user SERVER db1_server OPTIONS (user 'db1_user', password 'db1_pwd');

-- 映射DB1的messages表为外部表
CREATE FOREIGN TABLE db1_messages (
    message_id INT,
    message TEXT,
    created_date TIMESTAMP
) SERVER db1_server OPTIONS (schema_name 'public', table_name 'messages');

-- 同理创建DB2的外部表db2_messages

• 实时跨库查询（直接生成目标格式数据）：

SELECT 'DB1-' || message_id AS message_id, message, created_date FROM db1_messages
UNION ALL
SELECT 'DB2-' || message_id AS message_id, message, created_date FROM db2_messages;

• 用pg_cron做定时增量同步（基于created_date做增量标识）：

-- 启用pg_cron扩展
CREATE EXTENSION IF NOT EXISTS pg_cron;

-- 每天凌晨同步DB1新增数据到统一库
SELECT cron.schedule('sync-db1-to-db3', '0 0 * * *', $$
INSERT INTO db3.messages (message_id, message, created_date)
SELECT 'DB1-' || message_id, message, created_date
FROM db1_messages
WHERE created_date > (SELECT COALESCE(MAX(created_date), '1970-01-01') FROM db3.messages WHERE message_id LIKE 'DB1-%');
$$);

-- 同理添加DB2的同步任务

• 优势：无需额外中间件，原生能力支撑，增量同步效率高，实时查询灵活。

2. pg_dump+psql 全量+增量批量迁移 #

• 全量迁移（修改主键标识后导入）：

# 导出DB1的messages表数据（仅导出数据）
pg_dump -h db1_host -U db1_user -d db1 -t messages --data-only > db1_messages.sql

# 批量修改导出SQL中的message_id，添加DB1前缀
sed -i 's/^\(INSERT INTO messages.*VALUES (\)\([0-9]*\)/\1'\''DB1-\2'\''/' db1_messages.sql

# 导入到统一库DB3
psql -h db3_host -U db3_user -d db3 -f db1_messages.sql

• 增量迁移（基于时间戳过滤数据）：

# 导出DB1中上次同步时间之后的新增数据
pg_dump -h db1_host -U db1_user -d db1 -t messages --data-only --where "created_date > '2024-01-30 01:00:00'" > db1_messages_incremental.sql

# 修改主键后导入DB3
sed -i 's/^\(INSERT INTO messages.*VALUES (\)\([0-9]*\)/\1'\''DB1-\2'\''/' db1_messages_incremental.sql
psql -h db3_host -U db3_user -d db3 -f db1_messages_incremental.sql

• 优势：操作简单，适合一次性全量迁移或数据量较小的场景。

二、基于ETL/CDC工具的方案 #

1. Apache Airflow 调度式同步 #

• 编写Python任务实现增量同步，利用Airflow做定时调度和状态监控：

import psycopg2
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def sync_db1_to_db3():
    # 连接子库DB1
    conn_db1 = psycopg2.connect(host='db1_host', dbname='db1', user='db1_user', password='db1_pwd')
    cur_db1 = conn_db1.cursor()
    # 连接统一库DB3
    conn_db3 = psycopg2.connect(host='db3_host', dbname='db3', user='db3_user', password='db3_pwd')
    cur_db3 = conn_db3.cursor()
    
    # 获取DB3中DB1数据的最后同步时间
    cur_db3.execute("SELECT COALESCE(MAX(created_date), '1970-01-01') FROM messages WHERE message_id LIKE 'DB1-%'")
    last_sync_time = cur_db3.fetchone()[0]
    
    # 查询DB1新增数据
    cur_db1.execute("SELECT message_id, message, created_date FROM messages WHERE created_date > %s", (last_sync_time,))
    rows = cur_db1.fetchall()
    
    # 插入DB3并修改主键
    for row in rows:
        cur_db3.execute("INSERT INTO messages (message_id, message, created_date) VALUES (%s, %s, %s)", 
                        (f'DB1-{row[0]}', row[1], row[2]))
    
    conn_db3.commit()
    # 关闭连接
    cur_db1.close()
    conn_db1.close()
    cur_db3.close()
    conn_db3.close()

# 定义调度DAG，每小时执行一次
with DAG('sync_postgres_dbs', start_date=datetime(2024,1,30), schedule_interval='@hourly') as dag:
    sync_db1_task = PythonOperator(task_id='sync_db1_to_db3', python_callable=sync_db1_to_db3)
    # 同理添加DB2的同步任务

• 优势：调度灵活，支持复杂依赖关系，适合多库多表的大规模同步场景，可监控任务运行状态。

2. Debezium CDC 实时变更同步 #

• 配置Debezium监听子库的数据变更，实时同步到统一库，同步时自动添加库标识前缀：
• 核心连接器配置（以DB1为例）：

name=db1-connector
connector.class=io.debezium.connector.postgresql.PostgresConnector
database.hostname=db1_host
database.port=5432
database.dbname=db1
database.user=db1_user
database.password=db1_pwd
database.server.name=db1
table.include.list=public.messages
transforms=addDbPrefix
transforms.addDbPrefix.type=org.apache.kafka.connect.transforms.ReplaceField$Value
transforms.addDbPrefix.renames=message_id:message_id
transforms.addDbPrefix.prefix=DB1-

• 优势：实时性强，可捕获所有数据变更（插入、更新、删除），适合对数据实时性要求高的报表/仪表盘场景。

三、方案选择建议 #

• 追求低成本、无额外中间件：优先选FDW+pg_cron方案；
• 一次性全量迁移：选pg_dump+psql方案；
• 多库多表、需调度监控：选Apache Airflow方案；
• 要求实时数据同步：选Debezium CDC方案。

内容的提问来源于stack exchange，提问作者Sathish Kumar