如何让Logstash输出至MongoDB并实现Elasticsearch读取该库数据？

Elasticsearch本身无法直接读取MongoDB数据，需要借助中间工具来实现同步。结合你已经在用Logstash的现状，以下两种方案最适合你：

方案一：复用Logstash同步（新手友好） #

利用Logstash的MongoDB输入插件拉取数据，再输出到Elasticsearch，无需额外部署新组件。

1. 安装Logstash MongoDB输入插件 #

在Logstash安装目录下执行：

bin/logstash-plugin install logstash-input-mongodb

2. 编写同步配置文件（示例：mongodb-to-es.conf） #

配置包含输入、过滤（可选）、输出三个部分：

input {
  mongodb {
    uri => "mongodb://localhost:27017" # 有认证则改为mongodb://user:password@localhost:27017
    database => "your_target_db" # 目标MongoDB数据库名
    collection => "your_target_collection" # 目标集合名
    codec => "json"
    # 增量同步配置（可选，避免重复读取）
    tracking_field => "@timestamp" # 用MongoDB中存储的时间字段作为同步标记
    last_run_metadata_path => "/var/log/logstash/mongo_sync_last.yml" # 存储同步进度的文件，需确保Logstash有权限读写
  }
}

filter {
  # 可选：处理字段冲突（比如MongoDB的_id会和Elasticsearch的_id重复）
  mutate {
    rename => { "_id" => "mongodb_document_id" }
  }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"] # Elasticsearch地址
    index => "mongo-logs-%{+YYYY.MM.dd}" # 按日期生成ES索引
    # 有ES认证则添加以下配置
    # user => "elastic"
    # password => "your_es_password"
  }

  # 可选：输出到控制台便于调试
  stdout {
    codec => rubydebug
  }
}

3. 启动Logstash同步任务 #

执行命令加载配置：

bin/logstash -f mongodb-to-es.conf

4. 验证同步结果 #

用Kibana Dev Tools查询数据：

GET mongo-logs-*/_search
{
  "query": { "match_all": {} }
}

或用curl：

curl -X GET "http://localhost:9200/mongo-logs-*/_search?pretty"

方案二：MongoDB官方连接器（生产环境推荐） #

MongoDB官方推出的同步工具，支持实时增量同步，适合长期稳定的生产场景。

1. 下载并安装连接器 #

选择匹配你MongoDB和Elasticsearch版本的连接器，解压到本地目录。

2. 编写同步配置文件（示例：sync_config.json） #

{
  "mongo": {
    "uri": "mongodb://localhost:27017"
  },
  "elasticsearch": {
    "uri": "http://localhost:9200"
  },
  "sync": {
    "name": "mongo-es-sync",
    "database": "your_target_db",
    "collection": "your_target_collection",
    "index": "mongo-logs",
    "pipeline": [] // 可选：MongoDB聚合管道，用于数据转换
  }
}

3. 启动连接器 #

执行命令启动同步：

./bin/mongodb-connector -c sync_config.json

关键注意事项 #

• 版本兼容性：确保Logstash插件/连接器与MongoDB、Elasticsearch的版本匹配，避免兼容性问题。
• 权限配置：MongoDB用户需有目标集合的读取权限；Elasticsearch用户需有索引创建、数据写入权限。
• 增量同步：使用tracking_field时，必须确保该字段是递增的（如时间戳、自增ID），否则会重复同步数据。

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