使用Boto3迁移DynamoDB数据遇嵌套字典参数验证失败求助

问题背景 #

使用Python boto3迁移DynamoDB数据时，已处理Decimal类型，但嵌套字典属性（如personalInfo）写入目标表时触发参数验证错误，提示嵌套字段不符合DynamoDB参数格式。

错误原因 #

代码中使用了boto3低级客户端（boto3.client('dynamodb')），该客户端要求严格遵循DynamoDB原生数据格式：

• 嵌套字典必须用M类型标识，直接写入字段名会被判定为非法参数
• 手动构造Item时，所有层级都要显式指定类型（如S字符串、M映射），嵌套结构遗漏M标识就会报错

另外代码中存在冗余操作：通过JSON序列化反序列化处理Decimal，既没必要还可能引入额外问题。

解决方案 #

核心优化点 #

1. 改用boto3资源客户端（boto3.resource('dynamodb')）：资源客户端支持自动将Python原生类型（包括嵌套字典、Decimal）映射为DynamoDB格式，无需手动指定S/M等类型标识。
2. 简化Decimal处理：资源客户端默认支持Decimal类型，无需JSON序列化反序列化；如果需要兼容JSON，可使用simplejson替代标准json模块，它能直接序列化Decimal。
3. 修复删除数据逻辑：原代码错误调用scan结果的delete_item，应使用源表对象的delete_item方法。

修改后的完整代码 #

import boto3
import argparse
import simplejson as json
from decimal import Decimal
from boto3.dynamodb.conditions import Key, Attr

def migrate_data(region, sourcetable, desttable):
    try:
        # 初始化资源客户端，自动处理类型映射
        dynamodb = boto3.resource('dynamodb', region_name=region)
        source_table = dynamodb.Table(sourcetable)
        dest_table = dynamodb.Table(desttable)

        # 扫描源表获取目标数据
        sresponse = source_table.scan(
            FilterExpression=Attr('expirationDate').contains('2021-') & Attr('entityType').eq('prospect')
        )
        data = sresponse['Items']

        print(f"## 2021年记录数量: {len(data)}")

        # 写入目标表
        print(f"## 正在将数据从{sourcetable}迁移到{desttable}")
        for item in data:
            # 资源客户端自动处理嵌套字典和Decimal，无需手动构造类型
            dest_table.put_item(Item=item)
            print(f"## 已写入记录ID: {item['id']}")

        print(f"## 成功迁移{len(data)}条记录")

        # 从源表删除已迁移数据
        print("## 正在删除源表中的已迁移记录")
        for item in data:
            source_table.delete_item(Key={'id': item['id']})
            print(f"## 已删除记录ID: {item['id']}")

    except Exception as e:
        print(f"## 迁移过程中出错: {str(e)}")

if __name__ == '__main__':
    print("###################     DynamoDB表数据迁移工具       ####################\n")

    parser = argparse.ArgumentParser(description="DynamoDB表数据迁移工具")
    parser.add_argument("-r", "--region", dest="region", default="us-east-1", help="AWS区域，默认us-east-1")
    parser.add_argument("-st", "--sourcetable", dest="sourcetable", required=True, help="源表名称")
    parser.add_argument("-dt", "--desttable", dest="desttable", required=True, help="目标表名称")

    args = parser.parse_args()
    migrate_data(args.region, args.sourcetable, args.desttable)

关键说明 #

• 资源客户端的put_item方法会自动将Python字典转换为DynamoDB的M类型，嵌套字典无需额外处理
• 使用simplejson替代标准json模块，是为了如果后续需要序列化数据（比如日志）时能直接处理Decimal，避免报错
• 如果必须使用低级客户端，嵌套字典需要显式用M包裹，示例如下（不推荐，仅作参考）：

# 低级客户端嵌套字典写法示例
'personalInfo': {
    'M': {
        'firstName': {'S': jitem['personalInfo']['firstName']},
        'lastName': {'S': jitem['personalInfo']['lastName']},
        'residentialAddress': {
            'M': {
                'zip': {'S': jitem['personalInfo']['residentialAddress']['zip']}
                # 其他字段同理
            }
        }
    }
}

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