PostgreSQL 如何同步数据到 Elasticsearch？
PostgreSQL 同步到 Elasticsearch 的方案：Logstash JDBC 定时轮询、逻辑解码 + Debezium CDC，或应用双写。本文对比方案并给出 Logstash 配置。
两条主路：定时轮询用 Logstash 的 jdbc input（按更新时间增量拉）；准实时用 Debezium 读 PG 的 WAL 逻辑解码流式推送。选哪个看实时性要求。
方案一：Logstash JDBC 轮询
input {
jdbc {
jdbc_connection_string => "jdbc:postgresql://pg:5432/shop"
jdbc_user => "reader"
jdbc_password => "${PG_READER_PASSWORD}"
jdbc_driver_library => "/opt/postgresql.jar"
jdbc_driver_class => "org.postgresql.Driver"
last_run_metadata_path => "/var/lib/logstash/products-jdbc-last-run.yml"
statement => "SELECT * FROM products WHERE updated_at >= :sql_last_value ORDER BY updated_at ASC, id ASC"
schedule => "*/2 * * * *"
tracking_column => "updated_at"
tracking_column_type => "timestamp"
use_column_value => true
}
}
output {
elasticsearch {
hosts => ["https://es:9200"]
user => "${ES_USER}"
password => "${ES_PASSWORD}"
ssl_certificate_authorities => ["/etc/logstash/certs/ca.crt"]
index => "products"
document_id => "%{id}"
}
}
document_id 使用主键可避免重放产生新ID，但旧事件仍可能覆盖新值。示例用 >= 和排序减少同时间戳边界遗漏，仍不能彻底解决延迟提交、时间精度及并发分页问题；需重叠回查、对账或CDC。状态文件必须持久化且每个input独立，目录可写；凭据应来自Logstash keystore等受控来源。
方案二：Debezium 逻辑解码（准实时）
PG 开逻辑复制（wal_level = logical），Debezium PG 连接器读 WAL 变更事件，经 Kafka 写入 ES。延迟取决于负载和配置；连接器发出删除事件，仍需配置兼容 sink 将其应用到 ES。通常还要维护 Kafka/Connect 和复制槽、publication、权限及WAL保留。
方案对比
| 方案 | 延迟 | 删除同步 | 复杂度 |
|---|---|---|---|
| JDBC 轮询 | 分钟级 | ❌（需软删除） | 低 |
| Debezium CDC | 依负载/配置，可接近实时 | 需sink正确处理删除 | 高 |
| 应用双写 | 依实现 | 需自理 | 需处理部分失败、一致性和重试 |
常见问题（FAQ）
Q：没有 updated_at 列怎么办？ 加列并建触发器维护，或全量定期重建索引。
Q：大表首次全量同步很慢？ 分批（按主键范围）并行跑，导入期调大 ES 的 refresh_interval。
