如何通过Python与boto3查询DynamoDB指定用户的最大timestamp？

这个需求完全可行，具体实现取决于你的DynamoDB表结构，以下是两种常见场景的处理方式：

1. 最优场景：表以userid为分区键、timestamp为排序键 #

如果你的transaction表设计为分区键（Partition Key）= userid，排序键（Sort Key）= timestamp，可以直接用query操作高效获取结果，这是DynamoDB推荐的方式：

import boto3

# 初始化DynamoDB客户端
dynamodb = boto3.client('dynamodb')

def get_user_max_transaction_ts(user_id):
    response = dynamodb.query(
        TableName='transaction',
        KeyConditionExpression='userid = :uid',
        ExpressionAttributeValues={':uid': {'S': str(user_id)}},
        ScanIndexForward=False,  # 按timestamp降序排列
        Limit=1  # 只取最大的那一条记录
    )
    # 判断是否存在交易记录
    if response['Items']:
        max_ts = response['Items'][0]['timestamp']['N']  # 若timestamp是字符串类型则用'S'，需匹配实际存储类型
        return f"用户{user_id}有交易记录，最新交易时间戳：{max_ts}"
    else:
        return f"用户{user_id}无交易记录"

# 调用示例
print(get_user_max_transaction_ts(123))

2. 非最优场景：表主键不是userid+timestamp #

如果你的表主键设计不符合上述结构，有两种补救方式：

• 创建全局二级索引（GSI）：新增一个GSI，以userid为分区键、timestamp为排序键，之后就能用上面的query方法高效查询，这是推荐的优化方案。
• 使用scan操作（不推荐）：如果不想修改表结构，只能用scan遍历全表筛选指定userid的记录，再手动找出最大timestamp，但数据量大时性能极差：

def get_user_max_transaction_ts_scan(user_id):
    response = dynamodb.scan(
        TableName='transaction',
        FilterExpression='userid = :uid',
        ExpressionAttributeValues={':uid': {'S': str(user_id)}}
    )
    items = response['Items']
    # 处理分页（单页返回数据超过1MB时需循环获取）
    while 'LastEvaluatedKey' in response:
        response = dynamodb.scan(
            TableName='transaction',
            FilterExpression='userid = :uid',
            ExpressionAttributeValues={':uid': {'S': str(user_id)}},
            ExclusiveStartKey=response['LastEvaluatedKey']
        )
        items.extend(response['Items'])
    
    if items:
        max_ts = max(item['timestamp']['N'] for item in items)
        return f"用户{user_id}有交易记录，最新交易时间戳：{max_ts}"
    else:
        return f"用户{user_id}无交易记录"

总结：只要你的表能通过键或索引定位到指定userid的记录，就能实现类似SQL的max(timestamp)查询，进而判断用户是否有交易记录。

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