一桥飞架南北-中国区与 Global 区域 DynamoDB 表双向同步 (上)

Amazon DynamoDB 是一种完全托管的 NoSQL 数据库服务，提供快速且可预测的性能，同时还能够实现无缝扩展。使用 DynamoDB，客户可以免除操作和扩展分布式数据库的管理工作负担，因而无需担心硬件预置、设置和配置、复制、软件修补或集群扩展等问题。目前Amazon DynamoDB global table已经在中国区上线，为部署多区域、多主机数据库提供了完全托管的解决方案，但是global table只能用于北京和宁夏之间或者global区域之间的DynamoDB表同步，如果客户需要将中国区和global区域的DynamoDB表做双活复制同步，就需要客户自行构建解决方案，本文介绍了通过lambda、DynamoDB stream、Kinesis Stream等托管服务实现中国区与Global区域DynamoDB表之间数据双活复制同步，其中包含了架构、部署步骤以及监控方法，希望当您有类似需求的时候，能从中获得启发，助力业务发展。

0. 架构

当我们自行搭建数据库双活同步方案时，通常要考虑以下问题：

如何捕获源端变化并持续复制到目标端，本文中我们使用DynamoDB stream+lambda来解决这一问题。
如何避免循环复制，即在新加坡区域DynamoDB表的变更被复制到北京区域DynamoDB表后，再次被捕获并复制回新加坡区域DynamoDB表，如此反复，本文中我们在DynamoDB中每个item添加last_updater_region属性，记录变更发生所在的区域，而后捕获变更的lambda会辨识这个属性值，如果避免循环复制。
如何处理同一个item的变更冲突，本文中我们采取类似Global table的last writer win的机制，我们在DynamoDB中每个item添加last_update_timestamp属性，记录最后变更时间戳，如果目标表上现有item的last_update_timestamp值小于变更记录中的last_update_timestamp值，则在目标表上应用此变更，否则丢弃。
如何处理网络延迟，中国区域和Global区域之间有较大的网络延迟，如果单纯用DynamoDB stream+lambda从本区域DynamoDB表中获取变更再通过internet远程写入目标区域的DynamoDB表，写入延迟会很高，当源表写入负载较高时，lambda复制速度无法满足处理要求，甚至lambda会出现大量超时错误。对此我们采取了以下措施：
- 如果客户有Direct connect，可以考虑在目标区域搭建代理，使lambda向目标区域DynamoDB表的远程写入通过代理完成，利用Direct connect来降低网络延迟提高网络稳定性，进而提高lambda的复制效率。
- 考虑添加Kinesis stream作为中转站，从源区域捕获的变更先写入目标区域的Kinesis stream，因为每个kinesis的PutRecords操作可以每次写入最多500个记录，这样可以实现更大的吞吐量，降低网络延迟造成的影响，最后目标区域的Kinesis stream中的记录再被目标区域的kinesis消费将变更复制到目标DynamoDB表中。

在北京区域创建以下对象：

3张DynamoDB表，
- user-cn存储用户信息，数据将和新加坡区域的user-sg保持同步
- loader_stats 表用来统计压测时对user-cn变更的记录数
- replicator-stats表用来统计复制完成的记录数，以便监控复制进度
User-cn开启DynamoDB stream，用于记录user-cn表的所有变更
Kinesis 流ddb_replication_stream_cn 用于记录新加坡区域DynamoDB表user-sg的所有变更
Lambda 函数replicator_kinesis将北京区域中ddb_replication_stream_cn上的变更记录写入user-cn表
Lambda 函数ddb_send_to_kinesis将从北京区域的DynamoDB stream中读取变更记录然后写入到新加坡区域的ddb_replication_stream_sg
Parameter store中存储新加坡区域IAM用户的AK/SK
如果客户有direct connect，可以创建VPC并在VPC内部署一个代理服务器，lambda也部署在VPC内，令lambda利用新加坡区域的代理服务器进而利用direct connect，从而减少网络延迟并提高网络稳定性

在新加坡区域创建以下对象：

