如何在AWS DynamoDB中按hash key列表+多条件过滤查询数据？

这确实是DynamoDB中挺常见的一个场景——当你需要匹配多个哈希键，同时还要附加一个状态过滤条件时，QUERY（单哈希键限制）和BatchGetItem（不支持过滤条件）都没法直接满足需求。下面给你几个不同的解决方案，你可以根据自己的业务场景来选择：

方案1：使用Scan操作（适合小表/离线场景） #

Scan是最直接的方式，你可以通过FilterExpression同时指定哈希键的IN列表和status的匹配条件。不过要注意，Scan会全表扫描，性能较差，消耗大量的读取容量单位（RCU），所以只适合数据量不大的表，或者非实时的离线任务。

代码示例（Python Boto3） #

import boto3

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('your-table-name')

# 初始Scan请求
response = table.scan(
    FilterExpression='#hk IN (:hash_list) AND #status = :approved',
    ExpressionAttributeNames={
        '#hk': 'your-hash-key-name',  # 替换成你的哈希键名称
        '#status': 'status'
    },
    ExpressionAttributeValues={
        ':hash_list': ['key1', 'key2', 'key3'],  # 替换成你的哈希键列表
        ':approved': 'approved'
    }
)

items = response['Items']

# 处理分页结果（如果表数据量大，Scan会分页返回）
while 'LastEvaluatedKey' in response:
    response = table.scan(
        FilterExpression='#hk IN (:hash_list) AND #status = :approved',
        ExpressionAttributeNames={
            '#hk': 'your-hash-key-name',
            '#status': 'status'
        },
        ExpressionAttributeValues={
            ':hash_list': ['key1', 'key2', 'key3'],
            ':approved': 'approved'
        },
        ExclusiveStartKey=response['LastEvaluatedKey']
    )
    items.extend(response['Items'])

print(f"找到符合条件的条目：{items}")

方案2：并行执行多个QUERY请求（无需修改数据结构，适合中等规模哈希键列表） #

虽然单个QUERY只能针对一个哈希键，但你可以在客户端对哈希键列表中的每个键发起独立的QUERY，然后合并结果。每个QUERY会高效定位到对应哈希键的分区，再通过FilterExpression过滤出status为"approved"的条目。这种方式比Scan高效得多，且不需要修改表结构。

代码示例（Python Boto3 + 线程池） #

import boto3
from concurrent.futures import ThreadPoolExecutor

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('your-table-name')
hash_key_list = ['key1', 'key2', 'key3']  # 替换成你的哈希键列表

def query_single_hash_key(hash_key):
    """查询单个哈希键下status为approved的条目"""
    response = table.query(
        KeyConditionExpression='#hk = :key',
        FilterExpression='#status = :approved',
        ExpressionAttributeNames={
            '#hk': 'your-hash-key-name',
            '#status': 'status'
        },
        ExpressionAttributeValues={
            ':key': hash_key,
            ':approved': 'approved'
        }
    )
    return response['Items']

# 控制并发数，避免触发DynamoDB限流（根据你的表容量调整）
with ThreadPoolExecutor(max_workers=5) as executor:
    results = executor.map(query_single_hash_key, hash_key_list)

# 合并所有结果
all_matching_items = []
for result in results:
    all_matching_items.extend(result)

print(f"找到符合条件的条目：{all_matching_items}")

方案3：创建全局二级索引（GSI）（适合频繁查询的生产场景） #

如果这类查询是你的高频操作，最推荐的方案是创建一个全局二级索引，重新组织数据结构来适配查询需求。你可以把status设为GSI的哈希键，原来的哈希键设为GSI的排序键——这样就能用QUERY直接查询status="approved"且排序键在指定列表中的条目，效率最高。

步骤1：创建GSI #

# 给现有表添加GSI（如果还没创建）
table.update(
    AttributeDefinitions=[
        {
            'AttributeName': 'status',
            'AttributeType': 'S'
        },
        {
            'AttributeName': 'your-hash-key-name',
            'AttributeType': 'S'
        }
    ],
    GlobalSecondaryIndexUpdates=[
        {
            'Create': {
                'IndexName': 'Status-HashKey-Index',
                'KeySchema': [
                    {
                        'AttributeName': 'status',
                        'KeyType': 'HASH'
                    },
                    {
                        'AttributeName': 'your-hash-key-name',
                        'KeyType': 'RANGE'
                    }
                ],
                'Projection': {
                    'ProjectionType': 'ALL'  # 按需投影字段可减少存储开销
                },
                'ProvisionedThroughput': {
                    'ReadCapacityUnits': 5,
                    'WriteCapacityUnits': 5
                }
            }
        }
    ]
)

步骤2：通过GSI执行查询 #

response = table.query(
    IndexName='Status-HashKey-Index',
    KeyConditionExpression='#status = :approved AND #hk IN (:hash_list)',
    ExpressionAttributeNames={
        '#status': 'status',
        '#hk': 'your-hash-key-name'
    },
    ExpressionAttributeValues={
        ':approved': 'approved',
        ':hash_list': ['key1', 'key2', 'key3']
    }
)

items = response['Items']

# 处理分页结果
while 'LastEvaluatedKey' in response:
    response = table.query(
        IndexName='Status-HashKey-Index',
        KeyConditionExpression='#status = :approved AND #hk IN (:hash_list)',
        ExpressionAttributeNames={
            '#status': 'status',
            '#hk': 'your-hash-key-name'
        },
        ExpressionAttributeValues={
            ':approved': 'approved',
            ':hash_list': ['key1', 'key2', 'key3']
        },
        ExclusiveStartKey=response['LastEvaluatedKey']
    )
    items.extend(response['Items'])

print(f"找到符合条件的条目：{items}")

方案对比 #

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