MongoShake最佳實踐

我們在去年開源了 MongoShake ，可以用于MongoDB的資料同步，滿足使用者多種需求，整體介紹文檔參考這裡。今天，我來主要介紹一下MongoShake的最佳實踐，主要涵蓋以下幾部分：

從MongoDB副本集同步到MongoDB副本集
從MongoDB副本集同步到MongoDB叢集版
從MongoDB叢集版同步到MongoDB叢集版
從MongoDB副本集同步到kafka通道
雲上MongoDB副本集的雙向同步
配置檔案參數解釋
其他問題Q&A說明。這部分我主要列了目前使用者存在的主流問題，這些問題在github上的wiki上都已經有說明，不過是英文的： https://github.com/alibaba/MongoShake/wiki/FAQ 。

MongoShake的下載下傳路徑我們提供在了

，點選可以下載下傳不同版本的mongoshake：

mongo-shake-x.y.z.tar.gz

。目前MongoShake提供了配置檔案啟動的方式，啟動的指令行：

./collector.linux -conf=collector.conf

，不同的平台請選擇不同的二進制檔案，如windows下是collector.windows。

0. 配置檔案

所有的參數都列在了配置檔案裡面，目前配置檔案較多，使用者可能比較困惑，正常情況下使用者預設就行，修改的參數隻有少部分。

MongoShake支援MongoDB副本集和叢集版的互相同步，還可以同步到kafka模式，下面列出了幾種常見的同步模式。

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:[email protected]:1001,10.2.2.2:2002,10.3.3.3:3003 #源端連接配接串資訊，逗号分隔不同的mongod
sync_mode = all # all 表示全量+增量，full表示僅全量，incr表示僅增量
incr_sync.tunnel.address = mongodb://username:[email protected]: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:[email protected]:1001,10.2.2.2:2002,10.3.3.3:3003 #源端連接配接串資訊，逗号分隔不同的mongod
sync_mode = all # all 表示全量+增量，document表示僅全量，oplog表示僅增量
incr_sync.tunnel.address = mongodb://username:[email protected]: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:[email protected]:1001,10.1.1.2:2002,10.1.1.3:3003;mongodb://username2:[email protected]:1001,10.2.2.2:2002,10.2.2.3:3003 #源端連接配接串資訊，逗号分隔同一個shard不同的mongod，分号分隔不同的shard。
mongo_cs_url = mongodb://username1:[email protected]:5555,10.5.5.6:5556 # 如果源端是sharding，此處需要配置源端sharding的cs的位址
mongo_s_url = mongodb://username_s:[email protected]:1010 # 如果希望采用change stream拉取，則還需要配置mongos的位址，一個就夠了
sync_mode = all # all 表示全量+增量，document表示僅全量，oplog表示僅增量
incr_sync.tunnel.address = mongodb://username:[email protected]: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:[email protected]:1001,10.2.2.2:2002,10.3.3.3:3003 #源端連接配接串資訊，逗号分隔不同的mongod
sync_mode = oplog # 如果目的端不是mongodb，僅支援增量同步模式
incr_sync.tunnel = kafka
incr_sync.tunnel.address = [email protected]:6379
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: 如果目的庫表存在，是否先删除目的庫再進行同步。

7. Q&A說明

此處記載幾個常見的問題：

Q. mongoshake是否會同步config, local, admin庫

A: 不會同步。如果使用者一定要把admin的庫同步到别的，那麼可以通過命名轉換功能（配置transform.namespace）把admin庫同步到别的資料庫，同時配置filter.pass.special.db參數：

filter.pass.special.db = admin
transform.namespace = admin.abc:target.abc # 把admin庫下面的abc同步到target庫的abc

Q. 從MongoDB同步到MongoDB發現全量階段目的端MongoDB壓力過大怎麼辦？

A: 使用者可以降低全量同步配置replayer的相關參數，以降低目的端壓力。

Q. 從MongoDB同步到MongoDB發現全量階段同步時間過久，怎麼辦？

A: 使用者可以定位一下看看目的慢的原因是什麼，正常情況下可以提高全量配置replayer的相關參數，以提高全量同步的性能。但也有可能，問題的瓶頸是源端/網絡端/通道cpu，記憶體等壓力過高，都是有可能的。

Q. 到底什麼是checkpoint？

A: checkpoint是記錄增量同步的位點情況，mongoshake的啟動就是根據這個位點來進行的，比如我們已經同步到10點鐘了，那麼這個時候挂掉了，如果沒有位點，那麼隻能從頭開始拉，位點的意義就是在于斷點續傳。在mongoshake中，位點資訊是以64位時間戳來标記的（ts字段，準确的說是32位時間戳+計數），如果發生挂掉重新開機，那麼會從源庫的oplog表中尋找這個位點，并從此開始往後進行同步。

到這裡，使用者可能會問，那麼mongoshake裡面位點是存儲在哪裡的？存儲的位置就是取決于使用者context的相關配置，以副本集為例，checkpoint是存儲于源庫的mongoshake庫的ckpt_default表，其大概内容長這個樣子：

rszz-4.0-2:PRIMARY> use mongoshake
switched to db mongoshake
rszz-4.0-2:PRIMARY> show collections
ckpt_default
rszz-4.0-2:PRIMARY> db.ckpt_default.find()
{ "_id" : ObjectId("3d75e8b872d91278c3be0cc9"), "name" : "rszz-4.0-2", "ckpt" : Timestamp(1566556865, 196096) }

其中ckpt對應的field就是checkpoint。

