剛開始我找到的方案是利用 ElasticSearch 的 River 來同步數據，並在 GitHub 上到了 MongoDB River 插件：elasticsearch-river-mongodb。但是，隨后我又在 ElasticSearch 官網上看了這篇博客：《Deprecating Rivers》，官方已經在 1.5 以后的版本棄用 River，為了用戶的遷移，會一直保留到 2.0 版本。
於是，我得另尋方案了。然后我又在網上找到了另外一個方案：mongo-connector。這個是 MongoDB 官方的開發人員用 Python 寫的一個工具，目前支持將 MongoDB 的數據同步到 Solr、ElasticSearch、MongoDB 中，並且支持用戶自己擴展。
首先安裝工具
yum install python-setuptools
easy_install pip
pip install cryptography
pip install "setuptools>=11.3"
pip install "elasticsearch>=6.1.0"
pip install mongo-connector
pip install elastic2-doc-manager
然后mongo開啟副本集，配置文件設置
vim mongodb.conf
replSet=es #設置副本集名稱
/ali/mongodb/bin/mongod -f /ali/mongodb/conf/mongodb.conf #啟動mongo
es:PRIMARY> rs.initiate() #進入mongo,初始化副本集
es:PRIMARY> rs.config() #查看副本集配置文件
es:PRIMARY> rs.status() #查看副本集狀態
es:PRIMARY> rs.status()
{
"set" : "es",
"date" : ISODate("2018-02-07T06:54:20.832Z"),
"myState" : 1,
"term" : NumberLong(2),
"heartbeatIntervalMillis" : NumberLong(2000),
"members" : [
{
"_id" : 0,
"name" : "iZwz9amcsywc3lxhketqdpZ:27017",
"health" : 1,
"state" : 1,
"stateStr" : "PRIMARY",
"uptime" : 1463,
"optime" : {
"ts" : Timestamp(1517985382, 1),
"t" : NumberLong(2)
},
"optimeDate" : ISODate("2018-02-07T06:36:22Z"),
"electionTime" : Timestamp(1517984997, 1),
"electionDate" : ISODate("2018-02-07T06:29:57Z"),
"configVersion" : 1,
"self" : true
}
],
"ok" : 1
}
mongo-connector -m 127.0.0.1:27017 -t 127.0.0.1:9200 -d elastic2_doc_manager #啟動同步服務
mongoDB需要用戶名密碼，使用下面的命令
mongo-connector -a root -p 123456 -m 127.0.0.1:27017 -t 127.0.0.1:9200 -d elastic2_doc_manager -n KYElog.* #說明，連接賬戶必須要超級管理員的權限 mongodb的root權限才行
mongo-connector -m 127.0.0.1:27017 -t 127.0.0.1:9200 -a username -p password -d elastic2_doc_manager-a mongodb用戶名
-p mongodb密碼
ES與MongoDB Insert插入操作的同步驗證
（1）Mongo端插入數據操作：
#Mongo創建數據庫（對應ES的Index） rs0:PRIMARY> use kyelog #Mongo中插入數據（其中col_02對應ES中的Type） rs0:PRIMARY> db.col_02.insert({name:"laoluo", birth:"1964-03-21", sex:"man", company:"chuizi"}); WriteResult({ "nInserted" : 1 }) rs0:PRIMARY> db.col_02.insert({name:"renzhengfei", birth:"1954-03-21", sex:"man", company:"huawei"});
查看是否同步到es
curl -XGET http://127.0.0.1:9200/kyelog/col_02/_search?pretty
然后更新數據或者刪除數據，查看es是否相應更改
導入mongodb中的數據到Elasticsearch集群
在安裝了monog-connector的機子的命令行中執行下面的命令（該命令為測試時的真實命令）。
mongo-connector --auto-commit-interval=0 -m ip_addr1:27018 -t ip_addr2:9200 -d elastic2_doc_manager -n db.collectionmongo-connector命令主要參數解析：
-m mongodb_host:port —— 數據源地址，mongodb數據庫地址。
-t target_host:port —— 數據目的地地址，elasticsearch/solr/mongodb集群地址。建議為集群中的協調節點的地址。
-d xxx_doc_manager —— 數據目的地的document類型。例如：
將mongodb中的數據同步到elasticsearch，使用elastic_doc_manager或elastic2_doc_manager。
將mongodb中的數據同步到solr，使用solr_doc_manager。
將mongodb中數據同步到其他mongodb，使用mongo_doc_manager。
-n db.collection ... —— 待同步的數據庫及其collection。默認同步所有數據庫。
-i filed_name ... —— 待同步的字段。默認同步所有字段。
-o mongodb_oplog_position.oplog —— mongo-connector的oplog。默認在mongo-connector命令執行目錄下創建oplog.timestamp文件。
建議重新分配存儲位置（也可重新分配存儲文件名），例如 /opt/mongo-connector.oplog。
--auto-commit-interval —— 數據同步間隔。默認在不同系統上有不同的值。設置為0表示mongodb中的任何操作立即同步到數據目的地。
--continue-on-error —— 一條數據同步失敗，日志記錄該失敗操作，繼續后續同步操作。默認為中止后續同步操作。
其他參數包括設置日志輸出行為（時間、間隔、路徑等）、設置mongodb登錄賬戶和密碼、設置（數據目的地）Http連接的證書等、設置mongo-connector的配置文件
。
使用mongo-connector同步數據注意事項
1. mongodb必須開啟副本集（Replica Set）。開啟副本集才會產生oplog，副本拷貝主分片的oplog並通過oplog與主分片進行同步。
mongo-connector也是通過oplog進行數據同步，故必須開啟副本集。
2. 使用mongo-connector命令同步數據時，-m參數中的mongodb地址應該是主/從分片的地址，
從該地址登錄可以看見並操作local數據庫（oplog存儲在local.oplog.rs）；不能使用mongoos地址。
3. 使用mongo-connector命令同步數據時 ，mongo-connector的oplog（參照-o參數）不能隨便刪除，
否則會引起重新同步所有數據的問題。該問題可以通過--no-dump選項關閉。
4. 生產環境下建議將mongo-connector配置為系統服務，運行mongo-connector時采用配置文件的方式。
踩過的坑
1. 數據庫A中有多個集合(A1, A2, A3)，且已開啟了副本集（Replica Set），但是集合A1可以同步，集合A2不能同步.
原因：oplog中有A1的操作記錄，沒有A2的操作記錄。
結論：開啟副本集（Replica Set）並不能保證一定能同步，oplog中必須包含待同步集合的操作記錄，才能通過mongo-connector同步到Elasticsearch集群。
2. mongodb3.x版本加強了安全機制，導致了在只擁有某個庫的權限時不能同步數據的問題。
原因：擁有某個庫的權限，並不能擁有oplog的讀取權限，而mongo-connector需要讀取oplog的權限。
結論：同步數據至少需要能夠讀取oplog的權限，確保當前mongodb用戶的權限能夠操作oplog，或者直接使用mongodb的管理員權限。