如何实时将Elasticsearch数据存入RDB？支持Logstash或ES管道吗？

Logstash 可完整实现 Elasticsearch 到 RDB 表的实时写入需求，原生 Elasticsearch Pipeline 不支持该场景，不用在反方向同步方案里浪费时间踩坑。

为什么 Elasticsearch Pipeline 不适用 #

• 不管是 Ingest Pipeline 还是 Index Pipeline，运行载体都是 ES 集群节点本身，内置处理器仅支持对写入 ES 链路内的文档做字段转换、过滤、路由操作，没有原生 RDB 输出能力。如果硬要通过自定义插件实现输出到库，开发和维护成本极高，生产环境完全不推荐。

Logstash 落地实操方案 #

这是目前该场景下最成熟的落地路径，核心逻辑是通过 Logstash 的 Elasticsearch 输入插件监听索引变更，再通过 JDBC 输出插件写入目标 RDB 表，配置和注意事项如下：

• 前置准备：如果是带Xpack的商业版ES，可直接开启索引变更订阅实现毫秒级同步；如果是开源版ES，通过时间戳游标轮询的方式也能做到秒级准实时，满足绝大多数业务需求。
• 核心配置参考，直接改对应参数就能用：

input {
  elasticsearch {
    hosts => ["http://ES节点IP:9200"]
    index => "待同步的ES索引名"
    schedule => { "every" => "1s" }
    # 用文档更新时间做游标，避免重复拉取全量数据
    tracking_column => "doc_update_time"
    tracking_column_type => "timestamp"
    user => "ES访问账号"
    password => "ES访问密码"
  }
}
filter {
  # 做字段映射、类型转换，对齐ES文档和RDB表的字段
  mutate {
    # 去掉ES自带的冗余元字段，避免写入时报错
    remove_field => ["@version", "_id", "_index", "_score", "_type"]
  }
}
output {
  jdbc {
    connection_string => "jdbc:数据库类型://数据库IP:端口/库名?useUnicode=true&characterEncoding=utf8"
    username => "数据库账号"
    password => "数据库密码"
    # 提前将对应数据库的JDBC驱动包放到Logstash的drivers目录，填对路径
    driver_jar_path => "/opt/logstash/drivers/mysql-connector-java-8.0.30.jar"
    # 用upsert逻辑写入，按主键冲突更新，避免重复数据
    statement => ["INSERT INTO target_table (col1, col2, update_time) VALUES (?,?,?) ON DUPLICATE KEY UPDATE col1=VALUES(col1), col2=VALUES(col2), update_time=VALUES(update_time)", "col1_val", "col2_val", "doc_update_time"]
  }
}

• 生产环境踩坑提示：
  • 绝对不要不带游标做全量拉取，高频全量扫描会直接打满ES集群资源
  • 目标RDB表必须提前设置主键，配合upsert逻辑才能保证数据一致性
  • 单表同步数据量超过千万级时，适当调大Logstash的worker数量、批量写入batch_size到1000-5000区间，同步性能会有明显提升

如果不想额外维护Logstash服务，小数据量场景也可以通过监听ES索引的变更钩子，写轻量消费脚本写入RDB，但稳定性、异常重试、断点续传能力都不如Logstash，生产环境优先选Logstash方案即可。

内容的提问来源于stack exchange，提问作者WooSub Shin