如何判定处理Kinesis事件的最后一个消费Lambda并触发第三方Lambda？

核心思路：批次标识+分布式计数追踪 #

Kinesis触发Lambda时，同一原始批次的50条记录可能被拆分到多个Lambda实例处理（取决于shard数量、Kinesis批量配置）。要判断是否是最后一个完成处理的实例，需要通过唯一批次标识和已处理记录数的原子统计来确认整个批次的处理状态。

具体实现步骤 #

1. 生产Lambda嵌入批次元数据
   在发送到Kinesis的每条记录中，添加自定义元数据，用于后续追踪批次：

• batch_id：每个批次的唯一UUID，确保50条记录归属同一个批次
• total_records：固定为50，标记当前批次的总记录数
• 可选record_index：记录在批次中的序号，用于校验完整性

示例Python代码片段：

import uuid
import json
import boto3

kinesis_client = boto3.client('kinesis')

def lambda_handler(event, context):
    documents = [...]  # 生成的50条文档列表
    batch_id = str(uuid.uuid4())
    total_records = 50

    kinesis_records = []
    for idx, doc in enumerate(documents):
        payload = json.dumps({
            "data": doc,
            "metadata": {
                "batch_id": batch_id,
                "total_records": total_records,
                "record_index": idx + 1
            }
        })
        kinesis_records.append({
            "Data": payload,
            "PartitionKey": f"batch-{batch_id}"  # 可选：强制同批次进入同一shard，减少拆分
        })

    kinesis_client.put_records(StreamName="your-stream-name", Records=kinesis_records)
    return {"statusCode": 200}

2. 消费Lambda处理并更新分布式计数
   使用DynamoDB作为分布式计数器，原子性更新批次的已处理记录数，当计数等于总记录数时，触发第三个Lambda：

• 先创建DynamoDB表，主键为batch_id，包含字段processed_count、total_records、is_completed
• 消费Lambda处理完当前批次的记录后，执行原子累加操作，判断是否完成整个批次

示例Python代码片段：

import boto3
import json

dynamodb = boto3.resource('dynamodb')
batch_table = dynamodb.Table('batch-processing-tracker')
lambda_client = boto3.client('lambda')

def process_documents(records):
    # 你的文档处理逻辑
    pass

def lambda_handler(event, context):
    processed_records = []
    batch_id = None
    total_records = 0

    # 解析Kinesis事件中的记录
    for record in event['Records']:
        payload = json.loads(record['kinesis']['data'])
        processed_records.append(payload['data'])
        batch_id = payload['metadata']['batch_id']
        total_records = payload['metadata']['total_records']

    # 执行文档处理
    process_documents(processed_records)

    # 原子更新已处理计数
    update_response = batch_table.update_item(
        Key={'batch_id': batch_id},
        UpdateExpression='SET processed_count = if_not_exists(processed_count, :start) + :incr, total_records = :total',
        ExpressionAttributeValues={
            ':incr': len(processed_records),
            ':start': 0,
            ':total': total_records
        },
        ReturnValues='ALL_NEW'
    )

    # 判断是否是最后一个处理实例
    current_count = update_response['Attributes']['processed_count']
    if current_count == total_records:
        # 触发第三个Lambda
        lambda_client.invoke(
            FunctionName='your-third-lambda-name',
            InvocationType='Event',
            Payload=json.dumps({'batch_id': batch_id})
        )
        # 标记批次处理完成（可选）
        batch_table.update_item(
            Key={'batch_id': batch_id},
            UpdateExpression='SET is_completed = :val',
            ExpressionAttributeValues={':val': True}
        )

    return {"statusCode": 200}

3. 可选：强制同批次单shard处理
   如果业务允许，可以通过设置相同的PartitionKey将整个50条批次的记录分配到同一个shard，此时Lambda会一次性处理整个批次，无需分布式计数，处理完成后直接触发第三个Lambda即可。但这种方式可能导致shard热点，需根据吞吐量评估使用。

关键注意事项 #

• 幂等性保障：Lambda可能重试，DynamoDB的原子更新本身具备幂等性；第三个Lambda需校验批次是否已处理，避免重复执行。
• 数据清理：给DynamoDB表设置TTL字段，自动清理已完成的批次记录，避免表数据膨胀。
• 错误处理：添加failed_count字段追踪处理失败的记录，当失败次数超过阈值时触发告警，避免批次永远无法完成。

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