Pandas与DynamoDB无缝对接：类型转换、批量写入与性能优化全攻略

做数据分析和后端开发的人，迟早会遇到这么一件事：业务数据存在DynamoDB里，但分析脚本是Python写的，数据到了手里得变成DataFrame才能干活。Pandas和DynamoDB的对接，听起来是个“调API取数据”的小事，真做起来就会碰到类型转换、批量写入限制、扫描性能、内存溢出这一堆问题。

这篇文章我会把完整的对接方案、我踩过的坑、还有经过验证的性能优化手段都整理出来。不管你是刚把DynamoDB的数据导进Pandas做报表，还是打算把清洗好的DataFrame批量写回DynamoDB，都能在这里找到可以直接抄作业的答案。

1. 为什么这个组合让人又爱又恨

1.1 DynamoDB的“反Pandas”设计哲学

DynamoDB是AWS提供的NoSQL数据库，底层基于分区存储，数据天然是“键值对”形态。它的设计目标是高并发、低延迟、水平扩展，一切围绕“按照主键快速查询”来优化。但Pandas是完全不同的逻辑，它是内存中的二维表格结构，擅长的是列式计算、聚合、透视、绘图。两者的思维方式几乎是相反的。

让这个组合变得麻烦的根源在于数据形态的差异。DynamoDB返回的数据结构是List[Dict]，每个Dict是一条记录的属性集合，属性值还带着类型标记，比如{"S": "hello"}、{"N": "123"}。Pandas需要的是一张规整的表，列名统一、类型明确、缺失值可以处理。把前者变成后者，不是简单的pd.DataFrame(data)就能搞定的，你至少要处理类型标记的剥离、嵌套结构的展开、数值字符串的转换这几件事。

听起来不难，但实际项目里DynamoDB表的设计往往很随意，同一个字段在不同记录里可能是字符串、可能是数字、甚至可能是空的，这种脏数据在Pandas里会直接变成object类型或NaN，后续聚合计算全乱套。

1.2 无缝对接到底意味着什么

我理解的“无缝”，不是写一次性的脚本把数据捞出来，而是建立一套可复用的读写链路：给定一个DynamoDB表名，能在几分钟内把全表或指定分区数据变成干净的DataFrame；给定一个DataFrame，能稳定高效地把数据写回DynamoDB，不丢数据、不超限、可控成本。

这套链路的实现难度不在API调用本身，而在边界情况：数据超过16MB的Scan限制怎么办、单个Item超过400KB怎么办、写入频率超过WCU限流怎么办、DataFrame里含NaN和复杂嵌套怎么办。把这些都处理好，才算真正的无缝。

2. 动手前必读：环境准备与依赖选型

2.1 核心依赖：boto3、pandas、awswrangler

所有对接工作都从boto3开始，这是AWS官方的Python SDK，负责连接DynamoDB。pandas用来承载和分析数据。如果只装这两个，你可以完成90%的工作，但代码会写得比较啰嗦，封装扫描和类型转换的逻辑至少在100行以上。所以我建议再加上awswrangler，这个库在AWS实验室基础上发展而来，现在由AWS官方维护，对Pandas和DynamoDB的对接做了深度封装。

pip install pandas boto3 awswrangler

如果你在国内网络环境，用清华源安装更稳：

pip install pandas boto3 awswrangler -i https://pypi.tuna.tsinghua.edu.cn/simple

版本方面，pandas的2.x系列在处理字符串和缺失值上明显比1.x更合理，awswrangler的3.x版本对DynamoDB读写接口做了一次重构，documents和items的概念更清晰。建议pandas用2.0以上，awswrangler用3.0以上，boto3保持最新就行。

2.2 凭证配置与本地调试的坑

连接DynamoDB需要AWS凭证。如果你是在EC2或Lambda上运行，直接用IAM角色；本地调试则用aws configure配置AK/SK。建议把凭证放到环境变量里，别写死在代码中，尤其当你用git管理代码时，凭证泄露的风险不可忽视。

本地调试还有一个容易忽略的问题：默认region一定要配，否则boto3会报RegionError。而且本地连的DynamoDB如果是云端实例，网络延迟会让你的调试体验非常糟糕，Scan几万条数据可能要等几十秒。我一般会把部分数据导到本地，用DynamoDB Local做功能验证，逻辑跑通后再连正式环境。

2.3 为什么我最终选择了awswrangler

如果你只用boto3，读数据要自己写分页、类型转换、NaN处理，写数据要自己处理批量写入的切分与重试。这些逻辑不难，但分散注意力。awswrangler把这些封装成了两行代码：

