如何用Python Lambda以status为分区键查询非主键的DynamoDB

一、先确认你的GSI结构 #

首先要确保你创建的GSI满足以下要求：

• 把status设为GSI的分区键（Partition Key）
• 投影属性（Projection）至少包含user_id（只选需要的字段能大幅提升查询性能）
• 记好这个GSI的名称（比如叫Status-Active-Index，后面代码里要精准用到）

二、Python Lambda函数实现代码 #

用boto3编写查询逻辑时，推荐用query操作（比scan高效得多，因为是利用索引精准匹配），以下是完整示例：

import boto3

def lambda_handler(event, context):
    # 初始化DynamoDB资源
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('Testing-Table')
    
    # 核心查询参数：指定GSI和查询条件
    query_params = {
        'IndexName': 'Status-Active-Index',  # 替换成你实际的GSI名称
        'KeyConditionExpression': boto3.dynamodb.conditions.Key('status').eq('active'),
        'ProjectionExpression': 'user_id'  # 只返回需要的user_id字段，减少数据传输
    }
    
    # 处理分页：DynamoDB单次查询最多返回1MB数据，需循环获取全部结果
    all_active_users = []
    response = table.query(**query_params)
    all_active_users.extend(response['Items'])
    
    while 'LastEvaluatedKey' in response:
        query_params['ExclusiveStartKey'] = response['LastEvaluatedKey']
        response = table.query(**query_params)
        all_active_users.extend(response['Items'])
    
    # 提取所有user_id并返回
    active_user_ids = [item['user_id'] for item in all_active_users]
    return {
        'statusCode': 200,
        'body': active_user_ids
    }

三、关键细节说明 #

• IndexName：必须和你创建的GSI名称完全一致，否则会抛出「索引不存在」的错误
• KeyConditionExpression：这里用Key('status').eq('active')精准匹配GSI的分区键，这是query操作的核心，相比scan能节省大量读写资源
• ProjectionExpression：指定只返回user_id，避免传输不必要的字段，降低延迟和成本
• 分页处理：如果你的active用户数量较多，单次查询返回不完，需要通过LastEvaluatedKey循环获取所有结果

四、重要注意事项 #

• 表的主键user_id是表的唯一标识，无法被替代，它负责数据的基础存储和主键查询；GSI是附加索引，专门用来支持这种非主键的查询场景
• GSI会产生额外的读写费用，且数据同步有毫秒级的轻微延迟，但对于你的查询需求来说完全可以接受
• 务必确保GSI的投影属性包含user_id，否则查询结果里不会出现这个字段

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