如何用boto3将CSV数据追加至AWS DynamoDB而非覆盖现有数据

你的核心问题是当前用put_item会覆盖主键重复的现有数据，要实现“空表全插，非空追加不覆盖”的逻辑，以下是具体修改方案：

问题根源 #

你当前代码里的put_item是DynamoDB的无条件写入操作：如果目标主键不存在就插入新条目，如果已经存在就完全替换原有条目，这就是你觉得“删除现有数据”的原因——其实是被同主键的CSV数据覆盖了。

修改后的完整代码 #

import boto3
import pandas as pd
import json

# 读取CSV并转为DynamoDB接受的格式
AdvgCountries_json = json.loads(
    pd.read_csv('data.csv').to_json(orient='records')
)
lst_Dics = [{'item': AdvgCountries_json, 'table':'xxxxxxx'}]

# 初始化DynamoDB资源
dynamodb = boto3.resource('dynamodb')

def is_table_empty(tablename):
    """判断表是否为空，用scan(Limit=1)确保准确性"""
    table = dynamodb.Table(tablename)
    response = table.scan(Limit=1)
    return len(response['Items']) == 0

def insertDynamoItem(tablename, item_lst, skip_existing=True):
    """批量插入数据，可选择跳过已存在主键的条目"""
    dynamoTable = dynamodb.Table(tablename)
    # 使用batch_writer提升批量写入效率
    with dynamoTable.batch_writer() as batch:
        for record in item_lst:
            try:
                if skip_existing:
                    # ********注意替换成你表的分区键名称********
                    # 比如你的表主键是'id'，就改成attribute_not_exists(id)
                    batch.put_item(
                        Item=record,
                        ConditionExpression='attribute_not_exists(your_partition_key)'
                    )
                else:
                    batch.put_item(Item=record)
            except Exception as e:
                print(f"条目插入失败: {record}, 错误信息: {e}")
    print('批量操作完成')

# 主逻辑：判断表状态后执行对应操作
for element in lst_Dics:
    table_name = element['table']
    items = element['item']
    if is_table_empty(table_name):
        print(f"表 {table_name} 为空，插入所有CSV数据")
        insertDynamoItem(table_name, items, skip_existing=False)
    else:
        print(f"表 {table_name} 非空，追加CSV数据（跳过已存在主键的条目）")
        insertDynamoItem(table_name, items, skip_existing=True)

关键说明 #

1. 表空判断：用scan(Limit=1)而不是describe_table的ItemCount，因为后者是AWS的近似统计值，最多有15分钟延迟，而前者会实际查询一条数据，结果更准确。
2. 避免覆盖的核心：ConditionExpression='attribute_not_exists(your_partition_key)'这个条件会告诉DynamoDB：只有当这条数据的主键（分区键）不存在时才执行插入，完全不会影响现有数据。
3. 批量写入优化：用batch_writer代替逐条调用put_item，它会自动将请求打包成批量操作，减少API调用次数，提升上传速度。

注意事项 #

• 必须把代码里的your_partition_key替换成你自己DynamoDB表的分区键名称（比如你的表主键是country_code就改成这个）。
• 如果你的表是复合主键（分区键+排序键），条件表达式需要同时检查两个键：attribute_not_exists(分区键) AND attribute_not_exists(排序键)。

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