配置Logstash同步MongoDB与Elasticsearch：如何自动同步数据更新？

核心逻辑 #

要捕获MongoDB的文档修改、删除等操作，需要让Logstash的mongodb输入插件监听MongoDB的oplog（操作日志）——oplog会记录集群内所有写操作（插入、更新、删除），通过监听它就能实现增量同步和变更捕获。

配置修改步骤 #

1. 调整Input部分的MongoDB插件配置 #

需要开启oplog监听功能，同时注意：MongoDB必须是副本集模式（单节点也可配置为单节点副本集），且Logstash使用的MongoDB账号需具备读取oplog的权限。

修改后的input配置：

input {
  mongodb {
        uri => '<my_mongo_db_uri>'
        placeholder_db_dir => '/opt/logstash-mongodb/'
        placeholder_db_name => '<my_placeholeder>.db'
        collection => '<my_mongo_db_collection>'
        batch_size => 5000
        # 开启oplog监听
        oplog => true
        # 先全量同步现有数据，再自动切换到增量监听；无需全量可设为false
        initial_sync => true
        # oplog默认存储在local库，无需修改
        oplog_db => 'local'
        oplog_collection => 'oplog.rs'
  }
}

2. 新增Filter逻辑处理不同操作类型 #

MongoDB的oplog会标记操作类型（insert/update/delete），需要根据类型给事件添加对应的Elasticsearch动作标签：

filter {
  mutate {
    remove_field => ["_id"]
  }

  # 根据oplog操作类型设置ES动作
  if [operation] == "insert" {
    mutate {
      add_field => { "[@metadata][action]" => "index" }
    }
  } elsif [operation] == "update" {
    mutate {
      add_field => { "[@metadata][action]" => "update" }
      # 用MongoDB文档ID作为ES文档ID，确保更新精准对应
      add_field => { "[@metadata][_id]" => "%{document_id}" }
    }
  } elsif [operation] == "delete" {
    mutate {
      add_field => { "[@metadata][action]" => "delete" }
      add_field => { "[@metadata][_id]" => "%{document_id}" }
    }
  }
}

3. 调整Output部分的Elasticsearch插件配置 #

使用filter中动态生成的动作和文档ID，实现对应操作：

output {
        stdout {
                codec => rubydebug
        }

        elasticsearch {
                action => "%{[@metadata][action]}"  # 动态匹配操作动作
                index => "<my_index>"
                hosts => ["http://<my_elasticsearch_ip>:9200"]
                document_id => "%{[@metadata][_id]}"  # 关联MongoDB与ES的文档ID
                doc_as_upsert => true  # 处理update时文档不存在的情况，自动插入
        }
}

关键注意事项 #

• 副本集要求：单节点MongoDB需先配置为单节点副本集，否则无oplog可用。
• 权限配置：MongoDB账号需拥有目标集合的read权限，以及local库oplog集合的read权限。
• ID一致性：必须保证MongoDB文档ID与Elasticsearch文档ID一致，否则更新/删除操作无法精准匹配目标文档。

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