import awswrangler as wr df = wr.dynamodb.read_items(table_name="my_table") wr.dynamodb.put_df(df=df, table_name="my_table")

它底层还是boto3，但把涨经验的部分都替你做了。我的建议是：先会用boto3，理解原理，然后日常开发直接用awswrangler提效。这和我用pandas但不排斥SQL是一个道理——工具帮你省时间，但底层原理帮你排坑。

3. 从DynamoDB读数据到Pandas：核心细节拆解

3.1 用boto3手动实现读取与类型转换

先看最原始的方案，理解这个过程到底发生了什么。

Step 1：扫描表，拿到原始数据。

import boto3 from boto3.dynamodb.types import TypeDeserializer dynamodb = boto3.client("dynamodb", region_name="us-east-1") deserializer = TypeDeserializer() def fetch_all_items(table_name): items = [] response = dynamodb.scan(TableName=table_name) items.extend(response.get("Items", [])) while "LastEvaluatedKey" in response: response = dynamodb.scan( TableName=table_name, ExclusiveStartKey=response["LastEvaluatedKey"] ) items.extend(response.get("Items", [])) return items raw_items = fetch_all_items("my_table") deserialized = [{k: deserializer.deserialize(v) for k, v in item.items()} for item in raw_items] df = pd.DataFrame(deserialized)

Step 2：理解类型反序列化。TypeDeserializer的作用是把{"S": "hello"}变成"hello"，把{"N": "123"}变成Decimal("123")。注意这里有个巨坑：DynamoDB的数字类型反序列化后是Decimal，不是int或float。如果你直接丢给pandas做运算，某些情况下它会报错或行为诡异。

Step 3：将Decimal列转为pandas的数值类型。我在项目里一般这样做：

import pandas as pd from decimal import Decimal def decimal_to_number(series): if series.map(lambda x: isinstance(x, Decimal)).any(): return pd.to_numeric(series.astype(str)) return series

先转成字符串，再用pd.to_numeric，顺手把缺失值变成NaN，这个顺序能避免很多int/float精度问题。

3.2 用awswrangler一行读取

如果你赶时间，用awswrangler能少写很多代码：

import awswrangler as wr df = wr.dynamodb.read_items( table_name="my_table", as_dataframe=True, )

read_items方法返回的是干净的DataFrame：Decimal自动转成float或int，嵌套结构尽量展开成多层列名，空值处理也比较合理。它的max_item_on_page参数控制每页读取条数，影响内存占用和网络往返次数，数据量大时可以调成1000，减少请求次数。

3.3 类型映射对照表

实操中这张表要烂熟于心：

最常见的坑是Number类型。如果你直接用boto3读取，得到的Decimal列如果不转换，后续的df["price"].sum()是能算的，但df["price"].astype(int)在某些pandas版本会出问题。awswrangler默认把Number转换成float，如果你需要保留整数精度，比如订单号、ID号，建议手动处理。

3.4 嵌套数据的展开策略

DynamoDB的Map和List类型对应JSON嵌套结构，直接放进DataFrame里会变成object列，每个单元格是一个dict或list，分析起来很痛苦。我有两种处理方式。

方式一：让awswrangler自动展开Map。它的read_items会把Map类型展开成多级列索引，比如属性address包含city和zip，生成的是("address", "city")和("address", "zip")两列。这种方式适合做透视和筛选，但如果嵌套特别深，列索引会变得难懂。

方式二：手动展开固定字段，最实用的方式：

def normalize_nested(df, column): extracted = pd.json_normalize(df[column].dropna()) extracted.columns = [f"{column}.{c}" for c in extracted.columns] df = pd.concat([df.drop(columns=[column]), extracted], axis=1) return df

遇到某些记录里address字段缺失，dropna()会自动忽略，生成的子列对应位置是NaN，不会错位。这个方法适合嵌套层级固定、字段名一致的场景，比通用展开逻辑更可控。

4. 从Pandas写数据到DynamoDB：批量写入的完整方案

4.1 单条写入不可取，批量写入有肉限

一次put_item只能写一条数据，如果DataFrame有10万行，循环写入会产生10万次HTTP请求，耗时和费用都不可控。所以必须用batch_write_item，它允许在一次请求中最多写入25条（或删除25条），小于等于16MB的数据容量。

aws官方推荐的切分逻辑是按25条一组切分，每组写一次。如果你用boto3手动处理：

