Python使用dynamodb.client.batch_write_item导入大型CSV到DynamoDB报错排查

错误根因 #

你的代码存在三个核心问题：

1. 请求结构不符合接口规范
   batch_write_item 要求RequestItems下对应表名的列表中，每个元素对应单条记录的写入请求。你直接将所有记录的列表传入了单个PutRequest的Item字段，导致参数类型校验失败（接口期望Item是单条记录的字典，实际传入了列表）。
2. 低阶客户端需要显式声明字段类型
   你使用的boto3.client('dynamodb')属于低阶API，要求Item的每个字段都要明确指定数据类型（例如字符串格式为{'S': '字段值'}，数字格式为{'N': '数字字符串'}），不能直接传入Python原生的int、str类型。
3. 未做批次拆分
   batch_write_item单次请求最多支持25条记录写入，你一次性提交全量数据即使结构正确也会触发接口限制。

修复方案 #

推荐使用更易用的DynamoDB高阶Resource接口，不需要手动处理字段类型，代码修改如下：

1. 替换DynamoDB实例初始化逻辑：

# 将原有的dynamodb_client初始化替换为高阶Resource
dynamodb = boto3.resource('dynamodb', region_name='ap-south-1',
    aws_access_key_id=response['Credentials']['AccessKeyId'],
    aws_secret_access_key=response['Credentials']['SecretAccessKey'],
    aws_session_token = response['Credentials']['SessionToken'])
table = dynamodb.Table("sample_table")

2. 重写批量写入函数：

def batch_write(items):
    batch_size = 25
    # 按25条为一批拆分数据
    for idx in range(0, len(items), batch_size):
        batch_chunk = items[idx:idx+batch_size]
        # 自动处理批次提交逻辑
        with table.batch_writer() as writer:
            for item in batch_chunk:
                writer.put_item(Item=item)
    print("全量数据写入完成")

如果必须使用低阶client接口，需要额外增加字段类型转换逻辑，同时调整请求结构：

# 字段类型转换函数
def to_dynamo_type(item):
    dynamo_item = {}
    for key, value in item.items():
        if isinstance(value, str):
            dynamo_item[key] = {'S': value}
        elif isinstance(value, int):
            dynamo_item[key] = {'N': str(value)}
        # 其他数据类型可按需求扩展
    return dynamo_item

def batch_write(items):
    table_name = "sample_table"
    batch_size = 25
    for idx in range(0, len(items), batch_size):
        batch_chunk = items[idx:idx+batch_size]
        # 构造符合要求的请求结构
        request_list = [
            {"PutRequest": {"Item": to_dynamo_type(item)}}
            for item in batch_chunk
        ]
        try:
            dynamodb_client.batch_write_item(RequestItems={
                table_name: request_list
            })
        except Exception as e:
            print(f"批次{idx//batch_size}写入失败: {e}")

内容的提问来源于stack exchange，提问作者Amol P Suryavanshi