MySQL表转DynamoDB：单表关联维护、建表及查询方法咨询

一、单表结构设计 #

DynamoDB单表设计的核心是用**分区键（PK）和排序键（SK）**把关联数据聚合到同一个分区，实现高效查询。针对你的场景，推荐如下表结构：

• 分区键（Partition Key）：pk（字符串类型），格式为RUN#<run_id>，用来标识同一个Run的所有关联记录
• 排序键（Sort Key）：sk（字符串类型），用不同前缀区分主记录和详情记录：
  • 主记录（对应原run_master）：sk设为MASTER
  • 详情记录（对应原run_details）：sk设为DETAIL#<detail_id>（detail_id可以是自增序号或UUID）

额外添加item_type字段（字符串类型），用来快速区分主/详情记录，方便后续过滤。

数据示例 #

主记录：

{
  "pk": "RUN#1",
  "sk": "MASTER",
  "user_id": 1,
  "add_date": "2022-09-23 11:04:06",
  "item_type": "run_master"
}

详情记录：

{
  "pk": "RUN#1",
  "sk": "DETAIL#1",
  "track_name": "track1",
  "status": "fail",
  "item_type": "run_detail"
}

二、数据插入逻辑 #

你的脚本流程可以基本保留，只需把插入两张表改成插入同一张表：

1. 生成run_id（可用自增ID、UUID或业务唯一标识）
2. 插入Run主记录：指定pk为RUN#<run_id>，sk为MASTER，带上user_id、add_date等字段
3. 循环插入详情记录：每条记录的pk与主记录一致（RUN#<run_id>），sk用DETAIL#<序号/UUID>，带上track_name、status等字段

用Python boto3的示例代码：

import boto3

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

# 生成run_id（示例用固定值，实际可替换为UUID或自增逻辑）
run_id = "1"

# 插入run_master记录
table.put_item(
    Item={
        'pk': f'RUN#{run_id}',
        'sk': 'MASTER',
        'user_id': 1,
        'add_date': '2022-09-23 11:04:06',
        'item_type': 'run_master'
    }
)

# 批量插入run_details记录
detail_list = [
    {"track_name": "track1", "status": "fail"},
    {"track_name": "track2", "status": "pass"},
    {"track_name": "track3", "status": "fail"}
]

for idx, detail in enumerate(detail_list, start=1):
    table.put_item(
        Item={
            'pk': f'RUN#{run_id}',
            'sk': f'DETAIL#{idx}',
            'track_name': detail['track_name'],
            'status': detail['status'],
            'item_type': 'run_detail'
        }
    )

三、通过Run Id查询所有详情记录 #

利用DynamoDB的Query操作（比Scan高效得多），根据pk定位到目标Run的分区，再通过begins_with过滤出所有详情记录：

def fetch_run_details(run_id):
    response = table.query(
        KeyConditionExpression='pk = :pk_val AND begins_with(sk, :sk_prefix)',
        ExpressionAttributeValues={
            ':pk_val': f'RUN#{run_id}',
            ':sk_prefix': 'DETAIL#'
        }
    )
    return response['Items']

# 查询run_id=1的所有详情
run_details = fetch_run_details("1")
for detail in run_details:
    print(f"Track: {detail['track_name']}, Status: {detail['status']}")

如果需要同时获取Run主记录和详情，可以先查询该run_id下的所有记录，再在客户端过滤：

def fetch_run_with_details(run_id):
    response = table.query(
        KeyConditionExpression='pk = :pk_val',
        ExpressionAttributeValues={':pk_val': f'RUN#{run_id}'}
    )
    items = response['Items']
    # 分离主记录和详情
    run_master = next(item for item in items if item['item_type'] == 'run_master')
    run_details = [item for item in items if item['item_type'] == 'run_detail']
    return {"master": run_master, "details": run_details}

额外优化提示 #

• 如果需要按user_id查询该用户的所有Run记录，可以创建全局二级索引（GSI），设置GSI的分区键为user_id，排序键为add_date，这样能高效按用户+时间排序查询。
• 推荐用UUID生成run_id，避免自增ID导致的热点写入问题（短时间内大量写入同一个分区）。
• 如果track_name在同一个Run内唯一，可以把详情记录的sk设为DETAIL#<track_name>，省去额外的detail_id。

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