DynamoDB PutItem中attribute_not_exists条件持续失败求助

核心错误原因 #

1. 异常处理范围错误：你把try-except包裹了整个循环，只要某一次PutItem触发ConditionalCheckFailedException，整个循环就会中断，后续所有数据都无法写入，这是导致大量数据缺失的主要原因。
2. 条件表达式逻辑的理解偏差：你的条件表达式attribute_not_exists(ItemId) OR attribute_not_exists(ItemType)本身逻辑是对的（仅当(ItemId, ItemType)组合的项不存在时才写入），但异常处理方式导致单次失败就终止流程，掩盖了问题本质。

解决方案 #

根据你的需求选择以下两种方案：

方案1：仅写入不存在的(ItemId, ItemType)组合，重复项跳过 #

调整异常处理到循环内部，确保单个项失败不影响后续写入：

import boto3
import pandas as pd
from botocore.exceptions import ClientError

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(table_name)  # 替换为你的表名

t_df = df[df['product'] == product_category]

# 直接遍历DataFrame行，避免列表长度不一致问题
for idx, row in t_df.iterrows():
    item_id = row['id']
    item_type = row['transaction_detail']
    try:
        table.put_item(
            Item={
                'ItemId': item_id,
                'ItemType': item_type,
                'ProductCategory': product_category
            },
            # 条件表达式：仅当该(ItemId, ItemType)组合的项不存在时写入
            ConditionExpression='attribute_not_exists(ItemId) OR attribute_not_exists(ItemType)'
        )
        print(f"已写入: ItemId={item_id}, ItemType={item_type}")
    except ClientError as ce:
        if ce.response['Error']['Code'] == 'ConditionalCheckFailedException':
            # 项已存在，跳过
            print(f"项已存在，跳过: ItemId={item_id}, ItemType={item_type}")
        else:
            # 处理其他错误
            print(f"写入失败: {ce}")

print("所有数据处理完成")

方案2：允许覆盖已存在的(ItemId, ItemType)组合 #

如果不需要避免重复，直接移除条件表达式，PutItem本身会自动覆盖已存在的项：

import boto3
import pandas as pd
from botocore.exceptions import ClientError

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(table_name)  # 替换为你的表名

t_df = df[df['product'] == product_category]

for idx, row in t_df.iterrows():
    item_id = row['id']
    item_type = row['transaction_detail']
    try:
        table.put_item(
            Item={
                'ItemId': item_id,
                'ItemType': item_type,
                'ProductCategory': product_category
            }
            # 移除ConditionExpression，存在则覆盖，不存在则写入
        )
        print(f"已写入/更新: ItemId={item_id}, ItemType={item_type}")
    except ClientError as ce:
        print(f"处理失败: {ce}")

print("所有数据处理完成")

额外优化建议 #

• 避免使用zip_longest遍历列表：直接遍历DataFrame的行更直观，也能避免两个列表长度不一致时出现None值写入的问题。
• 批量写入优化：如果数据量很大，建议使用batch_writer()提升写入效率，减少API调用次数：

with table.batch_writer() as batch:
    for idx, row in t_df.iterrows():
        batch.put_item(
            Item={
                'ItemId': row['id'],
                'ItemType': row['transaction_detail'],
                'ProductCategory': product_category
            }
            # 如需条件判断，batch_writer不支持ConditionExpression，需单独处理
        )

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