Elasticsearch 如何与 MySQL 集成（数据同步）？
MySQL 数据同步到 Elasticsearch 的四种方案：Logstash JDBC 插件定时拉取、监听 binlog（Canal/Debezium）实时同步、应用双写、以及 CDC 平台。本文对比各方案并给出 Logstash 配置示例。
主流方案是两类：定时轮询——Logstash 的 jdbc input 按更新时间增量拉取；实时同步——Debezium/Canal 捕获 binlog，再由适配的消费者/连接器写入 ES。小表低频用前者，大表实时性要求高用后者。
方案一：Logstash JDBC 定时拉取
input {
jdbc {
jdbc_connection_string => "jdbc:mysql://mysql:3306/shop"
jdbc_user => "reader"
jdbc_password => "${MYSQL_READER_PASSWORD}"
jdbc_driver_library => "/opt/mysql-connector-j.jar"
jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
use_column_value => true
last_run_metadata_path => "/var/lib/logstash/products-jdbc.yml"
statement => "SELECT * FROM products WHERE updated_at >= :sql_last_value ORDER BY updated_at, id"
schedule => "*/1 * * * *"
tracking_column => "updated_at"
tracking_column_type => "timestamp"
}
}
output {
elasticsearch {
hosts => ["es:9200"]
index => "products"
document_id => "%{id}" # 用主键当文档 ID，重复同步变更新
}
}
每个 JDBC input 使用独立、持久且可写的 checkpoint 路径。相同时间戳用 >= 回放边界可减少遗漏，但不能捕获所有迟提交/回填时间的更新；需重叠窗口与对账，或选择 CDC。
关键技巧：sql_last_value 记录上次同步位点实现增量；document_id 用主键让重复写入变成更新而不是重复文档，但迟到的旧事件仍可能覆盖新值，需要顺序、版本控制或对账。
方案二：binlog 实时同步
Debezium（Kafka Connect）或 Canal 监听 MySQL binlog，再通过适配 sink/消费者处理主键、更新与删除。实际延迟取决于链路负载和积压，但架构复杂度高（多一套 CDC 组件）。
方案三：应用双写
业务代码写完 MySQL 顺手写 ES。简单但耦合——ES 挂了影响主流程，仅重试不能保证事务一致性，需幂等、顺序与失败补偿；可靠场景可用同一事务写 outbox 再异步投递。
选型建议
| 场景 | 方案 |
|---|---|
| 报表/搜索，分钟级延迟可接受 | Logstash JDBC |
| 搜索要求准实时 | binlog CDC |
| 极小项目 | 应用双写 |
常见问题（FAQ）
Q：删除的数据怎么同步？ 轮询方案感知不到物理删除——业务上改软删除（加 deleted 标记），或 CDC 可捕获 DELETE，但下游仍需把删除事件映射为 ES 删除；软删除标记也需明确处理策略。
Q：全量初始化怎么做？ 全量快照与增量位点必须协调，避免导入期间的更新/删除遗漏；可使用 CDC 一致快照流程，轮询方案则需建立可靠水位、回放和对账，不能简单去条件再切换。
