做数据分析和后端开发的人迟早会遇到这么一件事业务数据存在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_namemy_table) wr.dynamodb.put_df(dfdf, table_namemy_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_nameus-east-1) deserializer TypeDeserializer() def fetch_all_items(table_name): items [] response dynamodb.scan(TableNametable_name) items.extend(response.get(Items, [])) while LastEvaluatedKey in response: response dynamodb.scan( TableNametable_name, ExclusiveStartKeyresponse[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_namemy_table, as_dataframeTrue, )read_items方法返回的是干净的DataFrameDecimal自动转成float或int嵌套结构尽量展开成多层列名空值处理也比较合理。它的max_item_on_page参数控制每页读取条数影响内存占用和网络往返次数数据量大时可以调成1000减少请求次数。3.3 类型映射对照表实操中这张表要烂熟于心DynamoDB类型原始值示例boto3反序列化结果awswrangler转换后Pandas dtypeString (S){S: hello}hellohelloobject 或 stringNumber (N){N: 123}Decimal(123)123.0float64Binary (B){B: b...}bytesbytesobjectBoolean (BOOL){BOOL: true}TrueTrueboolString Set (SS){SS: [a, b]}[a, b][a, b]objectNumber Set (NS){NS: [1, 2]}[Decimal(1), Decimal(2)][1.0, 2.0]objectMap (M){M: {...}}dict多层列展开或dictobjectList (L){L: [...]}listlistobjectNull (NULL){NULL: True}NoneNoneNaN最常见的坑是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], axis1) 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_nameus-east-1) table dynamodb.Table(my_table) def chunks(lst, n): for i in range(0, len(lst), n): yield lst[i:in] 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_nameus-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(Itemcleaned)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 v4.3 传值类型转换DataFrame到DynamoDB的字段类型pandas的dtype和DynamoDB的类型不是一码事写入前必须确认或转换Pandas dtype / 值DynamoDB类型处理方式int64Number (N)直接传入boto3自动处理float64Number (N)直接传入object字符串String (S)直接传入boolBoolean (BOOL)直接传入注意Python的True/Falsedatetime64String (S)建议转成ISO格式字符串再写入bytesBinary (B)直接传入listList (L)直接传入dictMap (M)直接传入NaN无法存储必须转成None或字符串否则写入报错最容易被忽略的是datetime64类型DynamoDB原生没有时间日期类型写入前必须做一次转换否则boto3会报TypeError。我通常在最外层加一个统一的转换函数保证所有的pandas数据类型都能安全映射到DynamoDB支持的格式。4.4 超过Item大小限制怎么办DynamoDB单条Item最大是400KB。如果你的DataFrame某一行很大比如存了长文本或二进制数据写入会直接失败。batch_writer会把它归入未处理项反复重试直到程序卡死或超时。应对方案有三个把大字段拆分到另一张表主表只存索引和元数据。压缩字段内容比如用gzip压缩JSON字符串再存入。对大对象做异构存储比如把文件放S3DynamoDB里只存S3路径。这三条按优先级排能拆分就拆分不能拆分就压缩。硬塞进DynamoDB的Item里迟早会在读写性能和账单上还回来。5. 真实场景中的性能调优与避坑实录5.1 能用Query就别用Scanawswrangler和boto3都支持Scan全表但Scan是全表扫描读容量消耗与表大小成正比Query是基于分区键的精准查询速度是毫秒级费用更低。如果你的业务只需要某一天的数据而表的分区键包含日期务必用Query限定分区键import boto3 from boto3.dynamodb.conditions import Key dynamodb boto3.resource(dynamodb, region_nameus-east-1) table dynamodb.Table(my_table) response table.query( KeyConditionExpressionKey(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(TableNametable_name) yield response[Items] while LastEvaluatedKey in response: response dynamodb.scan( TableNametable_name, ExclusiveStartKeyresponse[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如果每条平均2KB1个WCU约等于每秒写0.5条。50万条要在300秒内写入至少需要约3333个WCU实际建议分多次写入降低峰值。这些数字不用背但你要清楚容量单位决定了你的总耗时上限先把数据规模算清楚再动手比盲目调并发参数靠谱得多。5.4 大分区键的倾斜问题DynamoDB数据分布是基于分区键的如果某个分区键的值特别集中比如所有数据都挂在date2025-01-01下面那么Scan和Query都会在这个分区上形成热点读性能上不去还可能被限流。这种情况下即使你用Query只查这一天也会很慢。我的经验是如果是查询热点的数据先把该分区键的数据分批按时间范围或另一个维度细分比如按小时、按ID范围拆分再并行查询。如果业务上无法避免热点考虑在分区键设计上增加随机后缀把数据分散到多个分区。6. 常见问题速查表与排查心得6.1 高频问题排查表现象可能原因解决思路导入pandas后报ModuleNotFoundError环境依赖缺失重新pip install pandas确认当前Python环境import awswrangler时报错botocore版本过低awswrangler与botocore版本不匹配pip install --upgrade boto3 botocore awswrangler读取结果全为NaN使用了TypeDeserializer但没有处理嵌套检查原始JSON结构先展开再取字段Decimal无法被astype(int)转换pandas版本对Decimal支持不一致先用astype(str)过渡转数值batch_writer一直重试不结束单条Item超过400KB或写容量不足拆分大对象检查表容量或On-Demand是否开启写入时TypeError: Unsupported typeDataFrame包含datetime或NaN统一转成字符串或NoneScan太慢表数据量大RCU不足用Query限定范围或按Segment并发读取ProvisionedThroughputExceededException读写超出表容量使用batch_writer自动重试或提高表容量Lambda中pandas包太大无法部署部署包超250MB限制用Lambda Layers或容器镜像方式部署本地连远程DynamoDB很慢网络延迟高临时把数据导入本地或使用DynamoDB Local调试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配合chunkedTrue流式处理数据而不是一次全读。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要快得多也便宜得多。这叫“用架构解决问题”不硬扛。