如何处理DynamoDB BatchWriteItem最多25条的批量写入限制？

嘿，针对你这个DynamoDB批量写入的限制问题，结合你的三个Lambda函数流程，我整理了几个经过实践验证的最佳方案，你可以根据自己的需求来选：

方案1：同步分批次处理 + 自动重试未写入项 #

这是最直接、无需额外服务的方案，核心就是在SaveAllToDynamoDBLambda里把250条数据拆成每25条一组，逐个调用BatchWriteItem，同时处理DynamoDB返回的UnprocessedItems（因为流量限制或其他原因，部分项可能写入失败）。

具体实现思路： #

• 把从GetNItemsFromExternalSourceLambda传来的250条数据，按25条为一组拆分（刚好10组）
• 对每个批次调用BatchWriteItem API
• 循环处理返回的UnprocessedItems，直到所有数据都写入成功（建议加指数退避重试，避免频繁请求触发限流）

简化代码示例（Python）： #

import boto3
import time

dynamodb = boto3.client('dynamodb')
TABLE_NAME = '你的目标表名'

def split_into_batches(items, batch_size=25):
    # 拆分列表为指定大小的批次
    return [items[i:i+batch_size] for i in range(0, len(items), batch_size)]

def lambda_handler(event, context):
    # 接收上游传来的250条数据
    raw_items = event['items']
    # 转换为DynamoDB接受的Item格式（根据你的数据结构调整）
    dynamo_items = [convert_to_dynamo_item(item) for item in raw_items]
    
    batches = split_into_batches(dynamo_items)
    
    for batch in batches:
        request_items = {
            TABLE_NAME: [{'PutRequest': {'Item': item}} for item in batch]
        }
        
        response = dynamodb.batch_write_item(RequestItems=request_items)
        
        # 处理未成功写入的项，指数退避重试
        unprocessed = response.get('UnprocessedItems', {})
        retry_count = 0
        while unprocessed and retry_count < 5:
            time.sleep(2 ** retry_count)  # 指数退避：1s, 2s, 4s...
            response = dynamodb.batch_write_item(RequestItems=unprocessed)
            unprocessed = response.get('UnprocessedItems', {})
            retry_count += 1
        
        # 如果重试5次还失败，可以把数据放到死信队列后续处理
        if unprocessed:
            print(f"Failed to write items after retries: {unprocessed}")
            # 可选：发送到SQS死信队列
    
    # 写入完成后，触发AnalyzeDynamoDBLambda或者返回结果给上游
    return {
        "status": "success",
        "total_items": len(raw_items),
        "batches_processed": len(batches)
    }

def convert_to_dynamo_item(raw_item):
    # 根据你的数据结构，把原始数据转为DynamoDB的类型格式，比如：
    return {
        "id": {"S": raw_item["id"]},
        "value": {"N": str(raw_item["value"])}
        # 其他字段...
    }

方案2：SQS解耦 + 异步批量写入 #

如果不想让SaveAllToDynamoDBLambda同步等待所有写入完成（比如250条数据写入耗时较长，会占用Lambda执行时间），可以用SQS做中间件来解耦流程：

具体实现思路： #

1. GetNItemsFromExternalSourceLambda拿到250条数据后，调用SaveAllToDynamoDBLambda
2. SaveAllToDynamoDBLambda把数据拆成25条一组，每组作为一条消息发送到SQS队列
3. 配置SQS触发一个专门的BatchWriteToDynamoLambda（也可以复用原有的SaveAllToDynamoDBLambda，只要调整逻辑），每个消息对应一个25条的批次
4. BatchWriteToDynamoLambda处理单个批次的写入，并处理UnprocessedItems
5. 所有SQS消息处理完成后，再触发AnalyzeDynamoDBLambda（可以用CloudWatch Events或者Step Functions监听SQS空队列事件）

优势： #

• 解耦数据获取和写入流程，SaveAllToDynamoDBLambda可以快速返回，不用等待写入完成
• SQS自带重试机制，某个批次写入失败会自动重试（可配置重试次数和死信队列）
• 可以水平扩展BatchWriteToDynamoLambda的并发数，提升写入速度

方案3：Step Functions编排全流程（适合复杂分页+多步骤场景） #

如果你的流程涉及分页（比如需要多次调用外部API获取数据），且后续还要触发AnalyzeDynamoDBLambda，用AWS Step Functions来编排整个流程会更健壮，也更容易监控和调试：

具体状态机流程： #

1. 调用GetNItems：触发GetNItemsFromExternalSourceLambda获取250条数据和分页信息
2. 拆分批次：用Step Functions的Map状态，把数据拆成25条一组，每组并行调用SaveAllToDynamoDBLambda（单批次处理）
3. 等待所有写入完成：Map状态会自动等待所有子任务完成
4. 触发分析：调用AnalyzeDynamoDBLambda
5. 分页判断：如果还有下一页数据，回到第一步继续获取和写入；否则流程结束

优势： #

• 可视化的流程编排，在控制台就能看到每一步的执行状态
• 自带错误处理和重试机制，不用自己写复杂的重试逻辑
• 天然支持分页循环，适合需要多次调用外部API的场景

关键注意事项 #

• 必须处理UnprocessedItems：DynamoDB的BatchWriteItem不会自动重试失败的项，一定要手动处理，否则会丢失数据
• 指数退避重试：避免频繁重试触发DynamoDB的限流，建议用指数退避算法（比如1s、2s、4s、8s的间隔）
• 数据一致性：BatchWriteItem是半原子的——每个PutRequest是独立的，部分失败不会回滚已成功的项，所以要确保重试逻辑能保证最终一致性
• 监控告警：用CloudWatch监控DynamoDB的写入指标（比如WriteThrottleEvents）、Lambda的执行错误、SQS的消息积压情况，及时发现问题

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