天天看點
傳回

dynamodb遷移的工具：dynamo-shake（又名nimo-shake）

DynamoShake基本功能

DynamoDB支援全量和增量的同步，程序啟動後會先進行全量同步，全量同步結束後進入增量同步的階段。

全量同步分為資料同步和索引同步兩部分，資料同步用于同步資料，資料同步結束後将會進行索引的同步，索引同步會同步預設的primary key，使用者自建的索引GSI如果MongoDB是副本集支援，叢集版目前暫時不支援同步。

增量同步隻同步資料，不同步增量同步過程中産生的索引。

此外，全量和增量同步階段不支援對原來的庫表進行DDL操作，比如删表，建表，建索引等。

dynamodb遷移的工具：dynamo-shake（又名nimo-shake）

斷點續傳

全量同步不支援斷點續傳功能，增量同步支援斷點續傳，也就是說如果增量斷開了，一定時間内恢複是可以隻進行增量的斷點續傳。但在某些情況下，比如斷開的時間過久，或者之前的位點（參考下文）丢失，那麼都會導緻重新觸發全量同步。

同步資料

所有源端的表會寫入到目的的一個庫（預設是dynamo-shake）的不同表中，比如使用者有table1，table2，那麼同步完後，目的端會有個dynamo-shake的庫，庫裡面有table1和table2的表。

在原生的dynamodb中，協定是包裹了一層類型字段，其格式是“key: type: value”格式，例如使用者插入了一條{hello: 1}，那麼dynamodb接口擷取的資料是{"hello": {"N": 1}}的格式。

Dynamo所有的資料類型：

  • String
  • Binary
  • Number
  • StringSet
  • NumberSet
  • BinarySet
  • Map
  • List
  • Boolean
  • Null

那麼我們提供2種轉換方式，raw和change，其中raw就是按照裸的dynamodb接口擷取的資料寫入：

rszz-4.0-2:PRIMARY> use dynamo-shake
switched to db dynamo-shake
rszz-4.0-2:PRIMARY> db.zhuzhao.find()
{ "_id" : ObjectId("5d43f8f8c51d73b1ba2cd845"), "aaa" : { "L" : [ { "S" : "aa1" }, { "N" : "1234" } ] }, "hello_world" : { "S" : "f2" } }
{ "_id" : ObjectId("5d43f8f8c51d73b1ba2cd847"), "aaa" : { "N" : "222" }, "qqq" : { "SS" : [ "h1", "h2" ] }, "hello_world" : { "S" : "yyyyyyyyyyy" }, "test" : { "S" : "aaa" } }
{ "_id" : ObjectId("5d43f8f8c51d73b1ba2cd849"), "aaa" : { "L" : [ { "N" : "0" }, { "N" : "1" }, { "N" : "2" } ] }, "hello_world" : { "S" : "測試中文" } }

change表示剝離類型字段：

rszz-4.0-2:PRIMARY> use dynamo-shake
switched to db dynamo-shake
rszz-4.0-2:PRIMARY> db.zhuzhao.find()
{ "_id" : ObjectId("5d43f8f8c51d73b1ba2cd845"), "aaa" : [ "aa1", 1234 ] , "hello_world" : "f2" }
{ "_id" : ObjectId("5d43f8f8c51d73b1ba2cd847"), "aaa" : 222, "qqq" : [ "h1", "h2" ] , "hello_world" : "yyyyyyyyyyy", "test" : "aaa" }
{ "_id" : ObjectId("5d43f8f8c51d73b1ba2cd849"), "aaa" : [ 0, 1, 2 ], "hello_world" : "測試中文" }

使用者可以根據自己的需求制定自己的同步類型。

位點

增量的斷點續傳是根據位點來實作的，預設的位點是寫入到目的MongoDB中，庫名是dynamo-shake-checkpoint。每個表都會記錄一個checkpoint的表，同樣還會有一個status_table表記錄目前是全量同步還是增量同步階段。