3张DynamoDB表
- user-sg存储用户信息，数据将于北京区域的user-cn保持同步
- loader_stats 表用来统计压测时对user-sg变更的记录数
- replicator-stats表用来存储复制的变更记录数，以便监控复制进度
User-sg开启DynamoDB stream，用于记录user-cn表的所有变更
Kinesis stream ddb_replication_stream_sg 用于记录北京区域DynamoDB表user-cn的所有变更
Lambda 函数replicator_kinesis将新加坡区域中ddb_replication_stream_sg上的变更记录写入user-sg表
Lambda 函数ddb_send_to_kinesis将从新加坡区域的DynamoDB stream中读取变更记录然后写入到北京区域的ddb_replication_stream_cn
Parameter store中存储北京区域IAM用户的AK/SK
如果客户有跨境专线连接AWS中国区域和海外区域，可以创建VPC并在VPC内部署一个代理服务器，lambda也部署在VPC内，令lambda利用北京区域的代理服务器进而利用direct connect，从而减少网络延迟并提高网络稳定性

因为对象较多，我们以北京区域DynamoDB表user-cn中的变更数据向新加坡区域同步过程为例，将整个数据流梳理一遍，从新加坡DynamoDB表user-sg同步数据到user-cn的过程类似将不再赘述

当数据写入北京区域DynamoDB表user-cn后，变更记录首先会写入北京区域的DynamoDB stream
北京区域的Lambda ddb_send_to_kinesis将从北京区域的DynamoDB stream中读取变更记录，首先判断记录的last_updater_region是否是新加坡区域，如果是就丢弃，不是则写入到新加坡区域的ddb_replication_stream_sg，这一步将跨region，主要的网络延迟就在这一步
新加坡区域的Lambda replicator_kinesis将新加坡区域中ddb_replication_stream_sg上的变更记录写入user-sg表，在写入时会通过condition判断变更记录的时间戳是否大于当前记录的变更时间戳，只有大于才会写入。
新加坡区域的user-sg的变化记录又出现在DynamoDB stream中，并被新加坡区域的Lambda ddb_send_to_kinesis读取，而后判断记录的last_updater_region是否是北京区域，因为该变化恰恰是来自北京，所以该记录被丢弃，从而避免了循环复制。

1. 部署环境

1.1 准备压测机

网络设置

在北京与新加坡区域中各创建一个VPC，记录两个VPC的NAT gateway所对应的EIP。
在两个VPC内分别创建一个EC2用作压测机，本DEMO中选择的是large类型实例，为每一个EC2绑定一个EIP。
因为篇幅有限，具体步骤请见《Amazon Virtual Private Cloud用户指南》，此处不再赘述。

安装软件

1、因为后续操作中会大量使用AWS CLI，故而需要在2个压测服务器上确认安装python3等软件并升级AWS CLI到最新版本，以ec2-user用户运行如下命令：

yum list installed | grep -i python3
sudo yum install python3 -y
python3 -m venv my_app/env
source ~/my_app/env/bin/activate
pip install pip --upgrade
pip install boto3
echo "source ${HOME}/my_app/env/bin/activate" >> ${HOME}/.bashrc
source ~/.bashrc
pip install faker uuid
pip install --upgrade awscli

2、后续模拟加载数据到DynamoDB表的脚本会用到parallel，故而也需要安装parallel

http://git.savannah.gnu.org/cgit/parallel.git/tree/README
wget https://ftpmirror.gnu.org/parallel/parallel-20200322.tar.bz2
wget https://ftpmirror.gnu.org/parallel/parallel-20200322.tar.bz2.sig
gpg parallel-20200322.tar.bz2.sig
bzip2 -dc parallel-20200322.tar.bz2 | tar xvf -
cd parallel-20200322
./configure && make && sudo make install

配置profile

接下来在2个压测服务器上通过aws configure –profile 分别配置相应中国区域和Global区域IAM用户的AK/SK，以便后续能顺利运行AWS CLI命令。请生成分别名为cn和sg的profile，以便后面加载记录时使用。

1.2 建表

新加坡区域

在新加坡区域建立DynamoDB表user-sg，并开启DynamoDB stream（NEW_AND_OLD_IMAGES类型）,设置容量模式为On-Demand。再创建loader_stats和replicator_stats表用来记录更新记录以便比对数据同步进度。

