DynamoDB单表设计咨询：跟踪数值与独立更新记录建模

嘿，很高兴你在尝试DynamoDB单表设计——这确实是遵循最佳实践的正确方向，坚持下去就对了！针对你的场景，我有一个非常贴合的设计方案，能完美解决你遇到的那些阻碍：

我们会用**复合主键（PK/SK）**来区分两种类型的条目：当前数值记录和每次的更新日志。这样既能保证数据集中存储，又能精准隔离不同类型的查询需求。

关键索引设计 #

1. 主索引（PK=TrackingID, SK=SK） #

这是表的默认索引，能满足你的核心访问需求：

• 快速获取当前数值：用GetItem操作，条件为TrackingID = "<你的跟踪ID>" AND SK = "CURRENT"，这是O(1)的高性能操作。
• 查询某个跟踪项的所有更新记录：用Query操作，条件为TrackingID = "<你的跟踪ID>" AND SK BEGINS_WITH "UPDATE#"。因为SK前缀是UPDATE#，只会返回更新日志条目，完全不会带出当前数值项——完美解决你之前的查询问题！而且SK里的ISO时间戳天然支持按时间排序（默认升序，要最新的在前就设置ScanIndexForward = false）。

2. 全局二级索引（GSI）：按变更值+时间排序 #

如果需要跨多个跟踪项查询特定变更范围的记录，或者按变更值排序，我们可以创建一个GSI：

• GSI PK：EntityType（值固定为"METRIC_UPDATE"，把所有更新日志条目聚合到同一个分区）
• GSI SK：ChangeValue#<ISO_TIMESTAMP>（比如"+5#2024-05-20T14:30:00"）

这个GSI支持两种查询：

• 按特定变更值查询所有更新记录（并按时间排序）
• 按变更值范围查询（比如变更值在0到10之间的所有记录）

每次更新数值时，必须用DynamoDB事务操作（TransactWriteItems），确保当前数值和更新日志的原子性（要么都成功，要么都失败），避免数据不一致。

示例伪代码（Python）：

import boto3
from datetime import datetime

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('MetricTrackingTable')

# 配置参数
tracking_id = "daily_sales"
new_value = 150.0
current_time = datetime.utcnow().isoformat()

# 先获取当前值（首次创建则设为0）
response = table.get_item(Key={'TrackingID': tracking_id, 'SK': 'CURRENT'})
current_item = response.get('Item')
previous_value = current_item['CurrentValue'] if current_item else 0.0
change_value = new_value - previous_value

# 事务写入：更新当前值 + 写入更新日志
table.transact_write_items(
    TransactItems=[
        # 更新或创建当前数值条目
        {
            'Put': {
                'TableName': 'MetricTrackingTable',
                'Item': {
                    'TrackingID': tracking_id,
                    'SK': 'CURRENT',
                    'CurrentValue': new_value,
                    'LastUpdatedAt': current_time,
                    'EntityType': 'METRIC'
                }
            }
        },
        # 写入更新日志条目
        {
            'Put': {
                'TableName': 'MetricTrackingTable',
                'Item': {
                    'TrackingID': tracking_id,
                    'SK': f'UPDATE#{current_time}',
                    'ChangeValue': change_value,
                    'UpdatedAt': current_time,
                    'PreviousValue': previous_value,
                    'NewValue': new_value,
                    'EntityType': 'METRIC_UPDATE'
                }
            }
        }
    ]
)

1. 获取当前数值 #

response = table.get_item(Key={'TrackingID': 'daily_sales', 'SK': 'CURRENT'})
current_value = response['Item']['CurrentValue'] if 'Item' in response else 0.0

2. 查询某个跟踪项的所有更新记录（按时间降序） #

response = table.query(
    KeyConditionExpression='TrackingID = :tid AND SK BEGINS_WITH :prefix',
    ExpressionAttributeValues={
        ':tid': 'daily_sales',
        ':prefix': 'UPDATE#'
    },
    ScanIndexForward=False  # 最新的更新排在最前面
)
update_logs = response['Items']

3. 查询所有变更值为+5的更新记录 #

response = table.query(
    IndexName='EntityType-ChangeValueTime-index',  # 你的GSI名称
    KeyConditionExpression='EntityType = :etype AND SK BEGINS_WITH :cv',
    ExpressionAttributeValues={
        ':etype': 'METRIC_UPDATE',
        ':cv': '+5#'
    }
)
matching_updates = response['Items']

4. 查询变更值在0到10之间的所有更新记录 #

response = table.query(
    IndexName='EntityType-ChangeValueTime-index',
    KeyConditionExpression='EntityType = :etype AND SK BETWEEN :min_cv AND :max_cv',
    ExpressionAttributeValues={
        ':etype': 'METRIC_UPDATE',
        ':min_cv': '0#',
        ':max_cv': '10#'
    }
)
range_updates = response['Items']

这个设计完全符合DynamoDB单表设计原则，解决了你之前遇到的GSI读写问题和查询结果混杂的问题，同时满足所有访问模式需求！

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