import boto3 from boto3.dynamodb.conditions import Attr dynamodb = boto3.resource("dynamodb", region_name="us-east-1") table = dynamodb.Table("my_table") def chunks(lst, n): for i in range(0, len(lst), n): yield lst[i:i+n] def put_dataframe(df, table_name): table = dynamodb.Table(table_name) records = df.to_dict("records") for batch in chunks(records, 25): unprocessed = batch while unprocessed: response = table.batch_writer() # 实际中应使用batch_writer上下文管理器，见下文 break

写到这里我意识到直接手动处理重试逻辑很蠢，因为boto3提供了batch_writer，它是专门干这个的上下文管理器，自动处理批次切分、未写入项重试、以及背压控制。

4.2 最佳实践：用batch_writer实现稳定写入

代码如下：

import boto3 def put_df_to_dynamodb(df, table_name): dynamodb = boto3.resource("dynamodb", region_name="us-east-1") table = dynamodb.Table(table_name) records = df.to_dict("records") with table.batch_writer() as batch: for record in records: # 处理NaN值，DynamoDB不支持NaN直接存储 cleaned = {k: (None if pd.isna(v) else v) for k, v in record.items()} batch.put_item(Item=cleaned)

batch_writer会自动把数据按25条打包，发送失败会自动重试，还会基于DynamoDB的响应调整写入速度。这是我在生产环境中最稳的方案。

需要注意一点：pd.isna(v)不能直接处理列表、字典这类复杂对象，会报错。如果你的DataFrame里有嵌套结构，要对这类值单独判断：

def clean_value(v): if isinstance(v, (dict, list)): return v if pd.isna(v): return None return v

4.3 传值类型转换：DataFrame到DynamoDB的字段类型

pandas的dtype和DynamoDB的类型不是一码事，写入前必须确认或转换：

最容易被忽略的是datetime64类型，DynamoDB原生没有时间日期类型，写入前必须做一次转换，否则boto3会报TypeError。我通常在最外层加一个统一的转换函数，保证所有的pandas数据类型都能安全映射到DynamoDB支持的格式。

4.4 超过Item大小限制怎么办

DynamoDB单条Item最大是400KB。如果你的DataFrame某一行很大，比如存了长文本或二进制数据，写入会直接失败。batch_writer会把它归入未处理项反复重试，直到程序卡死或超时。

应对方案有三个：

• 把大字段拆分到另一张表，主表只存索引和元数据。
• 压缩字段内容，比如用gzip压缩JSON字符串再存入。
• 对大对象做异构存储，比如把文件放S3，DynamoDB里只存S3路径。

这三条按优先级排，能拆分就拆分，不能拆分就压缩。硬塞进DynamoDB的Item里，迟早会在读写性能和账单上还回来。

5. 真实场景中的性能调优与避坑实录

5.1 能用Query就别用Scan

awswrangler和boto3都支持Scan全表，但Scan是全表扫描，读容量消耗与表大小成正比；Query是基于分区键的精准查询，速度是毫秒级，费用更低。如果你的业务只需要某一天的数据，而表的分区键包含日期，务必用Query限定分区键：

import boto3 from boto3.dynamodb.conditions import Key dynamodb = boto3.resource("dynamodb", region_name="us-east-1") table = dynamodb.Table("my_table") response = table.query( KeyConditionExpression=Key("date").eq("2025-01-01") )

用awswrangler可以用where参数配合分区键，效果类似。我在接手一个数据量上亿的表格时，把Scan改成按天Query，读取时间从25分钟缩到3分钟，费用降了一个数量级。

5.2 Scan的内存管理与分页策略

虽然不推荐Scan，但有时候确实要全量导出，比如做数据仓库同步。全量Scan一个几GB的表，直接把结果全部放进列表，内存会爆。

推荐的模式是流式处理：每页数据转换完就追加到DataFrame，处理完立刻释放raw_items。awswrangler内部已经做了分页，但你要注意控制并发和每页大小。

如果你能接受一定的时间换内存，可以用生成器模式逐页读取：

def scan_table_paginated(table_name): response = dynamodb.scan(TableName=table_name) yield response["Items"] while "LastEvaluatedKey" in response: response = dynamodb.scan( TableName=table_name, ExclusiveStartKey=response["LastEvaluatedKey"] ) yield response["Items"]

然后每拿一批就pd.concat一次，用完一批就丢弃原始数据，控制内存峰值。

5.3 并发参数与RCU/WCU的权衡

DynamoDB的读和写受限于表的读写容量单位（RCU/WCU）。如果把读并发开得太大，会被系统限流，返回ProvisionedThroughputExceededException；开得太小，全量导出要跑很久。

对于按量计费（On-Demand）的表，没有容量限制的烦恼，但费用会飙升，所以还是要控制并发。对于预留容量的表，如果设置了auto scaling，它会试探性提高并发，但提高是缓慢的，你要估算好初始并发。