rszz-4.0-2:PRIMARY> use dynamo-shake42-checkpoint
switched to db dynamo-shake42-checkpoint
rszz-4.0-2:PRIMARY> show collections
status_table
zz_incr0
zz_incr1
rszz-4.0-2:PRIMARY>
rszz-4.0-2:PRIMARY>
rszz-4.0-2:PRIMARY> db.status_table.find()
{ "_id" : ObjectId("5d6e0ef77e592206a8c86bfd"), "key" : "status_key", "status_value" : "incr_sync" }
rszz-4.0-2:PRIMARY> db.zz_incr0.find()
{ "_id" : ObjectId("5d6e0ef17e592206a8c8643a"), "shard_id" : "shardId-00000001567391596311-61ca009c", "father_id" : "shardId-00000001567375527511-6a3ba193", "seq_num" : "", "status" : "no need to process", "worker_id" : "unknown-worker", "iterator_type" : "AT_SEQUENCE_NUMBER", "shard_it" : "", "update_date" : "" }
{ "_id" : ObjectId("5d6e0ef17e592206a8c8644c"), "shard_id" : "shardId-00000001567406847810-f5b6578b", "father_id" : "shardId-00000001567391596311-61ca009c", "seq_num" : "", "status" : "no need to process", "worker_id" : "unknown-worker", "iterator_type" : "AT_SEQUENCE_NUMBER", "shard_it" : "", "update_date" : "" }
{ "_id" : ObjectId("5d6e0ef17e592206a8c86456"), "shard_id" : "shardId-00000001567422218995-fe7104bc", "father_id" : "shardId-00000001567406847810-f5b6578b", "seq_num" : "", "status" : "no need to process", "worker_id" : "unknown-worker", "iterator_type" : "AT_SEQUENCE_NUMBER", "shard_it" : "", "update_date" : "" }
{ "_id" : ObjectId("5d6e0ef17e592206a8c86460"), "shard_id" : "shardId-00000001567438304561-d3dc6f28", "father_id" : "shardId-00000001567422218995-fe7104bc", "seq_num" : "", "status" : "no need to process", "worker_id" : "unknown-worker", "iterator_type" : "AT_SEQUENCE_NUMBER", "shard_it" : "", "update_date" : "" }
{ "_id" : ObjectId("5d6e0ef17e592206a8c8646a"), "shard_id" : "shardId-00000001567452243581-ed601f96", "father_id" : "shardId-00000001567438304561-d3dc6f28", "seq_num" : "", "status" : "no need to process", "worker_id" : "unknown-worker", "iterator_type" : "AT_SEQUENCE_NUMBER", "shard_it" : "", "update_date" : "" }
{ "_id" : ObjectId("5d6e0ef17e592206a8c86474"), "shard_id" : "shardId-00000001567466737539-cc721900", "father_id" : "shardId-00000001567452243581-ed601f96", "seq_num" : "", "status" : "no need to process", "worker_id" : "unknown-worker", "iterator_type" : "AT_SEQUENCE_NUMBER", "shard_it" : "", "update_date" : "" }
{ "_id" : ObjectId("5d6e0ef27e592206a8c8647e"), "shard_id" : "shardId-00000001567481807517-935745a3", "father_id" : "shardId-00000001567466737539-cc721900", "seq_num" : "", "status" : "done", "worker_id" : "unknown-worker", "iterator_type" : "LATEST", "shard_it" : "arn:aws:dynamodb:us-east-2:240770237302:table/zz_incr0/stream/2019-08-27T08:23:51.043|1|AAAAAAAAAAGsTOg0+3HY+yzzD1cTzc7TPXi/iBi7sA5Q6SGSoaAJ2gz2deQu5aPRW/flYK0pG9ZUvmCfWqe1A5usMFWfVvd+yubMwWSHfV2IPVs36TaQnqpMvsywll/x7IVlCgmsjr6jStyonbuHlUYwKtUSq8t0tFvAQXtKi0zzS25fQpITy/nIb2y/FLppcbV/iZ+ae1ujgWGRoojhJ0FiYPhmbrR5ZBY2dKwEpok+QeYMfF3cEOkA4iFeuqtboUMgVqBh0zUn87iyTFRd6Xm49PwWZHDqtj/jtpdFn0CPoQPj2ilapjh9lYq/ArXMai5DUHJ7xnmtSITsyzUHakhYyIRXQqF2UWbDK3F7+Bx5d4rub1d4S2yqNUYA2eZ5CySeQz7CgvzaZT391axoqKUjjPpdUsm05zS003cDDwrzxmLnFi0/mtoJdGoO/FX9LXuvk8G3hgsDXBLSyTggRE0YM+feER8hPgjRBqbfubhdjUxR+VazwjcVO3pzt2nIkyKPStPXJZIf4cjCagTxQpC/UPMtcwWNo2gQjM2XSkWpj7DGS2E4738biV3mtKXGUXtMFVecxTL/qXy2qpLgy4dD3AG0Z7pE+eJ9qP5YRE6pxQeDlgbERg==", "update_date" : "" }
{ "_id" : ObjectId("5d6e1d807e592206a8c9a102"), "shard_id" : "shardId-00000001567497561747-03819eba", "father_id" : "shardId-00000001567481807517-935745a3", "seq_num" : "39136900000000000325557205", "status" : "in processing", "worker_id" : "unknown", "iterator_type" : "AT_SEQUENCE_NUMBER", "shard_it" : "arn:aws:dynamodb:us-east-2:240770237302:table/zz_incr0/stream/2019-08-27T08:23:51.043|1|AAAAAAAAAAFw/qdbPLjsXMlPalnhh55koia44yz6A1W2uwUyu/MzRUhaaqnI0gPM8ebVgy7dW7dDWLTh/WXYyDNNyXR3Hvk01IfEKDf+FSLMNvh2iELdrO5tRoLtZI2fxxrPZvudRc3KShX0Pvqy2YYwl4nlBR6QezHTWx5H2AU22MGPTx8aMRbjUgPwvgEExRgdzfhG6G9gkc7C71fIc98azwpSm/IW+mV/h/doFndme47k2v8g0GNJvgLSoET7HdJYH3XFdqh4QVDIP4sbz8X1cpN3y8AlT7Muk2/yXOdNeTL6tApuonCrUpJME9/qyBYQVI5dsAHnAWaP2Te3EAvz3ao7oNdnA8O6uz5VF9zFdN1OUHWM40kLUsX4sHve7McEzFLgf4NL1WTAnPN13cFhEm9BS8M7tiJqZ0OzgkbF1AWfq+xg/O6c57/Vvx/G/75DZ8XcWIABgGNkWBET/vLDrgjJQ0PUZJZKNmmbgKKTyHgSl4YOXNEeyH7l6atuc2WaREDjbf7lnQO5No11sz4g3O+AreBcpGVhdZNhGGcrG/wduPYEZfg2hG1sfYiSAM8GipUPMA0PM7JPIJmqCaY90JxRcI1By24tpp9Th35/5rLTGPYJZA==", "update_date" : "" }

