如何批量读取DynamoDB流并确保续读位置正确？

要确保批处理程序每次都从正确的未读事件位置开始，核心是结合DynamoDB流的分片迭代器和持久化检查点记录，以下是具体实现方案：

核心原理 #

DynamoDB流的每个分片都有唯一标识，且每个流事件都带有全局唯一的SequenceNumber。我们需要为每个分片记录最后处理完成的事件序列号，下次启动时基于该序列号获取分片迭代器，即可从首个未读事件开始读取。

步骤实现 #

1. 准备检查点存储 #

需要一个可靠的存储来保存每个分片的处理进度，推荐用一个极简的DynamoDB表（远比你之前的日志表轻便），表结构如下：

• 主键：shardId（字符串类型，分片唯一ID）
• 属性：lastProcessedSequenceNumber（字符串类型，最后处理的事件序列号）
• 可选属性：updatedAt（时间戳，记录检查点更新时间）

也可以选择S3存储检查点文件，但DynamoDB更适合高频读写的小数据场景。

2. 程序启动时获取分片迭代器 #

每次启动批处理程序时，按以下步骤处理每个流分片：

• 调用DescribeStream API获取当前流的所有分片列表。
• 对每个分片，从检查点存储中读取对应的lastProcessedSequenceNumber：
  • 如果存在记录：调用GetShardIterator API，指定ShardIteratorType为AFTER_SEQUENCE_NUMBER，传入该序列号，这样会从该序列号之后的第一个事件开始读取。
  • 如果不存在记录：根据业务需求选择TRIM_HORIZON（从流中最早的事件开始）或LATEST（从当前最新的事件开始）作为迭代器类型。

3. 处理流记录并更新检查点 #

• 通过GetRecords API读取分片的事件记录，每次调用可以获取最多1MB或1000条记录。
• 处理完一批记录后（比如成功生成/更新客户ZIP文件），立即更新检查点存储中对应分片的lastProcessedSequenceNumber为这批记录中最后一条的SequenceNumber。
• 当GetRecords返回的NextShardIterator为null时，说明该分片已关闭（不会再有新事件），可以在检查点中标记该分片已完成，后续不再处理。

4. 确保幂等性 #

由于DynamoDB流可能会重复发送事件（比如网络重试），处理实体时要保证幂等：

• 可以基于实体的版本号或流事件的SequenceNumber判断是否已处理过该变更。
• 生成ZIP时，只保留每个客户实体的最新版本，避免重复操作。

Java代码示例（基于AWS SDK v2） #

以下是关键步骤的代码片段：

import software.amazon.awssdk.services.dynamodbstreams.DynamoDbStreamsClient;
import software.amazon.awssdk.services.dynamodbstreams.model.*;

public class StreamProcessor {
    private final DynamoDbStreamsClient streamsClient;
    // 假设checkpointStore是自定义的检查点存储实现
    private final CheckpointStore checkpointStore;

    public void processStream(String streamArn) {
        // 获取流的分片列表
        DescribeStreamResponse streamDesc = streamsClient.describeStream(DescribeStreamRequest.builder()
                .streamArn(streamArn)
                .build());

        for (Shard shard : streamDesc.streamDescription().shards()) {
            String shardId = shard.shardId();
            String lastSeqNum = checkpointStore.getLastSequenceNumber(shardId);

            // 获取分片迭代器
            GetShardIteratorRequest iteratorRequest = GetShardIteratorRequest.builder()
                    .streamArn(streamArn)
                    .shardId(shardId)
                    .shardIteratorType(lastSeqNum != null ? ShardIteratorType.AFTER_SEQUENCE_NUMBER : ShardIteratorType.TRIM_HORIZON)
                    .sequenceNumber(lastSeqNum)
                    .build();
            GetShardIteratorResponse iteratorResp = streamsClient.getShardIterator(iteratorRequest);
            String shardIterator = iteratorResp.shardIterator();

            // 循环读取记录
            while (shardIterator != null) {
                GetRecordsResponse recordsResp = streamsClient.getRecords(GetRecordsRequest.builder()
                        .shardIterator(shardIterator)
                        .limit(1000)
                        .build());

                // 处理记录逻辑
                processRecords(recordsResp.records());

                // 更新检查点
                if (!recordsResp.records().isEmpty()) {
                    String lastProcessedSeq = recordsResp.records().get(recordsResp.records().size() - 1).dynamodb().sequenceNumber();
                    checkpointStore.updateCheckpoint(shardId, lastProcessedSeq);
                }

                shardIterator = recordsResp.nextShardIterator();
            }
        }
    }

    private void processRecords(java.util.List<Record> records) {
        // 实现你的业务逻辑：解析实体变更，生成/更新客户ZIP文件
    }
}

// 自定义检查点存储接口示例
interface CheckpointStore {
    String getLastSequenceNumber(String shardId);
    void updateCheckpoint(String shardId, String sequenceNumber);
}

注意事项 #

• DynamoDB流的事件保留期为24小时，需确保批处理程序的运行间隔不超过24小时，否则会丢失超出保留期的事件。
• 若流存在多个分片，可并行处理不同分片，提升处理效率，每个分片的检查点独立维护。
• 检查点更新要保证原子性，避免部分处理后程序崩溃导致的重复处理或遗漏。

内容的提问来源于stack exchange，提问作者ivant