DynamoShake怎么从dynamodb迁移到mongodb

发布时间:2021-07-19 11:27:30 作者:Leah
来源:亿速云 阅读:169

DynamoShake怎么从dynamodb迁移到mongodb,相信很多没有经验的人对此束手无策,为此本文总结了问题出现的原因和解决方法,通过这篇文章希望你能解决这个问题。

DynamoShake基本功能

DynamoDB支持全量和增量的同步,进程启动后会先进行全量同步,全量同步结束后进入增量同步的阶段。
全量同步分为数据同步和索引同步两部分,数据同步用于同步数据,数据同步结束后将会进行索引的同步,索引同步会同步默认的primary key,用户自建的索引GSI如果MongoDB是副本集支持,集群版目前暂时不支持同步。
增量同步只同步数据,不同步增量同步过程中产生的索引。
此外,全量和增量同步阶段不支持对原来的库表进行DDL操作,比如删表,建表,建索引等。

DynamoShake怎么从dynamodb迁移到mongodb

断点续传

全量同步不支持断点续传功能,增量同步支持断点续传,也就是说如果增量断开了,一定时间内恢复是可以只进行增量的断点续传。但在某些情况下,比如断开的时间过久,或者之前的位点(参考下文)丢失,那么都会导致重新触发全量同步。

同步数据

所有源端的表会写入到目的的一个库(默认是dynamo-shake)的不同表中,比如用户有table1,table2,那么同步完后,目的端会有个dynamo-shake的库,库里面有table1和table2的表。
在原生的dynamodb中,协议是包裹了一层类型字段,其格式是“key: type: value”格式,例如用户插入了一条{hello: 1},那么dynamodb接口获取的数据是{"hello": {"N": 1}}的格式。
Dynamo所有的数据类型:

那么我们提供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" : [ "h2", "h3" ] }, "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" : [ "h2", "h3" ] , "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/qdbPLjsXMlPalnhh65koia44yz6A1W2uwUyu/MzRUhaaqnI0gPM8ebVgy7dW7dDWLTh/WXYyDNNyXR3Hvk01IfEKDf+FSLMNvh3iELdrO5tRoLtZI2fxxrPZvudRc3KShX0Pvqy2YYwl4nlBR6QezHTWx5H2AU22MGPTx8aMRbjUgPwvgEExRgdzfhG6G9gkc7C71fIc98azwpSm/IW+mV/h/doFndme47k2v8g0GNJvgLSoET7HdJYH3XFdqh5QVDIP4sbz8X1cpN3y8AlT7Muk2/yXOdNeTL6tApuonCrUpJME9/qyBYQVI5dsAHnAWaP2Te3EAvz3ao7oNdnA8O6uz5VF9zFdN1OUHWM40kLUsX4sHve7McEzFLgf4NL1WTAnPN13cFhEm9BS8M7tiJqZ0OzgkbF1AWfq+xg/O6c57/Vvx/G/75DZ8XcWIABgGNkWBET/vLDrgjJQ0PUZJZKNmmbgKKTyHgSl4YOXNEeyH7l6atuc2WaREDjbf7lnQO5No11sz4g3O+AreBcpGVhdZNhGGcrG/wduPYEZfg2hG1sfYiSAM8GipUPMA0PM7JPIJmqCaY90JxRcI1By24tpp9Th45/5rLTGPYJZA==", "update_date" : "" }

其中status_table表中"status_value" : "incr_sync"表示进入了增量阶段。增量的每个shard都会记录一个checkpoint,关于具体shard分裂的规则可以参考dynamodb的guan'fa官方文档。下面是增量表checkpoint的各个字段的说明:

索引

根据默认的primary key创建一个唯一索引,并且根据partition key创建shard key。用户自己的索引gsi目前不进行创建。

DynamoShake内部架构

本小节主要介绍DynamoShake的部分架构细节

全量同步

下图是基本的一个table的数据同步架构图(dynamo-shake会启动多个并发线程tableSyncer进行拉取,用户可控并发度),fetcher线程从源端dynamodb拉取数据后将数据推入队列,紧接着parser线程从队列中拿取数据并进行解析(dynamo协议转bson),executor负责聚合部分数据并写入mongodb。

DynamoShake怎么从dynamodb迁移到mongodb

增量同步

增量整体架构如下:

DynamoShake怎么从dynamodb迁移到mongodb


Fetcher线程负责感知stream中shard的变化,Manager负责进行消息的通知,或者创建新的Dispatcher进行消息的处理,一个shard对应一个Dispatcher。Dispatcher从源端拉取增量数据,并通过Batcher进行数据解析和打包整合,然后通过executor进行写入到MongoDB,同时会更新checkpoint。另外,如果是断点续传,那么Dispatcher会从旧的checkpoint位点开始拉取,而不是从头开始拉。

DynamoShake的使用

启动:./dynamo-shake -conf=dynamo-shake.conf,配置参数在dynamo-shake.conf中指定,以下是各个参数的意义:

DynamoFullCheck

DynamoFullCheck是一个用于校验DynamoDB和MongoDB数据是否一致的工具,目前仅支持全量校验,不支持增量,也就是说,如果增量同步阶段,那么源和目的是不一致的。
DynamoFullCheck只支持单向校验,也就是校验DynamoDB的数据是否是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

看完上述内容,你们掌握DynamoShake怎么从dynamodb迁移到mongodb的方法了吗?如果还想学到更多技能或想了解更多相关内容,欢迎关注亿速云行业资讯频道,感谢各位的阅读!

推荐阅读:
  1. 如何从BasicFile迁移到SecureFile存储(二)
  2. 从 golang flag 迁移到 cmdr

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

dynamoshake dynamodb mongodb

上一篇:nodejs中怎么访问MongoDB

下一篇:python中的EasyOCR库是什么

相关阅读

您好,登录后才能下订单哦!

密码登录
登录注册
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》