其中status_table表中

"status_value" : "incr_sync"

表示進入了增量階段。增量的每個shard都會記錄一個checkpoint，關于具體shard分裂的規則可以參考dynamodb的guan'fa官方文檔。下面是增量表checkpoint的各個字段的說明：

  • _id

     ：mongodb自帶主鍵id

  • shard_id

     ：shard的id，每個shard有一個唯一的id

  • father_id

     ：父shard的id，shard可能有一個父shard。

  • seq_num

     : 目前處理到的shard内部的sequence number，這個是主要的位點資訊。

  • status

     : 目前同步的階段，一共有以下幾個狀态：
    • "not process": 未處理
    • "no need to process"： 沒有必要處理
    • "prepare stage"：準備處理
    • "in processing": 進行中
    • "wait father finish"：等待父節點處理完畢再進行處理
    • "done"： 處理完畢

  • worker_id

     ：處理的worker id，目前暫未啟用

  • iterator_type

     ：shard的周遊方式

  • shard_it

     ：shard的疊代器位址，次要位點資訊。

  • update_date

     ：checkpoint更新的時間戳

索引

根據預設的primary key建立一個唯一索引，并且根據partition key建立shard key。使用者自己的索引gsi目前不進行建立。

DynamoShake内部架構

本小節主要介紹DynamoShake的部分架構細節

