批量导入CSV至AWS DynamoDB时如何避免限流与Lambda超时？

核心问题分析 #

原代码存在两个关键瓶颈：

• 一次性将整个CSV加载到内存，既浪费资源又增大单批次写入压力
• 未配置DynamoDB写入的重试策略，遇到限流时直接失败，没有自动重试逻辑

具体优化措施 #

1. 分块读取CSV，避免内存过载

不要一次性加载全量数据，改用pandas的chunksize参数分块读取，每块处理完成后再加载下一块，降低内存占用。

2. 启用DynamoDB自适应重试（带指数退避+抖动）

Boto3支持adaptive重试模式，会自动实现指数退避加抖动，比默认的standard模式更适合批量写入场景。通过配置DynamoDB客户端的重试参数，让SDK自动处理限流错误。

3. 调整Batch Writer参数

显式设置batch_writer的max_items和flush_amount，控制单次批量写入的条目数，避免触发更严格的限流。

4. 优化Lambda配置

• 提高Lambda内存配置（建议至少1024MB），内存越高CPU性能越强，处理速度更快
• 确认Lambda超时时间设置为最大值15分钟（如果仍不够，考虑用Step Functions拆分任务）

修改后的完整代码 #

import json
import io
import csv 
import boto3 
import pandas as pd
from decimal import Decimal
from botocore.config import Config

def lambda_handler(event, context):
    # 配置DynamoDB客户端，启用自适应重试（带指数退避+抖动）
    dynamodb_config = Config(
        retries={
            'max_attempts': 10,
            'mode': 'adaptive'
        }
    )
    s3_client = boto3.client("s3")
    dynamodb = boto3.resource('dynamodb', config=dynamodb_config)
    table = dynamodb.Table('batch-gun-data-test')

    bucket_name = event['Records'][0]['s3']['bucket']['name']
    s3_file_name = event['Records'][0]['s3']['object']['key']
    csvfile = s3_client.get_object(Bucket=bucket_name, Key=s3_file_name)
    data = io.BytesIO(csvfile["Body"].read())

    # 分块读取CSV，每块500条（可根据WCU调整）
    chunk_size = 500
    for df_chunk in pd.read_csv(data, sep='\t', chunksize=chunk_size):
        df_chunk = df_chunk[['SKU', 'SubCategory']].copy()
        
        # 配置Batch Writer，控制批量大小
        with table.batch_writer(max_items=25, flush_amount=10) as batch:
            for record in df_chunk.to_dict("records"):
                # 处理数值类型转换（如果需要）
                processed_record = json.loads(json.dumps(record), parse_float=Decimal)
                batch.put_item(Item=processed_record)

额外说明 #

• adaptive重试模式会根据DynamoDB的限流信号自动调整重试间隔，加入抖动避免请求风暴
• 分块大小chunk_size和batch参数max_items可根据你的WCU配置调整：WCU越高，可适当增大数值
• 如果CSV中有数值类型（比如浮点数），必须转换为Decimal类型才能写入DynamoDB，原代码注释掉的转换逻辑需要启用

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