MongoShake最佳實踐

 

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

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

MongoShake的下載路徑我們提供在了github上，點擊可以下載不同版本的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: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@10.5.5.5:5005, 10.6.6.6:6006, 10.7.7.7:7007 #目的端連接串信息，逗號分隔不同的mongod

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可以做負載均衡寫入。

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。目的端是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。
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可以做負載均衡寫入。
context.storage.url = mongodb://username1:password1@10.5.5.5:5555,10.5.5.6:5556 # 如果源端是sharding，此處需要配置源端sharding的cs的地址

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 = oplog # 如果目的端不是mongodb，僅支持增量同步模式
tunnel.type = kafka
tunnel.address = test@50.1.1.1:6379

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. 部分配置文件參數解釋

具體請查看配置文件的注釋，此處只做簡單解釋

• 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參數：

1. filter.pass.special.db = admin
2. 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: 會同步索引，顯示全量同步，全量同步完畢后會同步索引，接着是增量同步。

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

作者：燭昭