全量同步

下圖是基本的一個table的資料同步架構圖（dynamo-shake會啟動多個并發線程tableSyncer進行拉取，使用者可控并發度），fetcher線程從源端dynamodb拉取資料後将資料推入隊列，緊接着parser線程從隊列中拿取資料并進行解析（dynamo協定轉bson），executor負責聚合部分資料并寫入mongodb。

dynamodb遷移的工具：dynamo-shake（又名nimo-shake）
  • fetcher。目前fetcher線程隻有1個，用的是協定轉換驅動是aws提供的driver。fetcher的原理是調用driver進行批量抓取源庫的資料，抓到了就塞入隊列中，直到抓完目前table的所有資料。fetcher單獨分離出來主要是出于網絡IO考慮的，目前拉取受網絡影響，會比較慢。
  • parser。parser可以啟動多個，預設目前是2個，使用者可以通過

    FullDocumentParser

     進行控制。其主要就是從隊列中讀取資料，并解析成bson結構。parser解析後，資料按條寫入executor的隊列。parser線程單獨獨立出來主要是出于解析比較耗CPU資源考慮。
  • executor。executor也可以啟動多個，預設目前是4個，使用者可以通過

    FullDocumentConcurrency

    進行控制。executor從隊列中拉取，并進行batch聚合（聚合上限16MB，總條數1024）後寫入目的mongodb。

    目前所有表的資料寫入完後，tableSyncer将會退出。

增量同步

增量整體架構如下：

dynamodb遷移的工具：dynamo-shake（又名nimo-shake）

Fetcher線程負責感覺stream中shard的變化，Manager負責進行消息的通知，或者建立新的Dispatcher進行消息的處理，一個shard對應一個Dispatcher。Dispatcher從源端拉取增量資料，并通過Batcher進行資料解析和打包整合，然後通過executor進行寫入到MongoDB，同時會更新checkpoint。另外，如果是斷點續傳，那麼Dispatcher會從舊的checkpoint位點開始拉取，而不是從頭開始拉。

DynamoShake的使用

啟動：

./dynamo-shake -conf=dynamo-shake.conf

，配置參數在dynamo-shake.conf中指定，以下是各個參數的意義：

  • id: 修改會影響MongoDB上目的庫的名字
  • log.file：日志檔案，不配置将列印到标準輸出
  • log.level: log級别。推薦預設。
  • log.buffer: 列印是否帶緩存。推薦預設。
  • system_profile：列印内部堆棧的端口号。推薦預設。
  • http_profile：暫未啟用
  • sync_mode：同步模式，all表示全量+增量，full表示僅全量，incr表示僅增量（目前不支援）
  • source.access_key_id: dynamodb連接配接配置參數
  • source.secret_access_key: dynamodb連接配接配置參數
  • source.session_token: dynamodb連接配接配置參數，沒有可以留白
  • source.region: dynamodb連接配接配置參數
  • filter.collection.white：過濾白名單，隻同步指定的表
  • filter.collection.black：過濾黑名單，不通過指定的表。
  • qps.full：全量階段限速，1秒鐘發送多少個請求
  • qps.full.batch_num：全量階段限速，1個請求最多包括多少個item。
  • qps.incr：增量階段限速，1秒鐘發送多少個請求
  • qps.incr.batch_num：增量階段限速，1個請求最多包括多少個item。
  • target.type：目的端配置，目前僅支援mongodb
  • target.address: 目的端mongodb的連接配接串位址。
  • target.mongodb.type: mongodb是replica還是sharding
  • target.mongodb.exist：如果目的庫同名表存在，執行什麼行為。drop表示删除，rename表示重命名，留白表示不處理。
  • full.concurrency：全量同步的線程個數，1個線程對應1個表
  • full.document.concurrency：全量同步1個表内的并發個數。
  • full.document.parser：1個表内parser線程的個數
  • full.enable_index.primary：是否同步dynamodb的primary key。
  • full.enable_index.user：是否同步使用者自建的索引，目前不支援
  • convert.type：寫入的模式，raw表示裸寫入, change表示解析類型字段後寫入，參考上述文檔。
  • increase.concurrency：增量同步并發參數，1次最多抓取的shard個數
  • checkpoint.address = checkpont的存儲位址，預設不配置與目的庫一緻。
  • checkpoint.db = checkpoint寫入的db的名字，預設是$db-checkpoint。

