概要:
目的:增量同步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.
- English document
- 中文架构介绍文档
- 第一次使用，如何配置
- Tutorial
- FAQ document
- MongoShake最佳实践
- Performance test document
- WeChat discuss group
操作步骤: 最新版: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
- 从MongoDB副本集同步到MongoDB副本集
- 从MongoDB副本集同步到MongoDB集群版
- 从MongoDB集群版同步到MongoDB集群版
- 从MongoDB副本集同步到kafka通道
- 云上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之间双向同步：
- 需要搭建2个mongoshake，一个从A到B，另一个从B到A
- 两条mongoshake不能同时用全量+增量（all）模式，正常应该是一个库为空（假设B），另一个有数据。那么从A到B先发起一次全量+增量同步，等待全量同步完毕以后，再从B到A发起一次增量同步。
- 双向同步需要依赖gid的开启，这个可以联系烛昭（通过售后联系），开启gid将会重启实例造成秒级别闪断。
- gid用于记录数据的产生地，比如从A产生的数据导入到B以后，不会被再导入回A，这样就不会产生环形复制。需要注意的是，这个gid只能用于增量，这也是第2条为什么一个方向通道是全量+增量，另一个方向通道需要搭建增量的原因。
- 云下开源的mongodb不能使用双向同步，因为gid的修改是在内核里面，所以开源不支持。
- 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: 如果目的库表存在，是否先删除目的库再进行同步。