导读

MongoShake是基于MongoDB的通用型平台服务，作为数据连通的桥梁，打通各个闭环节点的通道。MongoShake以golang语言进行编写，通过读取MongoDB集群的Oplog操作日志，对MongoDB的数据进行复制，后续通过操作日志实现特定需求。

相比于MongoShake方式，DTS MongoDB->MongoDB同步具有如下优势：

• 可在阿里云DTS控制台上看到同步过程，比如现在同步任务正处于结构迁移，还是全量迁移，还是进入了增量迁移。可看到同步的RPS，BPS。可看到同步任务是否失败或者有延迟。
  
• 由于阿里云DTS是第三方迁移工具，数据源MongoDB和目标MongoDB可以属于不同的VPC网络。
  
• 支持ETL，库表映射等功能。
  
• 支持数据过滤，可以只同步某个表指定条件的数据。 也可以选择只同步某些DML或者DDL。
  
• 可以随时暂停、重启同步任务。支持断点续传。
  
• 全量并发切片支持单表同时存在多类型_id的情况
  
• 源端为MongoDB分片集群时，DTS采用从Shard读取数据和oplog的方式进行全量和增量同步，并且增量迁移期间源端不需要关闭balancer。
  

本文聚焦DTS MongoDB->MongoDB 和 MongoShake 数据同步的性能，分别针对副本集/分片集群架构、单表/多表、全量/增量同步进行性能的对比。

一、副本集

1.实例配置

源端：MongoDB4.4，8核16GB，300GB存储

目的端：MongoDB4.4，8核16GB，300GB存储

MongoShake2.6.6部署在阿里云ECS上：CentOS7，实例规格：ecs.u1-c1m2.2xlarge（8核16GB），通过 私网链接-专有网络（VPC）的形式连接源/目的端MongoDB实例

DTS通过反向VPC连接源/目的端MongoDB实例

2.单表全量迁移

2.1 源端数据准备

源端使用开源工具ycsb构建测试数据

ycsb使用方式可参考：https://github.com/brianfrankcooper/YCSB/

该项测试中，源端为单表1000W条数据，每条数据 10个field，每个field 100 bytes

使用 ycsb workloada 模板进行数据构建，部分参数如下：

recordcount=10000000
operationcount=10000000
workload=site.ycsb.workloads.CoreWorkload
readallfields=true
fieldcount=10
fieldlength=100
readproportion=0
updateproportion=0.3
scanproportion=0
insertproportion=0.7
requestdistribution=zipfian

ycsb 使用指令可参考：

cd ycsb所在目录
./bin/ycsb load mongodb -s -P workloads/workloada -p mongodb.url=mongodb://user:password@server1.example.com:9999,server2.example.com:9999/dbname -p table=tablename -threads 64

2.2 MongoShake单表全量迁移测试

MongoShake使用可参考：https://developer.aliyun.com/article/603329

2.2.1 MongoShake参数设置

修改MongoShake collector.config，部分参数如下

# --------------------------- full sync configuration ---------------------------
# the number of collection concurrence
# 并发最大拉取的表个数，例如，6表示同一时刻shake最多拉取6个表。
full_sync.reader.collection_parallel = 6
# the number of document writer thread in each collection.
# 同一个表内并发写的线程数，例如，8表示对于同一个表，将会有8个写线程进行并发写入。
full_sync.reader.write_document_parallel = 40
# number of documents in a batch insert in a document concurrence
# 目的端写入的batch大小，例如，128表示一个线程将会一次聚合128个文档然后再写入。
full_sync.reader.document_batch_size = 1024
# max number of fetching thread per table. default is 1
# 单个表最大拉取的线程数，默认是单线程拉取。需要具备splitVector权限。
# 注意：对单个表来说，仅支持索引对应的value是同种类型，如果有不同类型请勿启用该配置项！
full_sync.reader.parallel_thread = 3
# the parallel query index if set full_sync.reader.parallel_thread. index should only has
# 1 field.
# 如果设置了full_sync.reader.parallel_thread，还需要设置该参数，并行拉取所扫描的index，value
# 必须是同种类型。对于副本集，建议设置_id；对于集群版，建议设置shard_key。key只能有1个field。
full_sync.reader.parallel_index = _id
# enable majority write in full sync.
# the performance will degrade if enable.
# 全量阶段写入端是否启用majority write
full_sync.executor.majority_enable = true

2.2.2 MongoShake测试结果

迁移期间MongoShake所在ECS实例的cpu、内存、内网带宽等无瓶颈

通过目的端MongoDB实例的监控信息可得：MongoShake 上述单表全量迁移的QPS在8000个/s左右。

