python-3.x - 将 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"
    }
}

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

Error: Java::com.dbschema.MongoJdbcDriver not loaded. Are you sure you've included the correct jdbc driver in :jdbc_driver_library?
Exception: 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 { }
}