mongodb 增量同步之 MongoShake(1)

概要:

目的:增量同步mongodb 的数据(mongo需集群,或者副本集模式)

        1.官网文档

         2 基本安装,启动,监控

         3.小试牛刀,最佳实践

0.官网文档:

github: https://github.com/alibaba/MongoShake/

releases: https://github.com/alibaba/MongoShake/releases?spm=a2c4e.10696291.0.0.7aac19a4ZhZPfe

This is a brief introduction of Mongo-Shake, please visit english wiki or chinese wiki if you want to see more details including architecture, data flow, performance test, business showcase and so on.

1.基本安装: https://help.aliyun.com/document_detail/122621.html?spm=a2o8d.corp_prod_req_detail.0.0.3b1d23d3Cjcmsa

操作步骤:  最新版:https://github.com/alibaba/MongoShake/releases

1.执行如下命令下载MongoShake程序,并重命名为mongoshake.tar.gz
wget "http://docs-aliyun.cn-hangzhou.oss.aliyun-inc.com/assets/attach/196977/jp_ja/1608863913991/mongo-shake-v2.4.16.tar.gz" -O mongoshake.tar.gz
2.解压: tar zxvf mongoshake.tar.gz
3.执行vi collector.conf命令，修改MongoShake的配置文件collector.conf，涉及的主要参数说明如: https://developer.aliyun.com/article/719704
主要参数:
mongo_urls  :  源端MongoDB实例的ConnectionStringURI格式连接地址。
tunnel.address: 目标端MongoDB实例的ConnectionStringURI格式连接地址。
sync_mode    : 数据同步的方式，取值：
all：执行全量数据同步和增量数据同步。
full：仅执行全量数据同步。
incr：仅执行增量数据同步。
4.执行下述命令启动同步任务，并打印日志信息。
./collector.linux -conf=collector.conf -verbose
最新版(2.6.5):
./collector.linux -conf=collector.conf -verbose 2
5.观察打印的日志信息，当出现如下日志时，即代表全量数据同步已完成，并进入增量数据同步模式。
ngoshake/collector.(*ReplicationCoordinator).Run:80) finish full sync, start incr sync with timestamp: fullBeginTs[1560994443], fullFinishTs[1560994737]

2.监控MongoShake状态

增量数据同步开始后，您可以再开启一个命令行窗口，通过如下命令来监控MongoShake。 
 ./mongoshake-stat --port=9100

参数说明
logs_get/sec 每秒获取的oplog数量。
logs_repl/sec 每秒执行重放操作的oplog数量。
logs_success/sec 每秒成功执行重放操作的oplog数量。
lsn.time 最后发送oplog的时间。
lsn_ack.time 目标端确认写入的时间。
lsn_ckpt.time CheckPoint持久化的时间。
now.time 当前时间。
replset 源数据库的副本集名称。

3.MongoShake最佳实践,详见:https://developer.aliyun.com/article/719704

  1. 从MongoDB副本集同步到MongoDB副本集
  2. 从MongoDB副本集同步到MongoDB集群版
  3. 从MongoDB集群版同步到MongoDB集群版
  4. 从MongoDB副本集同步到kafka通道
  5. 云上MongoDB副本集的双向同步

1. 从MongoDB副本集同步到MongoDB副本集

假设源端是三副本：10.1.1.1:1001, 10.2.2.2:2002, 10.3.3.3:3003，目的端也是三副本：10.5.5.5:5005, 10.6.6.6:6006, 10.7.7.7:7007。同步模式是全量+增量同步。
则用户需要修改以下几个参数：

mongo_urls = mongodb://username:password@10.1.1.1:1001,10.2.2.2:2002,10.3.3.3:3003 #源端连接串信息，逗号分隔不同的mongod
sync_mode = all # all 表示全量+增量，full表示仅全量，incr表示仅增量
tunnel.address = mongodb://username:password@10.5.5.5:5005, 10.6.6.6:6006, 10.7.7.7:7007 #目的端连接串信息，逗号分隔不同的mongod
incr_sync.mongo_fetch_method = oplog # 如果希望以change stream拉取，该值需要配置change_stream，支持>=4.0.1版本。

