如何以order_name为匹配键用Pandas DataFrame更新DynamoDB表字段

你的报错根源很明确：你直接把Pandas的Series对象传给了boto3的update_item方法，而DynamoDB的API需要单个标量值（比如字符串、数字），无法处理整个Series。另外，因为你的DynamoDB表中order_number是可重复的，你需要先找到所有匹配该order_number的记录，再逐个更新它们。

修正步骤与代码 #

1. 遍历DataFrame的每一行：对df2中的每个order_number和对应的return_status单独处理
2. 查询匹配的DynamoDB记录：因为order_number可重复，需要先获取所有该值对应的记录（如果order_number是全局二级索引(GSI)，用query效率更高；如果不是，用scan）
3. 逐个更新记录：用每条记录的主键（比如假设你的表有一个分区键id，如果是复合主键需要包含分区键+排序键）来执行更新

以下是修正后的完整代码：

import boto3
from boto3.dynamodb.conditions import Key
import pandas as pd
import pprint  # 补充你之前遗漏的pprint导入

# Connect to ETL DB
# 保留你原有的数据加载逻辑
df1 = run_sql_df('SET NOCOUNT ON; select distinct order_number, return_status from etl_db.dbo.shopify_returns_portal')
df1 = df1.astype({"order_number":"str","return_status":"str"})
filename = 'test_file.csv'
df1.to_csv(str(filename),index=False)
df2 = pd.read_csv('test_file.csv')
df2 = df2.astype({"order_number":"str","return_status":"str"})

# 初始化DynamoDB资源（放在外层避免重复创建连接）
dynamodb = boto3.resource(
    "dynamodb",
    aws_access_key_id="XXXXXXXXXXXX",
    aws_secret_access_key="XXXXXXXXXXXXXXXXXX",
    region_name="eu-west-1"
)
table = dynamodb.Table('returns-portal')

def update_records_for_order(order_number, return_status):
    """更新DynamoDB中所有匹配order_number的记录的return_status"""
    # 第一步：查询所有order_number匹配的记录
    # 注意：如果order_number是GSI的分区键，用query更高效；否则替换为scan
    response = table.query(
        IndexName='order_number-index',  # 替换成你的GSI名称，无GSI则用scan
        KeyConditionExpression=Key('order_number').eq(order_number)
    )
    items = response.get('Items', [])
    
    # 处理分页结果（如果查询结果超过1MB会返回LastEvaluatedKey）
    while 'LastEvaluatedKey' in response:
        response = table.query(
            IndexName='order_number-index',
            KeyConditionExpression=Key('order_number').eq(order_number),
            ExclusiveStartKey=response['LastEvaluatedKey']
        )
        items.extend(response.get('Items', []))
    
    # 第二步：逐个更新每条记录
    update_results = []
    for item in items:
        # 替换为你的表的实际主键！！！
        # 如果是复合主键，需要同时填写分区键和排序键，比如：
        # 'partition_key': item['partition_key'],
        # 'sort_key': item['sort_key']
        try:
            update_response = table.update_item(
                Key={
                    'id': item['id']
                },
                UpdateExpression="SET return_status = :r",
                ExpressionAttributeValues={
                    ':r': return_status
                },
                ReturnValues="UPDATED_NEW"
            )
            update_results.append(update_response)
            print(f"成功更新order_number {order_number}的记录：{item['id']}")
        except Exception as e:
            print(f"更新order_number {order_number}的记录{item['id']}失败：{str(e)}")
    
    return update_results

if __name__ == '__main__':
    # 遍历df2的每一行处理更新
    for _, row in df2.iterrows():
        order_num = row['order_number']
        status = row['return_status']
        print(f"\n开始处理order_number: {order_num}")
        results = update_records_for_order(order_num, status)
        pprint.pprint(results, sort_dicts=False)

关键注意事项 #

• 主键字段替换：必须把代码中的'id': item['id']替换成你的DynamoDB表的实际主键（分区键，或分区键+排序键），这是update_item的必填项，DynamoDB需要唯一标识一条记录才能更新。
• 无GSI时的替代方案：如果你的表没有针对order_number创建GSI，query会报错，此时需要换成scan（但scan效率较低，数据量大的话建议创建GSI），scan的写法如下：

  response = table.scan(
      FilterExpression=Key('order_number').eq(order_number)
  )
  

• 批量操作优化：如果数据量很大，建议使用batch_write_item来提升性能，但要注意DynamoDB的批量操作限制（每次最多25个操作）。
• 权限验证：确保你的AWS密钥拥有DynamoDB的query/scan和update_item权限。

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