我的经验公式是：

• 一个RCU每秒读一条4KB的Item，如果每条Item平均2KB，那么1个RCU约等于每秒读2条。全表50万条，要300秒内读完，至少需要约834个RCU，按每个分区最多并行读3个Segment来算，并发线程数控制在15到30之间比较安全。
• 写入同理，一个WCU每秒写一条1KB的Item，如果每条平均2KB，1个WCU约等于每秒写0.5条。50万条要在300秒内写入，至少需要约3333个WCU，实际建议分多次写入降低峰值。

这些数字不用背，但你要清楚容量单位决定了你的总耗时上限，先把数据规模算清楚再动手，比盲目调并发参数靠谱得多。

5.4 大分区键的倾斜问题

DynamoDB数据分布是基于分区键的，如果某个分区键的值特别集中，比如所有数据都挂在date=2025-01-01下面，那么Scan和Query都会在这个分区上形成热点，读性能上不去，还可能被限流。

这种情况下，即使你用Query只查这一天，也会很慢。我的经验是：如果是查询热点的数据，先把该分区键的数据分批按时间范围或另一个维度细分，比如按小时、按ID范围拆分，再并行查询。如果业务上无法避免热点，考虑在分区键设计上增加随机后缀，把数据分散到多个分区。

6. 常见问题速查表与排查心得

6.1 高频问题排查表

6.2 那些坑了我很久的小问题

第一个坑：pandas的to_dict("records")会把NaN保留为float('nan')，而boto3在写入时会报“无法序列化NaN”的错误，因为DynamoDB不支持NaN。我第一次跑批量入库时在这个问题上卡了一个多小时，最后才知道要先统一转成None。

第二个坑：DynamoDB的Number类型精度问题。pandas的float64在某些边界值上会损失精度，尤其当你处理商品价格、交易金额时，差一分钱都会出大问题。建议在读取时对金额字段保留Decimal，或者在pandas中用float64但写入前用Decimal(str(v))手动转换。

第三个坑：batch_writer在Lambda这种短时运行环境中要注意超时。如果函数超时设置是15分钟，写入50万条数据基本不够，这种情况下应该把写任务拆成多个Lambda并发执行，或者改用Step Functions分批调度。

第四个坑：awswrangler的read_items在读大表时默认会把所有数据加载到内存，如果表有上亿条记录，16GB内存的机器也扛不住。解决办法是设置max_items_per_query配合chunked=True，流式处理数据而不是一次全读。

6.3 一个完整的端到端案例

最后分享一个我最近做的数据同步任务。场景是：DynamoDB里有一张用户行为日志表，每天新增200万条记录，需要每天同步到本地用于分析。

我的方案是：

• 分区键是date，排序键是user_id，所以每天的数据都集中在一个分区下。
• 先算出当天数据量，用DescribeTable拿到ItemCount估算RCU需求。
• 用boto3的Query加分页，每次取10000条，转成DataFrame后直接to_parquet落地，这样内存里始终只有1万条。
• 清洗完的数据会用batch_writer写入另一张结构化表，供下游SQL查询。

整个过程用AWS Glue Job跑，依赖打在产品包里，运行时间从原来的45分钟压到12分钟，主要靠的是Batch写入替代逐条写入，加合理设置分页大小。入库之后的表，用Athena可以直接查，但那就是另一个故事了。

7. 写在最后：我的习惯与建议

用了很长一段时间Pandas和DynamoDB的对接，我个人的体会是：不要总想着用一个库解决所有问题。boto3是地基，awswrangler是脚手架，Pandas是工作台，各司其职。理解boto3的Scan和Query、理解DynamoDB的类型系统、理解容量单位的含义，比记住某个库的API更重要。

另外一个小建议：任何写入DynamoDB的任务，都要先在小样本上跑通，再上全量。我习惯先把DataFrame截取前100条试写，确认类型和容量都没问题，再放开全量执行。这个习惯帮我避开了无数次写半截才发现字段类型不对的尴尬。

如果你的业务还在快速发展，表结构可能经常变，建议在Pandas和DynamoDB之间加一层Schema校验：读数据时检查必填字段是否存在，写数据时限制只写入允许的字段。这层校验看起来多余，但能防止上游数据格式变化把下游清洗任务悄悄带崩。

最后再分享一个冷门但实用的技巧：处理超大数据量时，把DynamoDB数据先导到S3（用S3 Parquet格式），再用Athena跑SQL，最后用Pandas读结果，比直接全表Scan再转DataFrame要快得多，也便宜得多。这叫“用架构解决问题”，不硬扛。