如何在Boto3中为DynamoDB Query方法正确配置KeyCondition

为什么会出现Query key condition not supported错误？ #

DynamoDB的Query操作有严格的规则：KeyConditionExpression必须包含分区键（Partition Key）的精确匹配，只有当你同时使用了排序键（Sort Key）时，才能对排序键使用范围运算符（比如gt、lt、between等）。

从你的描述和代码来看，Id应该是表的分区键，这时候你用Key('Id').gt(0)是不合法的——分区键在Query里只能用eq做精确匹配，不能用范围条件，这就是报错的根源。

解决方案1：调整表结构，让Query可以获取全表数据 #

如果你的目标是获取全表数据但不想用Scan，最直接的方法是调整表的键设计：

1. 新增一个固定值的分区键（比如GlobalPartition），所有数据的这个字段都设为同一个值（比如all_records）
2. 将原来的Id设为排序键（Sort Key）

这样调整后，你就可以用合法的KeyConditionExpression来执行Query获取全表数据了，修改后的代码如下：

def list_items(self):
    ProjectionExpression = "Id,#Name,Description"
    ean = {"#Name": "Name"}
    # 现在分区键是GlobalPartition，精确匹配固定值，排序键Id用范围条件
    esk = Key('GlobalPartition').eq('all_records') & Key('Id').gt(0)
    limit = settings.AWS_SCAN_LIMIT
    items = []
    tableResponse = table.query(
        ProjectionExpression=ProjectionExpression,
        ExpressionAttributeNames=ean,
        KeyConditionExpression=esk,
        Limit=limit
    )
    items.extend(tableResponse['Items'])
    while 'LastEvaluatedKey' in tableResponse:
        tableResponse = table.query(
            ProjectionExpression=ProjectionExpression,
            ExpressionAttributeNames=ean,
            ExclusiveStartKey=tableResponse['LastEvaluatedKey'],
            KeyConditionExpression=esk,
            Limit=limit
        )
        items.extend(tableResponse['Items'])
    return items

解决方案2：使用并行Scan提升全表扫描效率 #

如果不想修改表结构，那只能用Scan，但可以通过并行扫描来大幅提升速度。并行扫描的原理是把表分成多个逻辑段（Segments），同时用多个线程/进程扫描不同的段，最后合并结果。

下面是并行Scan的实现示例（用Python的concurrent.futures实现多线程）：

import concurrent.futures
from boto3.dynamodb.conditions import Attr

def scan_segment(segment, total_segments):
    """扫描单个段的函数"""
    items = []
    ProjectionExpression = "Id,#Name,Description"
    ean = {"#Name": "Name"}
    limit = settings.AWS_SCAN_LIMIT
    response = table.scan(
        ProjectionExpression=ProjectionExpression,
        ExpressionAttributeNames=ean,
        Limit=limit,
        Segment=segment,
        TotalSegments=total_segments
    )
    items.extend(response['Items'])
    while 'LastEvaluatedKey' in response:
        response = table.scan(
            ProjectionExpression=ProjectionExpression,
            ExpressionAttributeNames=ean,
            ExclusiveStartKey=response['LastEvaluatedKey'],
            Limit=limit,
            Segment=segment,
            TotalSegments=total_segments
        )
        items.extend(response['Items'])
    return items

def list_items(self):
    total_segments = 4  # 根据你的并发能力调整，比如4、8、16等
    items = []
    # 用线程池执行并行扫描
    with concurrent.futures.ThreadPoolExecutor(max_workers=total_segments) as executor:
        futures = [executor.submit(scan_segment, seg, total_segments) for seg in range(total_segments)]
        for future in concurrent.futures.as_completed(futures):
            items.extend(future.result())
    return items

注意事项 #

• 并行扫描的TotalSegments值不要设置过大，避免超出DynamoDB的并发限制，一般建议根据你的表大小和业务场景设置为4-16之间。
• 如果你的表有大量数据，并行扫描相比单线程Scan能显著减少耗时，但还是会消耗较多的读取容量单位（RCU），需要注意成本。

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