在新加坡区域的压测服务器上通过json创建DynamoDB表，先编写json如下，读者在工作中可以按照实际情况修改相关配置。

vim user-sg.json
{
        "AttributeDefinitions": [
            {
                "AttributeName": "PK",
                "AttributeType": "S"
            }
        ],
       "BillingMode": "PAY_PER_REQUEST",
        "TableName": "user-sg",
        "StreamSpecification": {
            "StreamViewType": "NEW_AND_OLD_IMAGES",
            "StreamEnabled": true
        },
        "KeySchema": [
            {
                "KeyType": "HASH",
                "AttributeName": "PK"
            }
        ]
}

然后用命令创建DynamoDB表user-sg

aws dynamodb create-table --cli-input-json file://user-sg.json --profile sg

最后可以验证

aws dynamodb describe-table --table-name user-sg --profile sg

然后用类似方法创建DynamoDB表replicator_stats和loader_stats并且初始化表中的统计值

vim replicator_stats.json

{
        "AttributeDefinitions": [
            {
                "AttributeName": "PK",
                "AttributeType": "S"
            }
        ],
       "BillingMode": "PAY_PER_REQUEST",
        "TableName": "replicator_stats",
        "KeySchema": [
            {
                "KeyType": "HASH",
                "AttributeName": "PK"
            }
        ]
}
aws dynamodb create-table --cli-input-json file://replicator_stats.json
aws dynamodb put-item --table-name replicator_stats --item '{ "PK": {"S":"replicated_count"}, "cnt": {"N":"0"}}'
vim loader_stats.json
{
        "AttributeDefinitions": [
            {
                "AttributeName": "PK",
                "AttributeType": "S"
            }
        ],
       "BillingMode": "PAY_PER_REQUEST",
        "TableName": "loader_stats",
        "KeySchema": [
            {
                "KeyType": "HASH",
                "AttributeName": "PK"
            }
        ]
}
aws dynamodb create-table --cli-input-json file://loader_stats.json
aws dynamodb put-item --table-name loader_stats --item '{ "PK": {"S":"loaded_count"}, "cnt": {"N":"0"}}'

北京区域

在北京区域的压测服务器上通过json创建DynamoDB表，创建方式和前述新加坡区域类似。先编写json如下，读者在工作中可以按照实际情况修改相关配置。

vim user-cn.json
{
        "AttributeDefinitions": [
            {
                "AttributeName": "PK",
                "AttributeType": "S"
            }
        ],
       "BillingMode": "PAY_PER_REQUEST",
        "TableName": "user-cn",
        "StreamSpecification": {
            "StreamViewType": "NEW_AND_OLD_IMAGES",
            "StreamEnabled": true
        },
        "KeySchema": [
            {
                "KeyType": "HASH",
                "AttributeName": "PK"
            }
        ]
}

然后用命令创建DynamoDB表user-cn

aws dynamodb create-table --cli-input-json file://user-cn.json --profile cn

最后可以验证

aws dynamodb describe-table --table-name user-cn --profile cn

用类似方法创建DynamoDB表replicator_stats和loader_stats （replicator_stats.json及loader_stats.json文件内容和新加坡区域一样）

aws dynamodb create-table --cli-input-json file://replicator_stats.json --profile cn
aws dynamodb put-item --table-name replicator_stats --item '{ "PK": {"S":"replicated_count"}, "cnt": {"N":"0"}}'
aws dynamodb create-table --cli-input-json file://loader_stats.json --profile cn
aws dynamodb put-item --table-name loader_stats --item '{ "PK": {"S":"loaded_count"}, "cnt": {"N":"0"}}'

1.3 准备parameter store里的参数

新加坡区域

创建/DDBReplication/TableCN/AccessKey （String）和/DDBReplication/TableCN/SecretKey（SecureString），分别存入中国区用户的AK/SK

aws ssm put-parameter --name /DDBReplication/TableCN/AccessKey --value <access_key> --type String --profile sg
aws ssm put-parameter --name /DDBReplication/TableCN/SecretKey --value <secret_key> --type SecureString --profile sg

