如何使用KCL Consumer重新处理AWS Kinesis数据流中的消息？

现有场景概述 #

我们通过与Kinesis数据流预配置分片数匹配的Auto Scaling Group（ASG）在EC2实例上运行KCL Consumer：

• 若数据流有n个预配置分片，最多部署n台EC2实例，每台单独消费一个分片
• 当前KCL Consumer的分片迭代器设置为LATEST，消息到达数据流后立即实时处理
• 用DynamoDB表存储各分片的检查点，用于跟踪KCL Consumer Worker的分片租赁与处理进度

需求 #

重新处理Kinesis数据流保留期内（默认7天）的所有消息，寻求简便可行的实现方案。

对现有理论方案的分析 #

方案一 #

• 步骤：停止KCL Consumer Worker → 删除关联的DynamoDB表 → 重启KCL Consumer服务
• 问题：
  • 删除DynamoDB表会丢失所有历史跟踪数据，后续恢复实时消费需重新初始化检查点逻辑，风险较高
  • 若数据流存在分片合并/拆分历史，删除表后KCL可能无法正确识别分片继承关系，导致部分历史数据漏处理

方案二 #

• 步骤：停止KCL Consumer → 更新各分片的检查点值至更早时间戳 → 重启KCL Consumer服务
• 问题：
  • KCL存储的检查点是分片的序列号（Sequence Number），而非时间戳，直接修改时间戳无法被KCL识别
  • 序列号与时间戳无直接转换公式，手动修改易出错，分片拆分/合并后的序列号逻辑更复杂，操作难度大

推荐的可靠方案 #

方案三：使用独立临时消费组重处理（官方推荐无侵入方案） #

无需修改原有生产环境消费逻辑，操作风险极低：

1. 部署一个临时KCL Consumer消费组（使用全新应用名称，对应全新的DynamoDB检查点表）
2. 将临时消费组的分片迭代器设置为TRIM_HORIZON（从数据流保留期的最早位置开始消费）
3. 按需配置临时ASG实例数（最多等于分片数n），实现并行快速消费全量历史数据
4. 重处理完成后，直接销毁临时消费组及对应的DynamoDB表即可

方案四：修改原有消费组检查点（适合无额外资源的场景） #

若必须使用原有消费组，可按以下步骤操作：

1. 完全停止所有KCL Consumer Worker
2. 进入DynamoDB控制台，找到对应的检查点表
3. 对每个分片条目，将checkpoint字段值修改为TRIM_HORIZON（KCL支持直接识别该特殊值，会从分片最早可用位置开始消费）
4. 重启KCL Consumer Worker，此时所有Worker会从数据流保留期起始位置开始消费
5. 重处理完成后，若需恢复实时消费，可在处理完所有历史数据后将分片迭代器改回LATEST（或等待KCL自动将检查点更新至最新位置）

方案对比 #

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