将 MongoDb 同步到 ElasticSearch

Question

我正在寻找一种将 MongoDB 中的集合与 Elastic Search (ES) 同步的方法。目标是将 MongoDB 作为主要数据源，并将 MongoDB 用作全文搜索引擎。 （我项目的业务逻辑是用python写的）。

网上有几种方法可用。

• Mongo 连接
• 河流插件
• logstash-input-mongodb（logstash 插件）see similar question
• 运输工具

但是，大多数建议都是几年前的，我找不到任何支持当前版本 ES (ES 7.4.0) 的解决方案。有人使用这样的构造吗？你有什么建议吗？

我考虑过放弃 MongoDB 作为主要数据源，只使用 ES 进行存储和搜索。虽然我读过 ES 不应该用作主要数据源。

---

编辑

谢谢@gurdeep.sabarwal。我按照你的方法。但是，我无法将 mongodb 同步到 ES。我的配置如下：

input {
    jdbc {
#        jdbc_driver_library => "/usr/share/logstash/mongodb-driver-3.11.0-source.jar"
        jdbc_driver_library => "/usr/share/logstash/mongojdbc1.5.jar"
#        jdbc_driver_library => "/usr/share/logstash/mongodb-driver-3.11.1.jar"

#        jdbc_driver_class => "mongodb.jdbc.MongoDriver"
#        jdbc_driver_class => "Java::com.mongodb.MongoClient"
        jdbc_driver_class => "Java::com.dbschema.MongoJdbcDriver"
        jdbc_driver_class => "com.dbschema.MongoJdbcDriver"
#        jdbc_driver_class => ""

        jdbc_connection_string => "jdbc:mongodb://<myserver>:27017/<mydb>"
        jdbc_user => "user"
        jdbc_password => "pw"
        statement => "db.getCollection('mycollection').find({})"
    }
}

output {
    elasticsearch {
        hosts => ["http://localhost:9200/"]
        index => "myindex"
    }
}

这让我离目标更近了一点。但是，我收到以下错误：

错误：Java::com.dbschema.MongoJdbcDriver 未加载。您确定您在 :jdbc_driver_library 中包含了正确的 jdbc 驱动程序吗？
异常：LogStash::ConfigurationError`

由于没有成功，我也尝试了注释版本，但没有成功。

Answer 1

1. 下载https://dbschema.com/jdbc-drivers/MongoDbJdbcDriver.zip
2. 解压并复制所有文件到路径(~/logstash-7.4.2/logstash-core/lib/jars/)
3. 修改下面的配置文件（mongo-logstash.conf）：
4. 运行：~/logstash-7.4.2/bin/logstash -f mongo-logstash.conf
5. 成功，请尝试！

ps：这是我在 stackoverflow 中的第一个答案 :-)

input {
  jdbc{
    # NOT THIS # jdbc_driver_class => "Java::mongodb.jdbc.MongoDriver"
    jdbc_driver_class => "com.dbschema.MongoJdbcDriver"
    jdbc_driver_library => "mongojdbc1.5.jar"
    jdbc_user => "" #no user and pwd
    jdbc_password => ""
    jdbc_connection_string => "jdbc:mongodb://127.0.0.1:27017/db1"
    statement => "db.t1.find()"
  }
}

output {
    #stdout { codec => dots }
    stdout { }
}

Answer 2

对于 ELK 堆栈，我已经使用（第一种和第二种）方法实现了，在进行研究时我遇到了多种方法，所以你可以选择任何人。 但我个人的选择是第一或第二，因为它为您提供了很多自定义选项。

如果您需要代码，请告诉我，我可以分享它的 sn-p。我不想回答太久！。

1.使用 dbSchemeJdbc jar(https://dbschema.com) 将数据从 mongodb 流式传输到 ElasticSearch。

a.OpenSource dbSchemeJdbc jar

b.您可以直接在logstash中编写原生mongodb查询或聚合查询。

您的管道可能如下所示：

input {
  jdbc{
    jdbc_user => "user"
    jdbc_password => "pass"
    jdbc_driver_class => "Java::com.dbschema.MongoJdbcDriver"
    jdbc_driver_library => "mongojdbc1.2.jar"
    jdbc_connection_string => "jdbc:mongodb://user:pass@host1:27060/cdcsmb"
    statement => "db.product.find()"
  }
}
output {
  stdout {
    codec => rubydebug 
  }
  elasticsearch {
    hosts => "localhost:9200"
    index => "target_index"
    document_type => "document_type"
    document_id => "%{id}"
  }
}

2.使用unityJdbc jar (http://unityjdbc.com) 将数据从mongodb流式传输到ElasticSearch

a.unityjdbc jar需要付费

b.您可以在logstash中编写SQL格式查询来从mongodb获取数据。

您的管道可能如下所示：

input {
  jdbc{
    jdbc_user => "user"
    jdbc_password => "pass"
    jdbc_driver_class => "Java::mongodb.jdbc.MongoDriver"
    jdbc_driver_library => "mongodb_unityjdbc_full.jar"
    jdbc_connection_string => "jdbc:mongodb://user:pass@host1:27060/cdcsmb"
    statement=> "SELECT * FROM employee WHERE status = 'active'" 
  }
}
output {
  stdout {
    codec => rubydebug 
  }
  elasticsearch {
    hosts => "localhost:9200"
    index => "target_index"
    document_type => "document_type"
    document_id => "%{id}"
  }
}

3.使用logstash-input-mongodb(https://github.com/phutchins/logstash-input-mongodb)插件将数据从mongodb流式传输到ElasticSearch

一种开源的

b.您获得的自定义选项非常少，它将转储整个集合，您不能编写查询或编写聚合查询等。

4.你可以用python或java编写你自己的程序，在elasticsearch中连接mongodb和索引数据，然后你可以使用cron来调度它。

5.你可以使用 node js Mongoosastic npm(https://www.npmjs.com/package/mongoosastic)，唯一的开销是它会在 mongo 和 ES 上提交更改以保持同步。

Answer 3

Monstache 似乎也是一个不错的选择，因为它支持 elasticsearch 和 mongodb 的最新版本：https://github.com/rwynn/monstache