如何在删除DynamoDB条目前检查重复或获取条目计数？

一、检查主键重复数据 #

由于你的表使用id（分区键）+sortKey（排序键）作为复合主键，DynamoDB本身会强制主键唯一性，理论上表中不会存在主键重复的条目。如果要验证或排查潜在问题，可以通过以下方式：

1. 针对待删除条目集合检查 #

如果你的待删除条目列表是预定义的，先在本地对列表去重（比如用Python的集合或字典），避免重复提交删除请求。之后可以用BatchGetItem批量查询这些主键是否存在，同时确认是否有意外的重复：

import boto3

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('sometable')

# 待删除的主键列表（示例）
delete_keys = [
    {'id': 'abcd', 'sortKey': 'efgh'},
    {'id': 'abcd', 'sortKey': 'efgh'}  # 故意重复的条目
]

# 本地去重：将主键转为元组存入集合，再转回字典
unique_keys = list({(k['id'], k['sortKey']): k for k in delete_keys}.values())

# 批量查询存在性
response = table.batch_get_item(
    RequestItems={
        'sometable': {
            'Keys': unique_keys
        }
    }
)

# 实际存在的条目数量
existing_count = len(response['Responses']['sometable'])
print(f"待删除列表中实际存在的条目数：{existing_count}")

2. 全表扫描排查重复（仅小表适用） #

如果要全表检查是否有主键重复（虽然DynamoDB不会允许，但可能因客户端重复写入导致的异常？不过实际主键重复写入会报错），可以用扫描配合聚合统计，但大表不推荐，因为扫描成本高且慢：

from collections import defaultdict

response = table.scan()
item_counts = defaultdict(int)

for item in response['Items']:
    pk = (item['id'], item['sortKey'])
    item_counts[pk] += 1

# 输出重复的主键（如果有的话）
duplicates = [pk for pk, count in item_counts.items() if count > 1]
if duplicates:
    print(f"发现重复主键：{duplicates}")
else:
    print("表中无主键重复条目")

二、获取待删除条目的计数 #

根据你的需求场景，有两种常见方式：

1. 已知待删除主键列表时 #

用上面的BatchGetItem查询后，统计返回的条目数量，就是实际存在的待删除条目数。

2. 按条件查询待删除条目时 #

如果是按分区键或其他条件筛选要删除的条目（比如某个id下的所有sortKey），可以用Query并指定Select='COUNT'来获取计数：

# 查询id为'abcd'的所有条目数量
response = table.query(
    KeyConditionExpression='id = :id_val',
    ExpressionAttributeValues={':id_val': 'abcd'},
    Select='COUNT'
)

count = response['Count']
print(f"待删除的条目总数：{count}")

三、批量删除的实现 #

DynamoDB的批量删除可以用batch_write_item接口，每次最多处理25个条目，需要处理返回的UnprocessedItems（如果有未处理的条目，需重试）：

def batch_delete_items(table, keys):
    # 拆分keys为每25个一组
    for i in range(0, len(keys), 25):
        batch = keys[i:i+25]
        request_items = {
            table.name: [{'DeleteRequest': {'Key': key}} for key in batch]
        }
        response = table.batch_write_item(RequestItems=request_items)
        
        # 处理未完成的条目（简单重试，实际可根据需求添加重试逻辑）
        while response.get('UnprocessedItems'):
            response = table.batch_write_item(RequestItems=response['UnprocessedItems'])

# 使用示例
batch_delete_items(table, unique_keys)

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