使用者可能還會問，那麼mongoshake的checkpoint是怎麼存儲的，什麼時候會存儲？答案是，checkpoint是定期存儲的，如果有資料同步，那麼checkpoint的更新會很頻繁（秒級别）；如果沒有資料同步，比如這個時候使用者源端就是沒有寫入，那麼預設是分鐘級别更新一次心跳checkpoint。假設mongoshake資料位點記錄是10:00:00，但是資料已經同步到10:00:10，這個時候mongoshake挂掉了，下次會從10:00:00重新開始同步，由于DML本身具有幂等性，資料的重複回放是OK的。那麼使用者可能會問，DDL怎麼辦？參考下面問題。

Q. 如何同步DDL？DDL能保證幂等性嗎？

A: 設定

replayer.dml_only = false

可以同步DDL。DDL不能保證幂等性。在mongoshake裡面，如果發現是DDL語句，會卡一個全局的barrier，讓目前DDL語句同步，然後再卡一個全局barrier，等目前DDL語句同步完成，再接着并發同步接下來的oplog，barrier的卡鎖伴随着checkpoint的強制重新整理。但是由于DDL同步和checkpoint的重新整理并不是一個院子操作，如果使用者恰好在同步DDL完成和checkpoint重新整理之間，程序挂掉了，那麼沒有辦法，重新開機以後肯定會持續報錯，使用者需要手動運維解決，比如跳過這個DDL，或者目的端近些一個反向操作（原來DDL是建庫，需要進行删除庫；原來是建索引，需要進行删索引操作）。但是，這個機率很低。

Q. mongoshake碰到同步出錯的情況，會跳過這個錯誤，繼續同步嗎？

A: 不會，會持續報錯，使用者需要關注日志的運作情況。

Q. 除了日志如何知道目前mongoshake的同步情況？

A: mongoshake有提供restful接口，可以監控mongoshake的内部同步情況，參考

wiki: How to monitor the MongoShake

，正常情況下lsn_ckpt.time會一直增長，如果這個數值不增長了，那麼就表示同步出現問題了。

Q. 全量同步是否會同步索引？

A: 2.1以下版本，目前會同步索引，先是全量同步，全量同步完畢後會同步索引，後面版本我們會考慮加入開關，全量同步完畢可以選擇性是否同步索引。接着是增量同步，增量同步階段，設定

replayer.dml_only = true

不會同步索引，false的話會同步索引。

Q. 選擇all模式同步，如果挂掉以後重新開機，是否會重新發起一次全量同步？進入增量階段，是否要修改為oplog增量模式，然後再重新開機？

A: 顯示全量同步，全量同步完畢會寫入checkpoint，然後進行增量同步，增量同步階段會不斷更新checkpoint。對于mongoshake來說，是否進行全量同步取決于checkpoint，如果checkpoint存在且合法，那麼隻會進入增量同步；否則，會先進行全量同步，然後是增量。那麼checkpoint怎麼知道是存在的，就是context裡面的參數的位置決定；那麼怎麼知道checkpoint是合法的？如果最老的oplog的時間位點小于checkpoint，那麼就是合法的，證明增量能接上。

Q. 我的資料量很大，幾百G/幾個T/幾十個T，all模式同步有什麼要注意的地方？

A: 最需要注意的就是，全量同步完畢，增量能否接上。mongoshake是根據oplog的位點來決定全量結束是否要進行增量，如果全量花了2天，但是oplog表隻能存儲1天，那麼全量同步完畢以後mongoshake就會報錯退出，這是因為增量的位點丢失了。是以如果全量很大，同步時間勢必很久，那麼就需要放大oplog的表的大小以維持增量oplog。

其他還需要注意的就是壓力問題，如何選取适當的參數應對對于源端/目的端的壓力，減少對業務的影響。使用者也可以決定從一個hidden結點，護着一個獨立的secondary結點進行拉取。

Q. 同步出現xxx錯誤，怎麼辦？

A: 這個我在github的wiki基本都有總結，可以先看下github的

faq

。如果裡面沒有提到的，可以送出issue進行回報。

Q. 目前mongoshake有校驗工具嗎?

A: 目前script下面有個comparison.py的全量校驗腳本校驗資料的一緻性。那麼如何進行增量的校驗呢？目前還沒有，敬請期待。

Q. sharding模式同步有啥需要注意的嗎？

A: 如果源端是sharding，那麼目前需要關閉balancer，在v2.1非穩定版本裡面，我們支援了不關閉balancer，使用者可以進行試用，但是目前這個是非穩定版本。此外源端如果是sharding，DDL的同步（replayer.dml_only = false）不能開啟。

Q. 我把資料寫入到kafka，怎麼從kafka中拉取資料，發現直接拉取出現亂碼

A: 請先看下

wiki: https://github.com/alibaba/MongoShake/wiki/FAQ#q-how-to-connect-to-different-tunnel-except-direct

kafka通道中的資料是有控制資訊在裡面的，使用者不好直接剝離，可以用receiver進行對接。目前receiver對接後所有oplog都列印到控制台，使用者也可以自己修改receiver代碼，對接下遊服務。當然，這個需要使用者懂golang語言，這可能有一些學習成本在裡面。後面我們會看看如何以更好的形式對接給使用者。

Q. 我發現mongoshake建立以後對源端性能有所影響，出現了一些慢查詢。

A: 目前全量階段由于是掃表等操作，會對源端産生一些影響。增量階段，正常情況影響很小，因為目前oplog找對應ts是二分模式。

Q. 是否可以把一個MongoDB同步到多個MongoDB？

A: 可以，需要啟動多個mongoshake。對于不同的mongoshake，可以修改寫入不同地方的checkpoint即可（比如表名修改一下），或者直接修改不同的collector.id。此外，需要注意的是，同一個目錄下啟動多個mongoshake，如果collector.id相同，是無法啟動成功的。參考：

https://github.com/alibaba/MongoShake/wiki/FAQ#q-how-to-sync-data-from-one-mongodb-to-several-target-mongodb