2. 从MongoDB副本集同步到MongoDB集群版

假设源同样是三副本：10.1.1.1:1001, 10.2.2.2:2002, 10.3.3.3:3003，目的端是sharding，有多个mongos：20.1.1.1:2021, 20.2.2.2:2022, 20.3.3.3:3033。

mongo_urls = mongodb://username:password@10.1.1.1:1001,10.2.2.2:2002,10.3.3.3:3003 #源端连接串信息，逗号分隔不同的mongod
sync_mode = all # all 表示全量+增量，document表示仅全量，oplog表示仅增量
tunnel.address = mongodb://username:password@20.1.1.1:2021;20.2.2.2:2022;20.3.3.3:3033 #目的端连接串信息，分号分割不同的mongos。也可以只配置部分，配置多个mongos可以做负载均衡写入。
incr_sync.mongo_fetch_method = oplog # 如果希望以change stream拉取，该值需要配置change_stream，支持>=4.0.1版本。

3. 从MongoDB集群版同步到MongoDB集群版

假设源是2节点：节点1是10.1.1.1:1001, 10.1.1.2:2002, 10.1.1.3:3003；节点2是10.2.2.1:1001, 10.2.2.2:2002, 10.2.2.3:3003，mongos：10.1.1.10:1010。目的端是sharding，有多个mongos：20.1.1.1:2021, 20.2.2.2:2022, 20.3.3.3:3033。

mongo_urls = mongodb://username1:password1@10.1.1.1:1001,10.1.1.2:2002,10.1.1.3:3003;mongodb://username2:password2@10.2.2.1:1001,10.2.2.2:2002,10.2.2.3:3003 #源端连接串信息，逗号分隔同一个shard不同的mongod，分号分隔不同的shard。
mongo_cs_url = mongodb://username1:password1@10.5.5.5:5555,10.5.5.6:5556 # 如果源端是sharding，此处需要配置源端sharding的cs的地址
mongo_s_url = mongodb://username_s:password_s@10.1.1.10:1010 # 如果希望采用change stream拉取，则还需要配置mongos的地址，一个就够了
sync_mode = all # all 表示全量+增量，document表示仅全量，oplog表示仅增量
tunnel.address = mongodb://username:password@20.1.1.1:2021;20.2.2.2:2022;20.3.3.3:3033 #目的端连接串信息，分号分割不同的mongos。
incr_sync.mongo_fetch_method = oplog # 如果希望以change stream拉取，该值需要配置change_stream，支持>=4.0.1版本。

4. 从MongoDB副本集同步到kafka通道

假设源同样是三副本：10.1.1.1:1001, 10.2.2.2:2002, 10.3.3.3:3003，目的kafka是50.1.1.1:6379，topic是test。

mongo_urls = mongodb://username:password@10.1.1.1:1001,10.2.2.2:2002,10.3.3.3:3003 #源端连接串信息，逗号分隔不同的mongod
sync_mode = incr # 如果目的端不是mongodb，仅支持增量同步模式
tunnel = kafka
tunnel.address = mongoshake_test@192.168.18.129:9092 #topic@ip:port
incr_sync.mongo_fetch_method = oplog # 如果希望以change stream拉取，该值需要配置change_stream，支持>=4.0.1版本。

5. 云上MongoDB副本集的双向同步