DynamoFullCheck

DynamoFullCheck是一個用于校驗DynamoDB和MongoDB資料是否一緻的工具，目前僅支援全量校驗，不支援增量，也就是說，如果增量同步階段，那麼源和目的是不一緻的。

DynamoFullCheck隻支援單向校驗，也就是校驗DynamoDB的資料是否是MongoDB的子集，反向不進行校驗。

另外，還支援抽樣校驗，支援隻校驗感興趣的表。

校驗主要分為以下幾部分：

  • 輪廓校驗。首先，校驗兩邊的表中數目是否一緻；接着，校驗索引是否一緻（目前沒做索引校驗）。注意，如果表中數目不一緻，将會直接退出，不會進行後續的校驗。

  • 精确校驗。精确校驗資料，原理是從源端拉取資料并解析，如果有唯一索引，那麼根據唯一索引查找MongoDB的doc，并對比一緻性；如果沒有唯一索引，那麼會根據整個doc在MongoDB中進行查找（比較重）。

    抽樣原理：

精确校驗的時候，如果啟用抽樣，那麼會對每個doc進行抽樣，判斷目前doc是否需要抽樣。原理比較簡單，比如按30%抽樣，那麼再0~100中産生一個随機數，如果是0~30的就校驗，反之不校驗。

DynamoFullCheck由于從源DynamoDB拉取也需要經過fetch，parse階段，是以一定程度上，該部分代碼複用了DynamoShake，不同的是DynamoFullCheck内部各個fetcher, parser, executor線程并發度都是1。

使用參數

full-check參數稍微簡單點，直接用的指令行注入，例如：

./dynamo-full-check --sourceAccessKeyID=BUIASOISUJPYS5OP3P5Q --sourceSecretAccessKey=TwWV9reJCrZhHKSYfqtTaFHW0qRPvjXb3m8TYHMe --sourceRegion=ap-east-1 -t="10.1.1.1:30441" --sample=300

 
Usage:
  dynamo-full-check.darwin [OPTIONS]

Application Options:
  -i, --id=                    target database collection name (default: dynamo-shake)
  -l, --logLevel=
  -s, --sourceAccessKeyID=     dynamodb source access key id
      --sourceSecretAccessKey= dynamodb source secret access key
      --sourceSessionToken=    dynamodb source session token
      --sourceRegion=          dynamodb source region
      --qpsFull=               qps of scan command, default is 10000
      --qpsFullBatchNum=       batch number in each scan command, default is 128
  -t, --targetAddress=         mongodb target address
  -d, --diffOutputFile=        diff output file name (default: dynamo-full-check-diff)
  -p, --parallel=              how many threads used to compare, default is 16 (default: 16)
  -e, --sample=                comparison sample number for each table, 0 means disable (default: 1000)
      --filterCollectionWhite= only compare the given tables, split by ';'
      --filterCollectionBlack= do not compare the given tables, split by ';'
  -c, --convertType=           convert type (default: raw)
  -v, --version                print version

Help Options:
  -h, --help                   Show this help message
存儲 sql 緩存 NoSQL mongodb 索引 nosql列存儲 nosql資料存儲 nosql資料存儲分析概覽 nosql存儲 ssd緩存存儲
上一篇: MacOS 等機器配置 Docker 啟用 Remote API
下一篇: 初識-----深入了解

繼續閱讀