码迷，mamicode.com
首页 > 数据库 > 详细

mongo-connector实现MongoDB与elasticsearch实时同步深入详解

时间：2018-12-14 13:02:20      阅读：402      评论：0      收藏：0      [点我收藏+]

标签：this   user   运行   repr   strong   详细   img   update   帮助   

引言：

验证表明：mongo-connector工具支持MongoDB与ES之间的实时增insert、删delete、改update操作。
对于历史数据，mongo-connector工具不能同步到ES中，根因是本身工具不支持(初步界定），还是没有这种场景，待查（进一步研究后再更新）。

1. mongo-connector 地址：

https://github.com/mongodb-labs/mongo-connector

2、 mongo-connector 工具简介

mongo-connector工具创建一个从MongoDB簇到一个或多个目标系统的管道，目标系统包括：Solr，Elasticsearch，或MongoDB簇。
该工具在MongoDB与目标系统间同步数据，并跟踪MongoDB的oplog，保持操作与MongoDB的实时同步。
该工具已经在python2.6，2.7,3.3+下进行验证。
mongo-connector工具是基于python开发的实时同步服务工具。它要求mongo运行在replica-set模式，且需要 elastic2_doc_manager将数据写入ES。
技术分享图片

3、 elastic2-doc-manager 工具简介

这是Elastic2.x版本的文档管理器。对应Elastic1.x版本需要使用 elastic-doc-manager。

4、ES与MongoDB同步步骤:

（1）安装 mongo-connector。

pip install mongo-connector

（2）安装 elastic2-doc-manager。

pip install elastic2-doc-manager

注意：
如果不安装（2）直接进入（3）、（4）则会报错：

[root@5b9dbaaa148a bin]# mongo-connector -m 10.8.5.99:27017 -t 10.8.5.101:9200 -d elastic2_doc_manager
Logging to mongo-connector.log.
Exception in thread Thread-1:
Traceback (most recent call last):
  File "/usr/lib64/python2.6/threading.py", line 532, in __bootstrap_inner
  self.run()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6

（3）mongo端启动

MongoDB 必须开启复制集，如果已经开启请忽略这一步：

1）通过 –replSet 设定副本集名称。

[root@b48eafd69929 bin]# ./mongod --replSet "rs0"
  • 1

2）将mongo与副本集成员连接

[root@b48eafd69929 bin]# ./mongo
MongoDB shell version: 3.2.4
connecting to: test
Server has startup warnings:
2016-07-05T09:49:01.330+0100 I CONTROL [initandlisten] ** WARNING: You are running this process as the root user, which is not recommended.
2016-07-05T09:49:01.330+0100 I CONTROL [initandlisten]
2016-07-05T09:49:01.331+0100 I CONTROL [initandlisten]
2016-07-05T09:49:01.331+0100 I CONTROL [initandlisten] ** WARNING: You are running on a NUMA machine.
2016-07-05T09:49:01.331+0100 I CONTROL [initandlisten] ** We suggest launching mongod like this to avoid performance problems:
2016-07-05T09:49:01.332+0100 I CONTROL [initandlisten] ** numactl --interleave=all mongod [other options]
2016-07-05T09:49:01.332+0100 I CONTROL [initandlisten]
2016-07-05T09:49:01.332+0100 I CONTROL [initandlisten] ** WARNING: /sys/kernel/mm/transparent_hugepage/enabled is ‘always‘.
2016-07-05T09:49:01.332+0100 I CONTROL [initandlisten] ** We suggest setting it to ‘never‘
2016-07-05T09:49:01.332+0100 I CONTROL [initandlisten]
2016-07-05T09:49:01.332+0100 I CONTROL [initandlisten] ** WARNING: /sys/kernel/mm/transparent_hugepage/defrag is ‘always‘.
2016-07-05T09:49:01.332+0100 I CONTROL [initandlisten] ** We suggest setting it to ‘never‘
2016-07-05T09:49:01.332+0100 I CONTROL [initandlisten]
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17

3）初始化副本集

> rs.initiate()
{
  "info2" : "no configuration specified. Using a default configuration for the set",
  "me" : "b48eafd69929:27017",
  "ok" : 1
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6

4）【验证】初始化副本集的配置

rs0:SECONDARY> rs.conf()
{
  "_id" : "rs0",
  "version" : 1,
  "protocolVersion" : NumberLong(1),
  "members" : [
  {
  "_id" : 0,
  "host" : "b48eafd69929:27017",
  "arbiterOnly" : false,
  "buildIndexes" : true,
  "hidden" : false,
  "priority" : 1,
  "tags" : {

  },
  "slaveDelay" : NumberLong(0),
  "votes" : 1
  }
  ],
  "settings" : {
  "chainingAllowed" : true,
  "heartbeatIntervalMillis" : 2000,
  "heartbeatTimeoutSecs" : 10,
  "electionTimeoutMillis" : 10000,
  "getLastErrorModes" : {

  },
  "getLastErrorDefaults" : {
  "w" : 1,
  "wtimeout" : 0
  },
  "replicaSetId" : ObjectId("577b74bd0ba41a313110ad62")
  }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35

5）【验证】副本集的状态。

rs0:PRIMARY> rs.status()
{
  "set" : "rs0",
  "date" : ISODate("2016-07-05T08:50:55.272Z"),
  "myState" : 1,
  "term" : NumberLong(1),
  "heartbeatIntervalMillis" : NumberLong(2000),
  "members" : [
  {
  "_id" : 0,
  "name" : "b48eafd69929:27017",
  "health" : 1,
  "state" : 1,
  "stateStr" : "PRIMARY",
  "uptime" : 115,
  "optime" : {
  "ts" : Timestamp(1467708606, 1),
  "t" : NumberLong(1)
  },
  "optimeDate" : ISODate("2016-07-05T08:50:06Z"),
  "infoMessage" : "could not find member to sync from",
  "electionTime" : Timestamp(1467708605, 2),
  "electionDate" : ISODate("2016-07-05T08:50:05Z"),
  "configVersion" : 1,
  "self" : true
  }
  ],
  "ok" : 1
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29

（4）ES端同步操作

[root@5b9dbaaa148a bin]# mongo-connector -m 10.8.5.99:27017 -t 10.8.5.101:9200 -d elastic2_doc_manager
Logging to mongo-connector.log.
  • 1
  • 2

参数含义:
-m: mongodb的地址与端口，端口默认为27017。
-t：ES的地址与端口，端口默认为9200。
-d：doc manager的名称，2.x版本为： elastic2-doc-manager。

5、ES与MongoDB Insert插入操作的同步验证

（1）Mongo端插入数据操作：

#Mongo创建数据库（对应ES的Index）
rs0:PRIMARY> use zhang_index
switched to db zhang_index

#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"});
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8

（2）Es端检索验证

[root@5b9dbaaa148a test_log]# curl -XGET http://10.8.5.101:9200/zhang_index/col_02/_search?pretty
{
  "took" : 4,
  "timed_out" : false,
  "_shards" : {
  "total" : 8,
  "successful" : 8,
  "failed" : 0
  },
  "hits" : {
  "total" : 2,
  "max_score" : 1.0,
  "hits" : [ {
  "_index" : "zhang_index",
  "_type" : "col_02",
  "_id" : "577b7d8ceb8e3dc2d1db12a9",
  "_score" : 1.0,
  "_source" : {
  "company" : "huawei",
  "name" : "renzhengfei",
  "birth" : "1954-03-21",
  "sex" : "man"
  }
  }, {
  "_index" : "zhang_index",
  "_type" : "col_02",
  "_id" : "577b7d4aeb8e3dc2d1db12a7",
  "_score" : 1.0,
  "_source" : {
  "company" : "chuizi",
  "name" : "laoluo",
  "birth" : "1964-03-21",
  "sex" : "man"
  }
  } ]
  }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37

6、 ES与MongoDB Update更新操作的同步验证

（1）MongoDB的更新update操作

rs0:PRIMARY> db.col_02.update({‘name‘:‘laoluo‘}, {$set:{‘name‘:‘luoyonghao‘}})
WriteResult({ "nMatched" : 1, "nUpserted" : 0, "nModified" : 1 })
rs0:PRIMARY>
rs0:PRIMARY> db.col_02.find().pretty()
{
  "_id" : ObjectId("577b7d4aeb8e3dc2d1db12a7"),
  "name" : "luoyonghao",
  "birth" : "1964-03-21",
  "sex" : "man",
  "company" : "chuizi"
}
{
  "_id" : ObjectId("577b7d8ceb8e3dc2d1db12a9"),
  "name" : "renzhengfei",
  "birth" : "1954-03-21",
  "sex" : "man",
  "company" : "huawei"
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18

（2）Es端检索更新后结果

[root@5b9dbaaa148a test_log]# curl -XGET http://10.8.5.101:9200/zhang_index/col_02/_search?pretty
{
  "took" : 1,
  "timed_out" : false,
  "_shards" : {
  "total" : 8,
  "successful" : 8,
  "failed" : 0
  },
  "hits" : {
  "total" : 2,
  "max_score" : 1.0,
  "hits" : [ {
  "_index" : "zhang_index",
  "_type" : "col_02",
  "_id" : "577b7d8ceb8e3dc2d1db12a9",
  "_score" : 1.0,
  "_source" : {
  "company" : "huawei",
  "name" : "renzhengfei",
  "birth" : "1954-03-21",
  "sex" : "man"
  }
  }, {
  "_index" : "zhang_index",
  "_type" : "col_02",
  "_id" : "577b7d4aeb8e3dc2d1db12a7",
  "_score" : 1.0,
  "_source" : {
  "company" : "chuizi",
  "name" : "luoyonghao",
  "birth" : "1964-03-21",
  "sex" : "man"
  }
  } ]
  }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37

7、 ES与MongoDB delete删除操作的同步验证

（1） MongoDB的删除delete操作

rs0:PRIMARY> db.col_02.remove({‘name‘:‘renzhengfei‘})
WriteResult({ "nRemoved" : 1 })
rs0:PRIMARY> db.col_02.find()
{ "_id" : ObjectId("577b7d4aeb8e3dc2d1db12a7"), "name" : "luoyonghao", "birth" : "1964-03-21", "sex" : "man", "company" : "chuizi" }
rs0:PRIMARY> db.col_02.find().pretty()
{
  "_id" : ObjectId("577b7d4aeb8e3dc2d1db12a7"),
  "name" : "luoyonghao",
  "birth" : "1964-03-21",
  "sex" : "man",
  "company" : "chuizi"
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12

（2）ES端检索删除后结果

结果表明，MongoDB删除的内容，ES端已经同步删除。

[root@5b9dbaaa148a test_log]# curl -XGET http://10.8.5.101:9200/zhang_index/col_02/_search?pretty
{
  "took" : 2,
  "timed_out" : false,
  "_shards" : {
  "total" : 8,
  "successful" : 8,
  "failed" : 0
  },
  "hits" : {
  "total" : 1,
  "max_score" : 1.0,
  "hits" : [ {
  "_index" : "zhang_index",
  "_type" : "col_02",
  "_id" : "577b7d4aeb8e3dc2d1db12a7",
  "_score" : 1.0,
  "_source" : {
  "company" : "chuizi",
  "name" : "luoyonghao",
  "birth" : "1964-03-21",
  "sex" : "man"
  }
  } ]
  }
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26

技术分享图片

参见详细介绍：

https://docs.mongodb.com/manual/tutorial/deploy-replica-set/

Mongo与ES同步的5种方式：

https://www.linkedin.com/pulse/5-way-sync-data-from-mongodb-es-kai-hao

常见Bug：

How to setup a MongoDB replica set for the connector？
https://docs.mongodb.com/manual/tutorial/deploy-replica-set/

——————————————————————————————————
更多ES相关实战干货经验分享，请扫描下方【铭毅天下】微信公众号二维码关注。
（每周至少更新一篇！）

技术分享图片
和你一起，死磕Elasticsearch！
——————————————————————————————————

2016年7月6日 23:22 思于家中床前

作者：铭毅天下
转载请标明出处，原文地址：
http://blog.csdn.net/laoyang360/article/details/51842822
如果感觉本文对您有帮助，请点击‘顶’支持一下，您的支持是我坚持写作最大的动力，谢谢！

再分享一下我老师大神的人工智能教程吧。零基础！通俗易懂！风趣幽默！希望你也加入到我们人工智能的队伍中来！http://www.captainbed.net

mongo-connector实现MongoDB与elasticsearch实时同步深入详解

标签：this   user   运行   repr   strong   详细   img   update   帮助   

原文地址：https://www.cnblogs.com/sjwucx/p/10118231.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！