2.3 DTS单表全量迁移测试

目前DTS MongoDB->MongoDB 全量迁移免费试用，使用2xlarge规格链路进行全量迁移测试

默认运行内存8GB，40个线程并发写目的端

通过目的端MongoDB实例的监控信息可得：DTS 上述单表全量迁移的QPS在11000个/s左右。

3.多表全量迁移

3.1 源端数据准备

源端使用开源工具ycsb构建测试数据

该项测试中，源端为5张表，每张表表500W条数据，每条数据 10个field，每个field 100 bytes

使用 ycsb workloada 模板进行数据构建，部分参数如下：

recordcount=5000000
operationcount=5000000
workload=site.ycsb.workloads.CoreWorkload
readallfields=true
fieldcount=10
fieldlength=100
readproportion=0
updateproportion=0.3
scanproportion=0
insertproportion=0.7
requestdistribution=zipfian

3.2 MongoShake多表全量迁移测试

迁移期间MongoShake所在ECS实例的cpu、内存、内网带宽等无瓶颈

通过目的端MongoDB实例的监控信息可得：MongoShake 上述多表全量迁移的QPS在18000个/s左右。

3.3 DTS 多表全量迁移测试

目前DTS MongoDB->MongoDB 全量迁移免费试用，使用2xlarge规格链路进行全量迁移测试

默认运行内存8GB，40个线程并发写目的端，同时考虑到读取数据对源端实例的压力，DTS限制了从源端拉取数据的线程数。

通过目的端MongoDB实例的监控信息可得：默认情况下 DTS 上述多表全量迁移的QPS在15000个/s左右。参照MongoShake 单表3个读线程的设置，在DTS 修改读线程数量后，上述多表全量迁移的QPS在20000个/s左右。

4.增量迁移

4.1 源端数据准备

源端使用开源工具ycsb构建测试数据

该项测试中，源端为1张表，单表500W条数据，每条数据 10个field，每个field 100 bytes

使用 ycsb workloada 模板进行数据构建，参数同上一小节

4.2 MongoShake增量迁移测试

4.2.1 MongoShake参数设置

修改MongoShake collector.config，部分参数如下

# --------------------------- incrmental sync configuration ---------------------------
# fetch method:
# oplog: fetch oplog from source mongodb (default)
# change_stream: use change to receive change event from source mongodb, support MongoDB >= 4.0.
# we recommand to use change_stream if possible.
incr_sync.mongo_fetch_method = oplog
# 同步模式，all表示全量+增量同步，full表示全量同步，incr表示增量同步。
sync_mode = incr
# hash的方式，id表示按文档hash，collection表示按表hash，auto表示自动选择hash类型。
# 如果没有索引建议选择id达到非常高的同步性能，反之请选择collection。
incr_sync.shard_key = auto
# 内部发送（写目的DB）的worker数目，如果机器性能足够，可以提高worker个数。
incr_sync.worker = 40
# batch_queue_size：每个worker线程的队列长度，worker线程从此队列取任务
# batching_max_size：一次分发给worker的任务最多包含多少个文档
# buffer_capacity：PendingQueue队列中一个buffer至少包含的文档个数，进行序列化
incr_sync.worker.batch_queue_size = 64
incr_sync.adaptive.batching_max_size = 1024
incr_sync.fetcher.buffer_capacity = 256
# 增量阶段写入端是否启用majority write
incr_sync.executor.majority_enable = true

4.2.2 MongoShake测试结果

迁移期间MongoShake所在ECS实例的cpu、内存、内网带宽等无瓶颈

通过目的端MongoDB实例的监控信息可得：MongoShake 上述增量迁移的QPS在44000个/s左右。

4.3 DTS 增量迁移测试

使用2xlarge规格链路进行增量迁移测试，40个线程写目的端。

通过目的端MongoDB实例的监控信息可得：DTS 上述增量迁移的QPS在49000个/s左右。

5.小结

二、分片集群

1.架构说明

当源端是MongoDB分片集群时，MongoShake支持两种增量同步方案：1.直接读取各个shard的oplog并回放，但需要源端实例始终关闭balancer；2.通过Mongos来获取Change Stream，可以不关闭balancer，但性能弱于oplog的方式。

DTS采用直接拉取oplog的方式进行增量同步，并且通过分布式子任务的方式，实现了各个shard对应的同步链路之间的协同，解决了源端实例shard之间发生chunk move给同步带来的问题，使得在增量迁移期间源端实例可以开启balancer。

考虑到从各个shard直接读取数据和oplog进行迁移的方案与源端为副本集时类似，本章主要进行源端在开启balancer情况下，MongoShake使用ChangeStream进行增量同步与DTS增量同步的性能对比。

