DynamoDB PutItem请求上限及批量插入REST API实现问询

1. DynamoDB PutItem 请求的相关限制 #

• PutItem 是单条数据插入的API，单条请求仅能插入1条Item，Item的最大大小为400KB（包含所有属性名和属性值的字节数）。
• 若要批量插入，推荐使用 BatchWriteItem API：
  • 单次请求最多可包含25个PutItem/DeleteItem操作
  • 单次请求的总Payload大小不能超过16MB

2. 批量插入功能实现示例 #

实现思路 #

1. 先对批量数据做统一验证（比如检查每条数据是否包含id字段）
2. 将数据按25条一组拆分（适配BatchWriteItem的限制）
3. 调用BatchWriteItem批量插入，同时处理未成功插入的重试逻辑

代码示例 #

import boto3
from botocore.exceptions import ClientError
import logging

logger = logging.getLogger(__name__)

class DynamoDBHandler:
    def __init__(self, table_name):
        self.dynamodb = boto3.resource('dynamodb')
        self.table = self.dynamodb.Table(table_name)

    def add_item(self, item):
        try:
            if item.get("id") is not None:
                self.table.put_item(Item=item)
        except ClientError as err:
            logger.exception(f"Add item failed in table {self.table.name}.")
            raise err

    def validate_batch_items(self, items):
        """验证批量数据，返回有效数据和无效数据"""
        valid_items = []
        invalid_items = []
        for idx, item in enumerate(items):
            if item.get("id") is None:
                invalid_items.append({"index": idx, "item": item, "reason": "Missing required 'id' field"})
            else:
                valid_items.append(item)
        return valid_items, invalid_items

    def batch_add_items(self, items):
        # 第一步：验证数据
        valid_items, invalid_items = self.validate_batch_items(items)
        if invalid_items:
            logger.warning(f"Found {len(invalid_items)} invalid items: {invalid_items}")
            # 可根据业务需求选择抛出异常、跳过无效数据或返回错误信息

        # 第二步：按25条一组拆分数据
        batch_size = 25
        batches = [valid_items[i:i+batch_size] for i in range(0, len(valid_items), batch_size)]

        # 第三步：批量插入并处理重试
        for batch in batches:
            request_items = {
                self.table.name: [
                    {"PutRequest": {"Item": item}} for item in batch
                ]
            }
            try:
                response = self.dynamodb.batch_write_item(RequestItems=request_items)
                # 处理未成功插入的项
                unprocessed = response.get('UnprocessedItems', {})
                while unprocessed:
                    logger.warning(f"Retrying unprocessed items")
                    response = self.dynamodb.batch_write_item(RequestItems=unprocessed)
                    unprocessed = response.get('UnprocessedItems', {})
            except ClientError as err:
                logger.exception(f"Batch add failed in table {self.table.name}.")
                raise err

        return {
            "success_count": len(valid_items),
            "failed_count": len(invalid_items),
            "invalid_items": invalid_items
        }

使用说明 #

• validate_batch_items 方法可根据业务需求扩展验证规则（比如字段类型、格式检查）
• 相比循环调用put_item，BatchWriteItem能大幅减少API请求次数，提升插入效率
• 内置的重试逻辑会自动处理DynamoDB返回的未处理项，保证数据插入可靠性

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