北京区域

创建/DDBReplication/TableSG/AccessKey （String）和/DDBReplication/TableSG/SecretKey（SecureString），分别存入Global用户的AK/SK

aws ssm put-parameter --name /DDBReplication/TableSG/AccessKey --value <access_key> --type String --profile cn
aws ssm put-parameter --name /DDBReplication/TableSG/SecretKey --value <secret_key> --type SecureString --profile cn

1.4 创建SQS

新加坡区域

创建2个标准SQS队列用于Lambda的On-failure Destination。

aws sqs create-queue --queue-name ddbstreamsg --profile sg
aws sqs create-queue --queue-name ddbreplicatorsg --profile sg

北京区域

创建2个标准SQS队列用于Lambda的On-failure Destination。

aws sqs create-queue --queue-name ddbstreamcn --profile cn
aws sqs create-queue --queue-name ddbreplicatorcn --profile cn

1.5 创建Kinesis Data Stream

新加坡区域

创建Kinesis Data Stream，因为DEMO中写入量有限，选取一个shard，实际生产中请酌情调整

aws kinesis create-stream --stream-name ddb_replication_stream_sg --shard-count 1 --profile sg

北京区域

创建Kinesis Data Stream，因为DEMO中写入量有限，选取一个shard，实际生产中请酌情调整

aws kinesis create-stream --stream-name ddb_replication_stream_cn --shard-count 1 --profile cn

1.6 创建Lambda role并且授权

请从iam_role_policy处下载相应的json文件，然后酌情修改相应的权限和用户信息，以备后续步骤创建role。

新加坡区域

需要创建两个role为后续lambda使用

1、将DynamoDB stream中event发送到中国区的Kinesis的lambda需要以下权限：

访问DynamoDB stream的权限
访问Parameter Store的权限
访问lambda On-failure destination的SQS队列的权限
访问VPC的权限：这部分权限我们使用现成的policyAWSLambdaBasicExecutionRole 、AWSLambdaVPCAccessExecutionRole

aws iam create-policy --policy-name ddb_send_to_kinesis_policy --policy-document file://iam_policy_example/ddb_send_to_kinesis_policy_sg.json --profile sg
aws iam create-role --role-name ddb_send_to_kinesis_role --assume-role-policy-document file://iam_policy_example/lambda-role-trust-policy.json --profile sg
aws iam attach-role-policy --role-name ddb_send_to_kinesis_role --policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaVPCAccessExecutionRole --profile sg
aws iam attach-role-policy --role-name ddb_send_to_kinesis_role --policy-arn arn:aws:iam::<Global account>:policy/ddb_send_to_kinesis_policy --profile sg

2、将新加坡Kinesis Data Stream中event写入新加坡区DDB需要以下权限：

访问Kinesis Data Stream的权限
访问CloudWatch的权限
访问user-cn与replicator_stats两个DynamoDB表的权限
访问lambda On-failure destination的SQS队列的权限
访问VPC的权限：这部分权限我们使用现成的policy AWSLambdaBasicExecutionRole 、AWSLambdaVPCAccessExecutionRole

aws iam create-policy --policy-name replicator_kinesis_policy --policy-document file://iam_policy_example/replicator_kinesis_policy_sg.json --profile sg
aws iam create-role --role-name replicator_kinesis_role --assume-role-policy-document file://iam_policy_example/lambda-role-trust-policy.json --profile sg
aws iam attach-role-policy --role-name replicator_kinesis_role --policy-arn arn:aws-cn:iam::aws:policy/service-role/AWSLambdaVPCAccessExecutionRole
aws iam attach-role-policy --role-name replicator_kinesis_role --policy-arn arn:aws:iam::<Global account>:policy/replicator_kinesis_policy
aws iam attach-role-policy --role-name replicator_kinesis_role --policy-arn arn:aws-cn:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole

北京区域

需要创建两个role为后续lambda使用

1、将DynamoDB stream中event发送到新加坡区的Kinesis需要以下权限：

