从DynamoDB Streams转Kinesis后，如何在Lambda中控制记录读取量？

我明白你想要从DynamoDB Streams迁移到Kinesis Streams，并且希望完全自主控制从流中读取的记录数量，而不是依赖Lambda默认的批量触发逻辑。下面我会详细解释两种对接模式，重点讲你需要的第二种实现方法。

1. 默认的事件触发模式（和DynamoDB Streams类似） #

当你直接把Lambda配置为Kinesis Streams的触发器时，行为和DynamoDB Streams非常像：

• Kinesis会自动检测流中的新记录，当积累到一定数量（默认100条，可在触发器配置里调整Batch size）或等待时间达到阈值（默认300秒）时，就会触发Lambda。
• Lambda被触发后，事件参数里会携带这批固定数量的记录，你只能处理这批记录，无法自主选择读取更多或更少。
• 这种模式下，Lambda的并发由Kinesis的分片数决定（每个分片最多对应一个Lambda实例），并且Lambda会自动管理checkpoint（处理完记录后自动更新分片的读取位置）。

但这种模式显然不符合你的需求，所以接下来重点讲主动拉取的方案。

2. 自主控制读取量的主动拉取模式 #

这种模式下，Lambda不会被Kinesis自动触发，而是由你指定触发方式（比如定时触发、API调用触发等），然后在Lambda代码里主动连接Kinesis Streams，自主选择读取的分片和记录数量。具体实现步骤如下：

步骤1：给Lambda配置必要的IAM权限 #

首先要确保Lambda的执行角色有访问Kinesis的权限，需要添加以下权限：

• kinesis:GetRecords：读取流中的记录
• kinesis:GetShardIterator：获取分片的迭代器（用于定位读取位置）
• kinesis:DescribeStream：获取流的分片信息
• kinesis:ListShards：列出流的所有分片

你可以在IAM控制台给Lambda角色添加自定义策略，示例策略如下：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "kinesis:GetRecords",
                "kinesis:GetShardIterator",
                "kinesis:DescribeStream",
                "kinesis:ListShards"
            ],
            "Resource": "arn:aws:kinesis:你的区域:你的账号ID:stream/你的流名称"
        }
    ]
}

步骤2：选择Lambda的触发方式 #

你需要自己触发Lambda，常见的触发方式有：

• 定时触发：用CloudWatch Events（现在叫EventBridge）设置固定间隔触发，比如每分钟跑一次，每次读取指定数量的记录。
• API触发：通过API Gateway暴露接口，手动或其他服务调用这个接口来触发Lambda读取记录。
• 其他事件触发：比如S3上传事件、其他Lambda调用等，根据你的业务场景选择。

步骤3：在Lambda代码中实现主动拉取逻辑 #

核心逻辑是：获取流的分片 → 为每个分片获取读取迭代器（从上次处理的位置开始，或者从头/最新位置） → 调用GetRecords指定读取数量 → 处理记录 → 保存处理的位置（checkpoint）以便下次继续读取。

这里用Python代码举个例子，展示如何读取指定数量的记录：

import boto3
import os
from botocore.exceptions import ClientError

kinesis_client = boto3.client('kinesis')
STREAM_NAME = os.environ['KINESIS_STREAM_NAME']
CHECKPOINT_TABLE = os.environ['CHECKPOINT_TABLE']  # 用DynamoDB保存checkpoint
dynamodb_client = boto3.client('dynamodb')

def get_shard_iterator(shard_id, sequence_number=None):
    # 确定迭代器类型：如果有上次的sequence_number，用AFTER_SEQUENCE_NUMBER，否则用TRIM_HORIZON（从头读）或LATEST（读最新）
    iterator_type = 'TRIM_HORIZON'
    if sequence_number:
        iterator_type = 'AFTER_SEQUENCE_NUMBER'
    
    response = kinesis_client.get_shard_iterator(
        StreamName=STREAM_NAME,
        ShardId=shard_id,
        ShardIteratorType=iterator_type,
        StartingSequenceNumber=sequence_number
    )
    return response['ShardIterator']

def save_checkpoint(shard_id, sequence_number):
    # 把分片的最后处理位置保存到DynamoDB
    try:
        dynamodb_client.put_item(
            TableName=CHECKPOINT_TABLE,
            Item={
                'shard_id': {'S': shard_id},
                'last_sequence_number': {'S': sequence_number}
            }
        )
    except ClientError as e:
        print(f"保存checkpoint失败: {e}")

def lambda_handler(event, context):
    # 1. 获取流的所有分片
    shards = kinesis_client.list_shards(StreamName=STREAM_NAME)['Shards']
    
    for shard in shards:
        shard_id = shard['ShardId']
        # 2. 从DynamoDB获取上次的checkpoint
        last_sequence_number = None
        try:
            response = dynamodb_client.get_item(
                TableName=CHECKPOINT_TABLE,
                Key={'shard_id': {'S': shard_id}}
            )
            if 'Item' in response:
                last_sequence_number = response['Item']['last_sequence_number']['S']
        except ClientError as e:
            print(f"获取checkpoint失败: {e}")
        
        # 3. 获取分片迭代器
        shard_iterator = get_shard_iterator(shard_id, last_sequence_number)
        
        # 4. 读取指定数量的记录（比如50条）
        records_response = kinesis_client.get_records(
            ShardIterator=shard_iterator,
            Limit=50  # 这里指定你想读取的记录数量，最大10000条
        )
        
        records = records_response['Records']
        if not records:
            print(f"分片{shard_id}没有新记录")
            continue
        
        # 5. 处理记录（这里替换成你的业务逻辑）
        for record in records:
            data = record['Data'].decode('utf-8')
            print(f"处理记录: {data}")
        
        # 6. 更新checkpoint，保存最后一条记录的SequenceNumber
        last_processed_sequence = records[-1]['SequenceNumber']
        save_checkpoint(shard_id, last_processed_sequence)
        
        print(f"分片{shard_id}处理完成，共处理{len(records)}条记录")
    
    return {'statusCode': 200, 'body': '处理完成'}

步骤4：处理分片并发和checkpoint #

• 分片并发：如果你的Kinesis流有多个分片，上面的代码是串行处理每个分片，如果你想提高处理速度，可以在Lambda中用多线程或异步处理每个分片，或者把每个分片的处理逻辑拆分到单独的Lambda调用。
• Checkpoint管理：上面的例子用DynamoDB保存每个分片的最后处理位置，这样下次Lambda触发时就能从上次结束的位置继续读取，避免重复处理记录。如果不需要精确的checkpoint，也可以用TRIM_HORIZON（从头读）或LATEST（读最新）模式，但这样可能会重复读取或丢失记录，根据你的业务需求选择。

注意事项 #

• GetRecords的Limit参数最大只能是10000条，所以如果需要读取更多记录，需要循环调用GetRecords，直到获取到足够数量的记录或者没有更多记录。
• Lambda的执行时间有限制（最大15分钟），所以如果单次需要读取大量记录，要确保在超时前完成处理。
• 要处理Kinesis的迭代器过期问题：分片迭代器默认有效期是5分钟，如果Lambda执行时间过长，迭代器可能过期，这时候需要重新获取迭代器。

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