2.实例配置

源端：MongoDB4.2，2个Mongos：8核16GB，3个shard：8核16GB 200GB存储

目的端：MongoDB4.4，2个Mongos：8核16GB，3个shard：8核16GB 200GB存储

MongoShake2.6.6部署在阿里云ECS上：CentOS7，实例规格：ecs.u1-c1m2.2xlarge（8核16GB），通过 私网链接-专有网络（VPC）的形式连接源/目的端MongoDB实例

DTS通过反向VPC连接源/目的端MongoDB实例

3.增量迁移

3.1源端数据准备

源端和目的端MongoDB实例已预先设置分片键：{"_id":"hashed"}

源端使用开源工具ycsb构建测试数据

该项测试中，源端为单表1000W条数据，每条数据 10个field，每个field 100 bytes

使用 ycsb workloada 模板进行数据构建，部分参数如下：

recordcount=10000000
operationcount=10000000
workload=site.ycsb.workloads.CoreWorkload
readallfields=true
fieldcount=10
fieldlength=100
readproportion=0
updateproportion=0.3
scanproportion=0
insertproportion=0.7
requestdistribution=zipfian

3.2 MongoShake Change Stream增量迁移测试

3.2.1 MongoShake参数设置

修改MongoShake collector.config，部分参数如下

# --------------------------- incrmental sync configuration ---------------------------
# fetch method:
# oplog: fetch oplog from source mongodb (default)
# change_stream: use change to receive change event from source mongodb, support MongoDB >= 4.0.
# we recommand to use change_stream if possible.
incr_sync.mongo_fetch_method = change_stream
# 同步模式，all表示全量+增量同步，full表示全量同步，incr表示增量同步。
sync_mode = incr
# hash的方式，id表示按文档hash，collection表示按表hash，auto表示自动选择hash类型。
# 如果没有索引建议选择id达到非常高的同步性能，反之请选择collection。
incr_sync.shard_key = auto
# 内部发送（写目的DB）的worker数目，如果机器性能足够，可以提高worker个数。
incr_sync.worker = 40
# batch_queue_size：每个worker线程的队列长度，worker线程从此队列取任务
# batching_max_size：一次分发给worker的任务最多包含多少个文档
# buffer_capacity：PendingQueue队列中一个buffer至少包含的文档个数，进行序列化
incr_sync.worker.batch_queue_size = 64
incr_sync.adaptive.batching_max_size = 1024
incr_sync.fetcher.buffer_capacity = 256
# 增量阶段写入端是否启用majority write
incr_sync.executor.majority_enable = true

3.2.2 MongoShake测试结果

在源端MongoDB分片集群开启balancer的情况下，MongoShake只支持从Mongos上获取ChangeStream来进行增量同步的方案。

迁移期间MongoShake所在ECS实例的cpu、内存、内网带宽等无瓶颈

通过目的端MongoDB实例的监控信息可得：

MongoShake 上述增量迁移过程中 Mongos1的QPS在28000个/s左右，Mongos2上未发现增量迁移写入流量，总迁移速度约为28000个/s。

Mongos1: 平均ops为28000个/s

Mongos2: 未发现增量迁移写入流量

3.3 DTS增量迁移测试

DTS采用直接拉取oplog的方式进行增量同步，为源端每个shard创建对应的分布式子任务进行协同，使得增量迁移期间源端MongoDB分片集群可开启balancer。

使用2xlarge规格链路进行增量迁移测试，40个线程写目的端。通过目的端MongoDB实例的监控信息可得：上述增量迁移过程中 Mongos1和Mongos2上的QPS均在23000个/s左右，总迁移速度约为46000个/s。

Mongos1: 平均ops为23000个/s

Mongos2：平均ops为23000个/s

4.小结

三、总结

本文聚焦DTS MongoDB->MongoDB 和 MongoShake 数据同步的性能，分别针对副本集/分片集群架构、单表/多表、全量/增量同步进行性能的对比，整体上DTS的同步性能优于MongoShake，具体结果如下：

四 、快来关注

1. 数据传输服务（Data Transmission Service，简称DTS）支持关系型数据库、NoSQL、大数据（OLAP）等数据源，集数据迁移、订阅、实时同步、校验功能于一体，能够解决公共云、混合云场景下，远距离、秒级异步数据传输难题。其底层基础设施采用阿里双11异地多活架构，为数千下游应用提供实时数据流，已在线上稳定运行7年之久，是一款沉淀了丰富实践经验的可靠产品。点击了解更多DTS相关信息
   
2. 欢迎加入钉群讨论交流：