云上副本集的双向同步可以参考副本集的单向同步，但是需要注意的有以下几点，假设A和B之间双向同步：

  1. 需要搭建2个mongoshake，一个从A到B，另一个从B到A
  2. 两条mongoshake不能同时用全量+增量（all）模式，正常应该是一个库为空（假设B），另一个有数据。那么从A到B先发起一次全量+增量同步，等待全量同步完毕以后，再从B到A发起一次增量同步。
  3. 双向同步需要依赖gid的开启，这个可以联系烛昭（通过售后联系），开启gid将会重启实例造成秒级别闪断。
  4. gid用于记录数据的产生地，比如从A产生的数据导入到B以后，不会被再导入回A，这样就不会产生环形复制。需要注意的是，这个gid只能用于增量，这也是第2条为什么一个方向通道是全量+增量，另一个方向通道需要搭建增量的原因。
  5. 云下开源的mongodb不能使用双向同步，因为gid的修改是在内核里面，所以开源不支持。
  6. sharding同样也支持双向同步

6. 部分配置文件参数解释

v2.4开始的参数请参考github wiki配置参数说明，下面是2.2及之前的参数。具体请查看配置文件的注释，此处只做简单解释

  • mongo_urls: 源mongodb的连接地址
  • mongo_connect_mode: 源端连接的模式，有几种模式可选：从seconary拉取；从primary拉取；secondary优先拉取；单节点拉取
  • sync_mode: sync模式，有几种模式可选：全量，增量，全量+增量
  • http_profile: 提供restful接口，用户可以查看一些内部运行情况，也可以对接监控。
  • system_profile: profile端口，可以查看进程运行的堆栈情况。
  • log: log日志相关参数。
  • filter.namespace.black: 黑名单过滤。黑名单内的库表不会被同步，剩下的同步。
  • filter.namespace.white: 白名单过滤。白名单内的库表会被同步，剩下的过滤掉。黑白名单最多只能配置一个，不配置会同步所有库表。
  • filter.pass.special.db: 有些特别的库表会被过滤，如admin，local, config库，如果一定要开启，可以在这里进行配置。
  • oplog.gids: 用于云上双向同步。
  • shard_key: 内部对数据多线程的哈希方式，默认collection表示按表级别进行哈希。
  • worker: 增量阶段并发写入的线程数，如果增量阶段性能不够，可以提高这个配置。
  • worker内部相关配置: worker.batch_queue_size, adaptive.batching_max_size, fetcher.buffer_capacity, 关于内部队列的相关配置，具体请参考github wiki文档。
  • worker.oplog_compressor: 压缩模式，如果是非direct模式开启这个可以减少网络传输的开销。
  • tunnel.address: 目的端对接的地址。
  • context.storage: checkpoint存储的位置，database表示把数据存入MongoDB，api表示把数据存入用户自己提供的http接口。
  • context.storage.url: checkpoint写入到哪个MongoDB，如果源是sharding，此处配置cs地址，checkpoint会写入admin库；如果是副本集，不配置，会默认写入源库，配置则写入配置的库里面。
  • context.address: checkpoint写入的表的名字。
  • context.start_position: checkpoint启动开始拉取的增量时间位点。如果本身checkpoint已经存在（参考上述context的位置），那么则按照context信息进行拉取，如果不存在，则按照这个位点进行增量拉取。
  • master_quorum: 如果以主备模式拉取同一个源，则这个参数需要启用。
  • transform.namespace: 命名空间的转换，a.b:c.d表示把源端a库下面的c表同步到目的端c库下面的d表。
  • replayer.dml_only: 默认不同步DDL，false表示同步DDL。DDL包括建表，删库，建索引等语句。
  • replayer.executor.upsert: 目的端如果update语句对应的主键id不存在，是否将update语句更改为insert语句。
  • replayer.executor.insert_on_dup_update: 目的端如果insert语句对应的主键id已经存在，是否将insert语句更改为update语句。
  • replayer.conflict_write_to: 对于写入冲突的情况，是否需要记录冲突的文档。
  • replayer.durable: 测试选项，false表示取消写入，只用于拉取调试。
  • replayer.collection_parallel: 全量同步按表并发的并发度。
  • replayer.document_parallel: 全量同步同一个表内并发写入的线程数。
  • replayer.document_batch_size: 全量同步一次性batch的大小。
  • replayer.collection_drop: 如果目的库表存在，是否先删除目的库再进行同步。

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/lshan/p/15338860.html