Amazon DynamoDB 表的跨区域复制实践——以北京至宁夏为例

发布于: Jun 17, 2022

Amazon Web Services 在中国境内有两个区，分别是北京区域和宁夏区域。假定复制路径是从北京到宁夏，那么根据复制应用所处区的不同，至少有三种配置方式，罗列如下：

左图是复制应用与源表在同一区域的情况
中图是复制应用与目标表在同一区域的情况
右图是复制应用在云下的情况

建议是采用复制应用在云端的配置，以降低网络延迟。故在后面的测试环节，不对第三种架构进行测试。

实现详述

表终端节点与流终端节点是相互独立的。要实现表复制，除需要开启表的流功能外，还需要进行必要的设置。有四种写入流的信息类型可以设置：

键值：仅传输更改表项的键值信息
新印象：传输更改表项的键值和更改后的信息
旧印象：传输更改表项的键值和更改前的信息
新旧印象：传输更改表项的键值和更改前后的信息

如果流功能关闭后再次开启，其终端节点会变动。

流结构

流由流记录构成。一条流记录包含该流所属表的一个数据项的更改。按组来组织流记录，又称为分区。分区类似于容器，包含多条流记录以及读取和遍历这些记录的必要信息。分区中的流记录会在二十四小时后删除。

流结构不是简单的线性结构。分区会按需被自动创建、拆分和删除。一个分区有可能有一个或者数个子分区，形成父子结构。访问时，必须按先父后子的顺序来处理流，以保证其正确性。在此，推荐使用 DynamoDB 流 Kinesis 适配器（以下简称流适配器）来简化流处理的复杂性。该适配器还可以正确处理新的或者过期的分区。本文不讨论如何使用更初级的流函数接口来处理流。

流函数接口的设计与 Kinesis 的函数接口很相似，例如下述四个关键函数： ListStreams, DescribeStream, GetShards 和 GetShardIterator。因此，在使用了流适配器后， Kinesis 客户端（以下简称流客户端）就可以匹配及处理流信息。该客户端就顺理成章的可以用来进一步简化流处理操作，如下图所示。

示例代码

组件	版本
Java	8
Dagger	2.x
Amazon Kinesis Client Library For Java	1.11.x
DynamoDB Streams Adapter For Java	1.5.x
DynamoDBLocal	1.11.x

本节基于 Java 语言和 Dagger 依赖注入机制，罗列相关源代码片段，构建一个跨区域 DynamoDB 表复制的应用程序。请注意，现阶段只支持单表的复制，未来可以拓展至多表的复制，其实现核心与机制是类似的。此外，囿于篇幅，以下仅罗列关键代码片段，省略了譬如异常处理、日志记录等代码片段。具体来说，主要依赖关系如下：

首先注册 Amazon Web Services 账号，拿到相应用户的访问密钥。然后构建密钥供应器 Amazon Credentials Provider。

表结构

关于数据模型，由于 DynamoDB 表是无结构的存储，为了简化测试，假定该表只有一个字符串类型的主键 id 。用户可以根据实际表结构及主键信息，对代码做相应修改。事实上，这个表可以存储任意数据。

@DynamoDBTable(tableName = "BCSReplicateTestTable")public class Item {
    private String id;
    
    @DynamoDBHashKey
    @DynamoDBAutoGeneratedKey
    public String getId() { return id; }
    public void setId(String id) { this.id = id; }}

目的表部分

以下创建目的表流复制工作器。

@Destination@Provides@Singletonstatic Worker getWorker(
        @Source IRecordProcessorFactory processorFactory,
        @Source KinesisClientLibConfiguration configuration,
        @Source AmazonDynamoDBStreamsAdapterClient adapterClient,
        @Destination AmazonDynamoDB dynamoDB,
        @Destination AmazonCloudWatch cloudWatch) {
    return new Worker.Builder()
            .recordProcessorFactory(processorFactory)
            .config(configuration)
            .kinesisClient(adapterClient)
            .dynamoDBClient(dynamoDB)
            .cloudWatchClient(cloudWatch)
            .build();}

开始复制

以上准备好各个工作组件，现在将其装配起来，开始表复制工作。

@Singletonpublic class Replicator implements AutoCloseable {
    private final ExecutorService executor;
    private final Worker replicator;
    @Inject
    Replicator(@Destination Worker replicator) {
        this.replicator = replicator;
        executor = Executors.newSingleThreadExecutor();
    }
    public void replicate() {
        executor.execute(replicator);
    }
    @Override
    public void close() {
        replicator.shutdown();
        executor.shutdown();
    }}

对于表的跨区域复制，测试侧重复制延迟及复制器性能。本节测试在预置模式下，采用不同写容单位（WCU, Write Capacity Unit）时延迟的变化情况，同时监控复制器的性能指标。一个写容单位表示对大小最多为 1 KB 的项目每秒执行一次写入。而事务写入请求需要 2 个写容单位才能对大小最多为 1 KB 的项目每秒执行一次写入。考虑到各种不稳定因素可能会对测试产生影响，如网络异常等，测试程序并不会就单个表项目的复制计算延迟，而是在固定写容单位下随机测试部分表项目的延迟，以尽可能接近真实情况。

以下编写两个测试程序，分别运行在源端和目的端。