如何在Lambda中通过DynamoDB流批次信息获取原表条目？

1. 提取关键流信息 #

从存储的流数据中，需要提取data.DDBStreamBatchInfo下的核心字段：

• streamArn：定位目标DynamoDB流的ARN
• shardId：指定要读取的流分片ID
• startSequenceNumber：批次起始序列号，对应目标记录
• 确认batchSize为1，说明该批次仅包含一条流记录

2. 使用AWS SDK读取DynamoDB流记录 #

以Python为例，通过boto3调用DynamoDB Streams API获取对应记录：

import boto3
from boto3.dynamodb.types import TypeDeserializer

def get_original_dynamodb_item(stream_arn, shard_id, sequence_number):
    # 初始化DynamoDB Streams客户端
    streams_client = boto3.client('dynamodbstreams', region_name='ap-southeast-2')
    
    # 获取分片迭代器，从指定序列号位置开始读取
    iterator_response = streams_client.get_shard_iterator(
        StreamArn=stream_arn,
        ShardId=shard_id,
        ShardIteratorType='AT_SEQUENCE_NUMBER',
        SequenceNumber=sequence_number
    )
    shard_iterator = iterator_response['ShardIterator']
    
    # 获取流记录
    records_response = streams_client.get_records(ShardIterator=shard_iterator)
    records = records_response['Records']
    
    # 定位目标记录（batchSize=1时直接取第一条）
    target_record = next((r for r in records if r['dynamodb']['SequenceNumber'] == sequence_number), None)
    
    if not target_record:
        return None
    
    # 将DynamoDB格式数据转换为普通JSON
    deserializer = TypeDeserializer()
    original_item = {
        k: deserializer.deserialize(v) for k, v in target_record['dynamodb']['NewImage'].items()
    }
    
    return original_item

# 传入从存储数据中提取的参数
stream_arn = "arn:aws:dynamodb:ap-southeast-2:xxxxxxx:table/prod-iot-vision-data-ingestion/stream/2022-10-07T03:38:45.908"
shard_id = "shardId-00000001689839413006-7931cd0d"
sequence_number = "7755608300000000009920716151"

original_item = get_original_dynamodb_item(stream_arn, shard_id, sequence_number)
print(original_item)

3. 关键细节说明 #

• 数据格式转换：DynamoDB流返回的NewImage/OldImage是带类型标记的结构（如{"S": "字符串值"}），需用TypeDeserializer转换为常规JSON格式。
• 操作类型适配：根据record['eventName']判断操作类型（如INSERT/MODIFY/REMOVE），选择读取NewImage（新增/修改后数据）或OldImage（修改前/删除前数据）。
• 权限配置：Lambda需附加包含dynamodbstreams:GetShardIterator和dynamodbstreams:GetRecords权限的IAM策略。

4. 注意事项 #

• 流数据保留期由DynamoDB表的流配置决定（24小时至365天），超过保留期将无法获取对应记录。
• 若批次batchSize大于1，需循环调用get_records，遍历从startSequenceNumber到endSequenceNumber的所有记录。

内容的提问来源于stack exchange，提问作者Kasun Sampath Weligamage