如何用Boto3在Lambda中获取DynamoDB各分区键的最新记录

注：这和你熟悉的MySQL中SELECT * FROM table GROUP BY Device ORDER BY TimeStamp DESC LIMIT 1的思路类似，但DynamoDB作为NoSQL数据库不支持直接的GROUP BY聚合，因此需要分两步实现：先获取所有唯一分区键，再逐个查询每个分区的最新记录。

核心思路 #

DynamoDB的Query操作仅针对单个分区键生效，要获取每个分区的最新排序键记录，需：

1. 收集所有唯一的Device（分区键）值；
2. 对每个Device执行Query，通过倒序排序+限制返回1条快速拿到最新记录。

步骤与代码实现 #

1. 获取唯一分区键值 #

如果表数据量较小，直接用Scan+投影表达式只获取Device字段即可；若数据量较大，建议创建仅包含分区键的全局二级索引（GSI）来优化性能，避免全表扫描开销。

2. Lambda函数完整代码 #

import boto3

# 初始化DynamoDB资源
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('你的表名')  # 替换为实际表名

def lambda_handler(event, context):
    # 第一步：获取所有唯一的Device
    unique_devices = set()
    # 使用分页器处理大量数据
    paginator = table.get_paginator('scan')
    for page in paginator.paginate(
        ProjectionExpression='Device',
        Select='SPECIFIC_ATTRIBUTES'
    ):
        for item in page['Items']:
            unique_devices.add(item['Device'])
    
    # 第二步：查询每个Device的最新记录
    latest_records = []
    for device in unique_devices:
        query_response = table.query(
            KeyConditionExpression=boto3.dynamodb.conditions.Key('Device').eq(device),
            ScanIndexForward=False,  # 按TimeStamp降序排列，最新记录在前
            Limit=1  # 只返回第一条（最新）记录
        )
        if query_response['Items']:
            latest_records.append(query_response['Items'][0])
    
    # 整理成预期的输出格式
    output_lines = []
    for record in latest_records:
        output_lines.append(f"{record['Device']}\t{record['TimeStamp']}\t{record['REMARKS']}")
    
    # 输出结果
    result = '\n'.join(output_lines)
    print(result)
    return {
        'statusCode': 200,
        'body': result
    }

关键细节 #

• 排序键有效性：确保TimeStamp字段是可排序类型（如ISO格式字符串、数字时间戳），否则倒序排序无法正确获取最新记录；
• 性能优化：数据量大时，创建GSI（分区键为Device，无排序键），将第一步的Scan替换为对GSI的Query，能大幅提升效率；
• 权限配置：Lambda的IAM角色需拥有DynamoDB的Scan（或GSI的Query）和Query权限。

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