访问DynamoDB stream的权限
访问Parameter Store的权限
访问lambda On-failure destination的SQS队列的权限
访问VPC的权限：这部分权限我们使用现成的policy AWSLambdaBasicExecutionRole 、AWSLambdaVPCAccessExecutionRole

aws iam create-policy --policy-name ddb_send_to_kinesis_policy --policy-document file://iam_policy_example/ddb_send_to_kinesis_policy_cn.json --profile cn
aws iam create-role --role-name ddb_send_to_kinesis_role --assume-role-policy-document file://iam_policy_example/lambda-role-trust-policy.json --profile cn
aws iam attach-role-policy --role-name ddb_send_to_kinesis_role --policy-arn arn:aws-cn:iam::aws:policy/service-role/AWSLambdaVPCAccessExecutionRole --profile cn
aws iam attach-role-policy --role-name ddb_send_to_kinesis_role --policy-arn arn:aws-cn:iam::<China account>:policy/ddb_send_to_kinesis_policy --profile cn

2、将Kinesis Data Stream中event写入DDB需要以下权限：

访问Kinesis Data Stream的权限
访问CloudWatch的权限
访问user-cn与replicator_stats2个DynamoDB表的权限
访问lambda On-failure destination的SQS队列的权限
访问VPC的权限：这部分权限我们使用现成的policy AWSLambdaBasicExecutionRole 、AWSLambdaVPCAccessExecutionRole

aws iam create-policy --policy-name replicator_kinesis_policy --policy-document file://iam_policy_example/replicator_kinesis_policy_cn.json --profile cn
aws iam create-role --role-name replicator_kinesis_role --assume-role-policy-document file://iam_policy_example/lambda-role-trust-policy.json --profile cn
aws iam attach-role-policy --role-name replicator_kinesis_role --policy-arn arn:aws-cn:iam::aws:policy/service-role/AWSLambdaVPCAccessExecutionRole --profile cn
aws iam attach-role-policy --role-name replicator_kinesis_role --policy-arn arn:aws-cn:iam::<China account>:policy/replicator_kinesis_policy --profile cn
aws iam attach-role-policy --role-name replicator_kinesis_role --policy-arn arn:aws-cn:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole --profile cn

（可选）安装代理

在有些实验中，我们发现在跨境公网传输数据时网络时延不稳定。如果客户已经有了Direct connect，则可以通过增加代理服务器的方式利用Direct connect，从而实现稳定的较低时延。在前述的VPC内再建一台EC2，选择实例类型c5.xlarge，并且在EC2上安装代理。这里以Squid为例

以ec2-user运行

sudo yum install squid
sudo vi /etc/sysctl.conf
net.ipv4.ip_forward = 1
sudo sysctl -p /etc/sysctl.conf
sudo sysctl -a |grep -w ip_forward

grep -n 'http_access deny all' /etc/squid/squid.conf
56:http_access deny all

这里修改http_access为allow all，那么代理服务器的安全组就需要设置为只对lambda所在VPC的NAT网段放行，否则不安全

grep -n http /etc/squid/squid.conf |grep -w all
56:http_access allow all

sudo vi /etc/squid/squid.conf
http_port 80
sudo systemctl start squid
sudo systemctl status squid
sudo systemctl enable squid.service

代理服务器的安全组设定

两个区域内代理服务器的安全组中设定只允许对端region lambda所在VPC的NAT Gateway的http/https流量进来。

到此为止，所有环境都已准备完毕，万事俱备只欠东风，接下来我们只需要创建同步数据的lambda并进行测试。因为篇幅有限，请读者阅读

附录

《Amazon DynamoDB开发人员指南》

《AWS Lambda开发人员指南》

亚马逊AWS官方博客

一桥飞架南北-中国区与 Global 区域 DynamoDB 表双向同步 (上)

0. 架构

1. 部署环境

1.1 准备压测机

网络设置

安装软件

配置profile

1.2 建表

新加坡区域

北京区域

1.3 准备parameter store里的参数

新加坡区域

北京区域

1.4 创建SQS

新加坡区域

北京区域

1.5 创建Kinesis Data Stream

新加坡区域

北京区域

1.6 创建Lambda role并且授权

新加坡区域

北京区域

（可选）安装代理

附录

本篇作者