如何用Python获取DynamoDB表的所有分区键列表？

当然可以！在Python里用boto3库就能高效实现这个需求，完全不用扫描整个表的所有属性。这里给你两种实用的方案：

方法一：带投影表达式的Scan（推荐） #

DynamoDB的Scan操作支持通过ProjectionExpression指定只返回你需要的字段——也就是分区键。这种方式能大幅减少数据传输量，还能降低读写容量的消耗，非常适合只需要分区键的场景。

完整代码示例 #

import boto3

def fetch_all_partition_keys(table_name, partition_key_col):
    # 初始化DynamoDB资源
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table(table_name)
    
    all_partition_keys = []
    last_evaluated_key = None

    # 循环处理分页结果（DynamoDB单次Scan最多返回1MB数据）
    while True:
        scan_params = {
            'ProjectionExpression': partition_key_col,
        }
        # 如果有分页标记，继续扫描下一页
        if last_evaluated_key:
            scan_params['ExclusiveStartKey'] = last_evaluated_key
        
        response = table.scan(**scan_params)
        # 提取当前页的分区键值并加入列表
        all_partition_keys.extend(item[partition_key_col] for item in response['Items'])
        
        # 更新分页标记，没有则退出循环
        last_evaluated_key = response.get('LastEvaluatedKey')
        if not last_evaluated_key:
            break

    # 如果你需要去重（仅适用于复合主键场景，简单主键的分区键本身唯一）
    # all_partition_keys = list(set(all_partition_keys))
    return all_partition_keys

# 调用示例
if __name__ == "__main__":
    target_table = "YourTableName"
    pk_column = "YourPartitionKeyName"
    result = fetch_all_partition_keys(target_table, pk_column)
    print(f"表中所有分区键：{result}")

关键说明 #

• ProjectionExpression：明确指定只返回分区键字段，避免冗余数据传输。
• 分页处理：DynamoDB单次Scan最多返回1MB数据，所以需要循环处理LastEvaluatedKey直到获取全部数据。
• 权限要求：确保你的IAM身份拥有dynamodb:Scan操作权限。

方法二：进阶优化——并行扫描（针对超大表） #

如果你的表数据量极大，单线程扫描速度太慢，可以用并行扫描提升效率。通过TotalSegments和Segment参数将表分成多个段，用多线程同时扫描不同的段：

简化示例（多线程版本） #

import boto3
from concurrent.futures import ThreadPoolExecutor

def scan_segment(segment, total_segments, table_name, pk_col):
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table(table_name)
    segment_keys = []
    last_key = None
    while True:
        params = {
            'ProjectionExpression': pk_col,
            'TotalSegments': total_segments,
            'Segment': segment
        }
        if last_key:
            params['ExclusiveStartKey'] = last_key
        response = table.scan(**params)
        segment_keys.extend(item[pk_col] for item in response['Items'])
        last_key = response.get('LastEvaluatedKey')
        if not last_key:
            break
    return segment_keys

def fetch_all_partition_keys_parallel(table_name, pk_col, total_segments=4):
    all_keys = []
    with ThreadPoolExecutor(max_workers=total_segments) as executor:
        futures = [executor.submit(scan_segment, seg, total_segments, table_name, pk_col) for seg in range(total_segments)]
        for future in futures:
            all_keys.extend(future.result())
    # 按需去重
    # all_keys = list(set(all_keys))
    return all_keys

# 调用示例
# result = fetch_all_partition_keys_parallel("YourLargeTableName", "YourPK", total_segments=8)

注意事项 #

• TotalSegments建议设置为和CPU核心数匹配，避免过度并发导致资源浪费。
• 并行扫描会消耗更多的RCU，需要根据表的容量模式调整。

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