如何优化Python解析DynamoDB PartiQL查询结果至DataFrame的过程

先修正现有代码的小问题 #

你的代码里有两处笔误/错误：

• 拼写错误：import panads as pd 应该改为 import pandas as pd
• 类型注解错误：parse_records 函数的返回值应该是 List[Dict] 而非 Dict
• 缺少类型导入：需要添加 from typing import List, Dict

方案一：优化现有Client解析逻辑（减少冗余代码，提升效率） #

虽然无法完全避免遍历记录（因为DynamoDB Client返回的是带类型标记的原始结构，必须逐个字段反序列化），但可以通过更简洁高效的写法优化：

1. 简化解析函数：用列表推导式替代显式循环，代码更紧凑且执行效率更高
2. 用生成器处理分页：避免一次性加载所有记录到内存，尤其适合大数据量场景

import boto3
from boto3.dynamodb.types import TypeDeserializer
import pandas as pd
from typing import List, Dict, Generator

connection = boto3.client('dynamodb', **config)

def parse_records(records: List[Dict]) -> List[Dict]:
    deserializer = TypeDeserializer()
    # 用列表推导式简化解析逻辑
    return [{k: deserializer.deserialize(v) for k, v in record.items()} for record in records]

def fetch_all_records(query: str) -> Generator[Dict, None, None]:
    """生成器：逐页获取并解析所有查询结果"""
    result = connection.execute_statement(Statement=query)
    while True:
        yield from parse_records(result['Items'])
        if 'NextToken' not in result:
            break
        result = connection.execute_statement(Statement=query, NextToken=result['NextToken'])

# 生成DataFrame（用from_records替代json_normalize，扁平结构更高效）
df = pd.DataFrame(fetch_all_records(query))
# 如果有嵌套结构，再用json_normalize：df = pd.json_normalize(fetch_all_records(query))

方案二：改用Boto3 Resource（自动处理类型反序列化，彻底省去手动解析） #

Boto3的resource对象会自动将DynamoDB的类型标记结构转换为Python原生类型，不需要手动写解析函数，相当于把遍历解析的工作交给了boto3内部优化的逻辑：

import boto3
import pandas as pd
from typing import Generator

# 使用Resource而非Client
dynamodb = boto3.resource('dynamodb', **config)

def fetch_all_records(query: str) -> Generator[Dict, None, None]:
    """生成器：逐页获取已反序列化的查询结果"""
    response = dynamodb.execute_statement(Statement=query)
    while True:
        yield from response['Items']  # Resource返回的Items已经是原生Python类型
        if 'NextToken' not in response:
            break
        response = dynamodb.execute_statement(Statement=query, NextToken=response['NextToken'])

# 直接生成DataFrame
df = pd.DataFrame(fetch_all_records(query))
# 嵌套结构用json_normalize：df = pd.json_normalize(fetch_all_records(query))

这个方案是最优解：既省去了手动解析的代码，又利用了boto3内部优化的反序列化逻辑，效率比自己写的遍历更高。

关键说明 #

• 为什么无法完全避免遍历：DynamoDB返回的每条记录都包含类型标记（如{'S': 'string'}、{'N': '123'}），必须逐个字段转换为原生类型，这个过程本质上需要遍历处理。但用Resource的话，这个遍历是boto3内部优化过的，比自定义实现更高效。
• 分页处理：无论用Client还是Resource，都需要处理NextToken来获取所有结果，生成器是最内存友好的方式。

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