如何将DynamoDB的数据增量迁移到表格存储

Amazon DynamoDB是一个彻底托管的NoSQL数据库服务，能够提供快速的、可预期的性能，而且能够实现无缝扩展。因为DynamoDB并能够根据实际需求对表进行扩展和收缩，这个过程既不须要中止对外服务，也不会下降服务性能，一经推出就收到了广大AWS用户的欢迎。

一样，表格存储是构建在阿里云飞天分布式系统之上的分布式NoSQL数据库服务。做为同 DynamoDB 很是类似的 __云NoSQL数据库服务__，表格存储的自动负载均衡机制能够自动对表进行扩展，实现数据规模与访问并发上的无缝扩展，提供海量结构化数据的存储和实时访问。

表格存储可使用户把操做和扩展分布式数据库的沉重负担，交给阿里云来处理，这样，用户就不须要担忧硬件配置、磁盘故障、机器故障、软件安装和升级等工做，能够更专一到业务逻辑中去。

今天，就给你们介绍如何将DynamoDB的数据增量迁移到表格存储。

数据转换规则

表格存储支持的数据格式有：

• String - 可为空，可为主键，为主键列时最大为 1 KB，为属性列时为2MB。
• Integer - 64 bit，整型，可为主键，8 Bytes。
• Binary - 二进制数据，可为空，可为主键，为主键列时最大为 1 KB，为属性列时为2MB。
• Double - 64 bit，Double 类型，8 Bytes。
• Boolean - True/False，布尔类型，1 Byte。

目前 DynamoDB 支持多种数据格式：

• 标量类型 - 标量类型可准确地表示一个值。标量类型包括数字、字符串、二进制、布尔值和 null。
• 文档类型 - 文档类型可表示具备嵌套属性的复杂结构 - 例如您将在 JSON 文档中找到的结构。文档类型包括列表和映射。
• 集类型 - 集类型可表示多个标量值。集类型包括字符串集、数字集和二进制集。

因为DynamoDB支持文档型数据类型，咱们须要将文档型转换为一个String类型或者Binary类型存储到表格存储中，在读取时须要反序列化成Json。

故，从DynamoDB迁移到表格存储时，咱们作以下的数据转换：

DynamoDB类型	数据示例	TableStore对应类型
id (N)	'123'	Integer
level (N)	'2.3'	Double, 不能为主键
afea (NULL)	TRUE	String,空字符串
binary (B)	0x12315	binary
binary_set (BS)	{ 0x123, 0x111 }	binary
bool (BOOL)	TRUE	boolean
list (L)	[ { "S" : "a" }, { "N" : "1" }]	string
map (M)	{ "key1" : { "S" : "value1" }}	string
str (S)	This is test！	string
num_set (NS)	{ 1, 2 }	string
str_set (SS)	{ "a", "b" }	string

增量实现机制

咱们使用DynamoDB的Stream数据流获取DynamoDB表中的增删改操做，将操做同步到表格存储中，为了不环境搭建，将同步程序运行在Lambda 中，流程以下图：

使用Stream数据流中的'eventName'字段来判别数据的增删改操做：

• "INSERT"： 插入数据，对应PutRow

• "MODIFY" ： 修改数据

  "REMOVE"：删除数据，对应 DeleteRow
  • 若是OldImage 与 NewImage的key相同，则为更新数据，对应Update
  • 若OldImage的Key 数量大于 NewImage的Key数量， 则为删除数据，将二者差集的keys作删除，对应Delete

须要特别注意的是：

1. 上述 Stream 中增删改操做转换行为符合业务的指望。
2. 表格存储目前还不支持二级索引，故只能同步主表的数据。
3. DynamoDB 中表的主键同TableStore中的主键保持一致，且数字类型的主键只能为整型。
4. DynamoDB 对单个项目的大小限制为400KB，表格存储中单行虽然没有限制，但一次提交的数据量不能超过4MB。 DynamoDB限制项参考 及 TableStore 限制项参考
5. 若是先进行全量数据迁移，则须要在全量迁移以前开启 Stream。因为 DynamoDB Stream 只能保存最近24小时数据，故全量数据须要在24小时内迁移完成，在全量迁移完成后才能开启 Lambda 的迁移任务。
6. 数据须要保证最终一致性。增量数据在同步时，可能会有对全量数据的重复写入，好比 T0 时刻开启 Stream 并进行全量迁移，T1 时刻完成，那么 T0 到 T1 之间的时间段内的DynamoDB 数据操做会同步写入到表格存储中。

操做过程

1. 在DynamoDB中建立数据表

咱们以表Source为例，主键为user_id(字符串类型)，排序键为action_time(数字)。因为DynamoDB的预留设置会影响读写的并发，故须要注意预留的设置。

2. 开启source 表的Stream

Stream模式须要为： 新旧映像 - 新旧项目的映像

3. 转到Lambda的控制台，建立相关的数据同步函数

实例函数名称为：data-to-table， 运行语言选择为 Python 2.7，使用 lambda-dynamodb-execution-role的角色。

4.关联Lambda的事件源

点击事件源的DynamoDB图标，进行事件源配置，选择 source 数据表批处理大小先选择为10进行小批量验证，在实际运行过程当中建议为100，因为表格存储的Batch操做最大为200条数据，故不能超过200。

5. 配置Lambda的函数。

点击 Lambda的函数图标，进行函数相关的配置。

因为tablestore须要依赖SDK及 protocolbuf等依赖包，咱们按照建立部署程序包 (Python)的方式进行 SDK依赖安装及打包。

使用的函数zip包为：lambda_function.zip 点击下载 能够直接本地上传，也能够先上传到S3。

处理程序入口为默认的 lambda_function.lambda_handler

基本设置中须要将超时事件设置在1分钟以上（考虑到批量提交的延时及网络传输时间）。

6. 配置Lambda的运行变量

在数据导入时，须要 TableStore 实例名、AK等相关信息，咱们可使用一下两种方式：

• 方案一（推荐）：直接在Lambda 中配置相关的环境变量，以下图.
  使用 Lambda的环境变量将使得同一函数代码zip包可以灵活的支持不一样的数据表，而不须要为每一个数据源修改代码包中的配置文件。

参考：Lambda环境变量说明

• 方案二： 也能够打开 lambda_function.zip 修改其中的example_config.py，再打包上传，或者上传以后在控制台上进行修改。
  

配置说明：

环境变量	必选	意义
OTS_ID	是	访问表格存储的AccessKeyId信息
OTS_SECRET	是	访问表格存储的AccessKeySecret信息
OTS_INSTANCE	是	导入的表格存储的实例名称
OTS_ENDPOINT	否	导入的表格存储的域名，若是不存在，则使用默认的实例公网域名
TABLE_NAME	是	导入的表格存储的表名
PRIMARY_KEY	是	导入的表格存储的表的主键信息，须要保证主键顺序，主键名称须要同源表保持一致

特别注意：

1. 相同的变量名称，优先会从Lambda中变量配置中读取，若是不存在，则会从 example_config.py中读取。
2. 因为AK信息表明这资源的访问权限，强烈建议使用只具备表格存储特定资源写权限的子帐号的AK，避免AK泄露带来的风险，使用参考

7. 在表格存储中建立数据表。

在表格存储控制台上建立数据表：__target__，主键为 user_id（字符串）和action_time(整型)。

8. 测试调试。

在lambda控制台上编辑事件源进行调试。

点击右上角的 配置测试事件，输入示例事件的json内容。
咱们准备了两个示例的 Stream示例事件：

1. test_data_put.json 模拟向DynamoDB中插入一条数据的事件，查看文件
2. test_data_update.json 模拟向DynamoDB中更新一条数据的事件，查看文件
3. test_data_update.json 模拟向DynamoDB中删除一条数据的事件，查看文件

咱们将上述三个事件的内容依次保存为putdata、updatedata、deletedata。

保存以后，选择须要使用的事件，点击测试：

执行结果提示成功的话，则在表格存储的 target表中就能够读到以下的测试数据。

依次选择putdata、updatedata和deletedata，会发现表格存储中的数据也会随之更新和删除。

9.正式运行

测试经过以后，咱们在DynamoDB中新写入一条数据，在表格存储中立刻就能够读到这条数据，以下图。

10. 问题调查

Lambda 运行的日志都会写入到 CloudWatch 中，在 CloudWatch 选择对应的函数名，则能够实时查询到 Lambda 的运行状态。

代码解析

Lambda函数中，主要的代码逻辑为lambda_function.py 查看代码，其余则为表格存储SDK的依赖。lambda_function.py中主要包含了一下几个function：

• def batch_write_row(client, put_row_items) - 将组合好的数据 Item （包括增删改）批量写到表格存储中
• def get_primary_key(keys) – 根据变量PRIMARY_KEY 拿到源表和目的表的主键信息。
• def generate_update_attribute(new_image, old_image, key_list) – 解析Stream中的Modify操做，是对部分属性列的更新仍是删除了部分属性列。
• def generate_attribute(new_image, key_list) – 获取单个Record中插入的属性列信息。
• def get_tablestore_client() – 根据变量中的实例名、AK信息等初始化表格存储的客户端。
• def lambda_handler(event, context) – Lambda的入口函数。

若是有更复杂的同步逻辑，也能够基于 lambda_function.py 进行修改。

lambda_function.py 中打印的状态日志没有区分 INFO 或者 ERROR，为了保证数据同步的一致性，还须要对日志进行处理，并监控运行状态或者使用 lambda 的错误处理机制保证对异常状况的容错处理。