DynamoDB与Pandas高效对接：类型转换、批量写入与工程化实践

最近两个项目同时在做一个是实时数据管道一个是离线分析任务两边数据源一碰发现交接环节全卡在同一个地方DynamoDB里的数据要进Pandas做清洗分析写好的分析结果又要落回DynamoDB供业务读取。这个需求听起来简单但真做起来坑比想象中多。很多人一上来就是boto3.scan()一把梭拿到返回结果直接pd.DataFrame(items)跑是能跑但一旦数据量上去、类型复杂点要么内存炸了要么Decimal报错要么嵌套字段裂成一堆奇奇怪怪的列。这篇文章把我实际对接过程中的完整链路、类型映射方案、批量写入优化、以及踩过的几个比较隐蔽的坑全部梳理出来给需要的同学一份能直接抄作业的参考。文章从我理解的两边数据模型差异讲起然后分别拆解读取方向和写入方向的实现细节最后分享一些工程化封装和并发控制上的经验。全部代码基于Python 3.8和boto3如果你用的版本比较老个别API参数名可能略有出入但核心思路完全通用。1. 先别急着写代码搞清DynamoDB和Pandas的数据模型差异对接之前我一直强调先把两边的数据模型想清楚因为绕过这个环节直接写代码后面八成要返工。DynamoDB是NoSQL存的是Item每个Item是一堆键值对没有强制表结构Pandas是二维表结构有明确的列和行每一列有统一的数据类型。这两者之间的映射关系处理不好后面全是雷。1.1 DynamoDB的Item本质一个带类型标签的JSONDynamoDB的底层存储格式是AttributeValue。每个Item本质上是一个字典但字典里的value不是普通的Python对象而是带类型标记的结构。比如一个字符串hello在DynamoDB里是{S: hello}一个数字42是{N: 42}一个布尔值是{BOOL: true}。之所以要这样设计是因为DynamoDB需要知道每个属性是什么类型才能在查询和索引时做出正确的行为。这里有一个非常容易让人困惑的点数字类型在存储层是字符串{N: 42}里的42是字符串形式的数字。这意味着如果你直接把这个字典丢给Pandas所有数字列都会变成object类型后面df[count].astype(int)这种操作就会疯狂报错。再看嵌套结构。DynamoDB支持MMap即嵌套JSON对象和LList即数组这意味着一个Item可以是一个很深的树形结构。而Pandas的DataFrame做嵌套并不自然虽然可以在一个cell里塞一个字典但做分析、聚合、筛选时非常别扭。这个差异决定了读取方向的核心工作把树状结构展平成表格或者至少展平成Pandas能高效处理的形态。DynamoDB支持的类型完整清单如下AttributeValue类型缩写说明对应的Python类型StringS字符串strNumberN数字存储为字符串DecimalBinaryB二进制数据bytesBooleanBOOL布尔值boolNullNULL空值NoneString SetSS字符串集合setNumber SetNS数字集合setBinary SetBS二进制集合setListL数组listMapM嵌套对象dict1.2 Pandas的视角强类型、二维、缺失值哲学Pandas这边DataFrame是一个强类型的二维表每一列有统一的dtype比如int64、float64、object、datetime64。数据缺失时用NaN浮点或NaT时间表示。Pandas的设计哲学是整齐的数据:每一列都是一个Series操作时按列对齐。这个设计在处理DynamoDB数据时会产生几个直接冲突第一DynamoDB的Item可以没有某个字段不同的Item字段不同但DataFrame列必须有值缺失的就是NaN。这两者的对应关系需要一个明确的规则。第二Pandas的object类型列虽然可以装任何东西但一旦你要做性能敏感的操作比如groupby、排序、mergeobject列的速度会被拖累。第三Pandas的整数列不支持缺失值int64里没有NaN如果你从DynamoDB读到一个字段有的Item里有、有的没有那这个列要么变float64NaN用浮点表示要么用可空整数类型Int64Pandas的扩展类型。这些细节决定了拿到数据之后的第一道处理工序。1.3 核心映射决策读取时转换还是读取后转换既然两边差异这么大一个基础问题是在哪个环节做类型转换我见过不少人在scan之后手动写一个循环对每个Item的每个字段做判断把Decimal转float、把布尔处理一下然后才丢给DataFrame。这种做法在小数据量下没问题可数据量一上来这个循环就是性能瓶颈而且代码巨丑。我的做法是写一个统一的递归转换函数在scan返回的原始Item上做一次全量转换把AttributeValue风格的数据转成原生Python类型然后再交给Pandas。这样做好处是转换逻辑集中在一个地方后续无论数据长什么样都能复用调试起来也方便。类型转换映射表如下DynamoDB存储类型转换目标备注Sstr原样NDecimal → float默认转float精度要求高可转intBOOLbool原样NULLNone原样SS / NS / BSset → list转成list方便Pandas处理Llist递归处理每个元素Mdict递归处理每个value这个映射表是整个对接方案的基石读取和写入两侧的代码都是围绕它展开的。2. 读取方向把DynamoDB的Item训练成规规矩矩的DataFrame从DynamoDB读数据常用的方式有两种Scan全表扫描和Query按分区键查询。在实际项目中分析任务很多时候确实需要全表所以Scan的使用率反而更高。2.1 为什么我更推荐用boto3的client而不是resourceboto3给DynamoDB提供了两种访问方式client和resource。resource是高级封装用起来像ORM比如table.scan()直接返回Item列表client是低级API更贴近HTTP接口原始语义比如dynamodb.scan()返回的是完整的{Items: [...]}响应体。对接Pandas时我更推荐用client。原因很简单client返回的数据结构是确定的你能拿到Count、ScannedCount、LastEvaluatedKey这些元信息resource虽然也返回但封装得比较柔有时候会让你忽略分页问题。而且client的API参数比如ProjectionExpression、FilterExpression、Limit和DynamoDB的官方文档一一对应排查问题时有文档可查。真正让我坚定选client的原因是可以配合练习HTTP语义的分页逻辑这个后面会细说。代码示例导航到DynamoDB控制台或直接使用已有表import boto3 import pandas as pd from boto3.dynamodb.types import TypeDeserializer dynamodb boto3.client(dynamodb, region_nameus-east-1) resp dynamodb.scan(TableNamemy_table) print(resp[Items][0]) # 输出类似下面的结构注意类型标签 # {user_id: {S: a1b2c3}, age: {N: 28}, tags: {SS: [vip, new]}}2.2 分页读取是一个默认必须处理的逻辑Scan一次最多返回1MB数据或者受到Limit参数限制所以无论你的表多大都不能指望一次拿到全部数据。正确姿势是用LastEvaluatedKey做循环直到返回结果里没有这个字段为止。这个逻辑写起来不复杂但有几个细节要留意如果有并行扫描需求可以用Segment和TotalSegments做并行分段扫描如果只要部分字段尽量用ProjectionExpression只取需要的字段减小网络传输量。分页扫描的标准写法def scan_table_to_df(table_name, projection_expressionNone): deserializer TypeDeserializer() items [] kwargs { TableName: table_name, Limit: 1000, } if projection_expression: kwargs[ProjectionExpression] projection_expression while True: resp dynamodb.scan(**kwargs) for raw_item in resp[Items]: # TypeDeserializer 将 AttributeValue 转为原生 Python 类型 items.append(deserializer.deserialize({M: raw_item})) if LastEvaluatedKey in resp: kwargs[ExclusiveStartKey] resp[LastEvaluatedKey] else: break df pd.DataFrame(items) return df这里使用了boto3.dynamodb.types.TypeDeserializer这个工具类能替你把带类型标签的AttributeValue递归转成Python原生类型。用完之后上面示例里的{user_id: {S: a1b2c3}, age: {N: 28}}会变成{user_id: a1b2c3, age: Decimal(28)}。不过TypeDeserializer有一个让你不省心的地方所有N类型都转成了Decimal。Decimal本身没问题但当你把它塞进Pandas的DataFrame时列类型会变成object而不是float或int。所以下一步得把Decimal统一转掉。2.3 递归转换把Decimal、Set、bytes都收拾干净最省心的办法是自己写一个递归转换函数因为只针对一到两层嵌套处理的话以后遇到嵌套更深的场景就得重写。这个函数可以直接消费TypeDeserializer的输出也可以直接处理原始AttributeValue看你的使用习惯。我给的方案是自己处理原始Item一步到位避免中间态def attr_value_to_python(item): 把AttributeValue风格的dict转成Pandas友好的Python对象。 converted {} for key, value in item.items(): if S in value: converted[key] value[S] elif N in value: # 数字保留精度但这里统一转成float/int大整数建议用int converted[key] float(value[N]) if . in value[N] else int(value[N]) elif BOOL in value: converted[key] value[BOOL] elif NULL in value: converted[key] None elif SS in value: converted[key] list(value[SS]) elif NS in value: converted[key] [float(v) if . in v else int(v) for v in value[NS]] elif BS in value: converted[key] list(value[BS]) elif L in value: converted[key] [attr_value_to_python({item: ele})[item] for ele in value[L]] elif M in value: converted[key] {k: attr_value_to_python({k: v})[k] for k, v in value[M].items()} else: converted[key] None return converted调用方式raw_items resp[Items] clean_items [attr_value_to_python(item) for item in raw_items] df pd.DataFrame(clean_items)需要提醒的是这个函数里面float(value[N]) if . in value[N] else int(value[N])的写法测试时要注意DynamoDB的数字字符串有时候是1.0有时候是1前者转float后者转int如果你的列里两种值都有Pandas会把整列顶成object建议统一用float或者统一用Decimal到Pandas再统一处理。2.4 嵌套结构的处理两种策略各有适用场景DynamoDB的Item里经常有嵌套Map比如用户信息里有{address: {city: 北京, street: xx路}}。这种结构进了DataFrame之后有两种处理方式第一种保持嵌套字典不展开直接放成一列object。适合分析时不怎么用嵌套字段、或者嵌套字段只是偶尔取一下的场景。操作方式也简单df[address]拿到一整列字典需要的时候自己再展开。第二种用pd.json_normalize展开成独立的列。适合分析任务需要反复筛选、聚合嵌套字段的场景。展开之后列名会变成address.city、address.street的形式用起来很顺手。我实际项目中两种都用了。数据量大、嵌套层级深的时候我倾向于在落到DataFrame之前先做一次白名单投影ProjectionExpression只取自己需要的字段然后只展开需要的嵌套部分。这样既不会搞出几十列也不会让DataFrame像一张乱糟糟的宽表。展开嵌套字段的例子from pandas import json_normalize df json_normalize(clean_items, sep_) # 如果只想展开特定列可以先提取成list of dict再normalize target_col df[metadata] normalized json_normalize(target_col, sep_)2.5 大数据量读取时避免内存爆炸全表Scan到一个大表最直接的问题是内存。假设单Item平均2KB100万条Item就是2GB左右的内存还好但如果单Item是20KB甚至上百KB那几百万条就是几十个GB直接OOM都不是开玩笑的。应对策略有三个层面的选择一是按需取列用ProjectionExpression只取计算必需的字段这招最立竿见影。很多人习惯SELECT *在DynamoDB里就是全量返回全部属性有时候明明只要两列却把所有大字段都拉回来了。二是分批落地不要把100万条Item全积累在一个list里再一次性pd.DataFrame可以每5万条做一次DataFrame边缘处理完就释放。也可以写成一个生成器函数Scan一页处理一页内存占用是恒定的。三是利用并行ScanDynamoDB支持Segment并行扫描把全表按段切开每段同时拉。读大表时速度提升明显但要注意消耗的读容量是成倍增加的生产环境要评估成本。3. 写入方向把DataFrame安全地倒进DynamoDB读取方向搞定了写入方向更值得认真对待。DynamoDB的写入是按项计费的每写一个Item消费写容量。如果用Pandas处理完之后DataFrame有几百万行你要做的核心工作是把DataFrame的每一行安全地转成一个DynamoDB的Item然后用批量写接口高效写入。3.1 DataFrame的行是怎么变成Item的Pandas的DataFrame每一行是一个Series每一列有自己的数据类型。要把一行变成DynamoDB的Item需要把每个单元格的值转成对应的AttributeValue格式。一个关键点是DynamoDB要求每个属性必须有类型标签而且类型要合法。float(nan)直接写成{N: nan}是不行的DynamoDB不认pd.NaT也一样。所以写入之前必须处理NaN和None。另一个关键点是Lambda或本地Python环境里如果你直接用Python原生的float/int/str生成Item去调DynamoDB APIboto3的client.put_item实际上可以接受Python原生类型它内部会自动序列化。但问题在于这种自动序列化不是递归的遇到嵌套的dict和list它不会深入处理容易报TypeError。所以最稳妥的方案是先给Pandas里的每一列定好转换函数把DataFrame的行转成一个只有原生Python类型的dict然后让boto3自己去做最后的序列化。这样干净利落可控性也强。3.2 类型装饰器DataFrame dtype到AttributeValue的转换桥这个函数是写入方案的核心我给个参考实现import pandas as pd from decimal import Decimal def dataframe_row_to_item(row: pd.Series) - dict: item {} for col_name, value in row.items(): if value is None or pd.isna(value): # DynamoDB里可以写NULL也可以直接跳过按业务需要二选一 continue if isinstance(value, (str,)): item[col_name] {S: value} elif isinstance(value, bool): item[col_name] {BOOL: value} elif isinstance(value, (int, float, Decimal)): item[col_name] {N: str(value)} elif isinstance(value, (list, tuple)): # 简单list元素类型一致时也可以用L item[col_name] {L: [{S: v} if isinstance(v, str) else {N: str(v)} for v in value]} elif isinstance(value, dict): # M类型需要递归构建 item[col_name] {M: {k: {S: str(v)} for k, v in value.items()}} else: raise TypeError(fUnsupported type for column {col_name}: {type(value)}) return item实际用的时候注意几点pd.isna(value)对单个值判断NaN、NaT、None都有效但如果你容器里面有一个NaN列表这行代码是判断不出来的。数字要转成字符串再传因为DynamoDB的N类型存储格式是字符串。bool是int的子类判断顺序上isinstance(value, bool)要在isinstance(value, int)之前否则布尔会被当成数字处理。如果DataFrame的列类型是datetime64这个函数没有处理所以要么你在写之前先把datetime列统一转成字符串ISO格式要么在函数里加一个分支。3.3 batch_writer的使用为什么它能省一半的时间和精力DynamoDB的batch_write_item一次最多写25个Itemboto3在此基础上封装了batch_writer你可以在with语句里不断put_item它会自动凑满25个一批发送并且自动处理重试包括放行限流导致的未处理项。用batch_writer的代码很简洁table boto3.resource(dynamodb, region_nameus-east-1).Table(my_table) with table.batch_writer() as batch: for _, row in df.iterrows(): batch.put_item(Itemdataframe_row_to_item(row))注意table.batch_writer()用的是resource层所以这里要和前面的client区分开。这句代码后面的语义是batch_writer会缓存Put请求攒够25个一次性提交如果遇到ProvisionedThroughputExceededException它会自动重试。但batch_writer不是万能的它有两个比较坑的点第一batch_writer不保证有序写入如果你有强顺序要求得自己在业务层面做控制。第二batch_writer单次请求最多16MB如果一个Item特别大超过400KB实际DynamoDB单Item上限400KB它会直接报错。所以如果你的DataFrame里有超大字段要先做压缩或分拆。一个我推荐的分批写法防止一次性塞太多行导致前面写入失败难回滚def write_df_to_dynamodb(df, table_name, batch_size2500): table boto3.resource(dynamodb, region_nameus-east-1).Table(table_name) total len(df) for start in range(0, total, batch_size): chunk df.iloc[start:start batch_size] with table.batch_writer() as batch: for _, row in chunk.iterrows(): batch.put_item(Itemdataframe_row_to_item(row)) print(fwritten {min(start batch_size, total)}/{total})每次交2500行也就是100个batch_writer请求万一某一段失败重跑这一段的成本也可控。3.4 写入速度优化并行分段和并发参数的调优大批量写入场景batch_writer单线程往往不够快。一个简单的并行方案是先给DataFrame加一个分片键比如把df切成n段然后用ThreadPoolExecutor并发跑多个batch_writer。但这里要小心一点DynamoDB的写容量是表级别的按WCU计费并发太高容易触发限流。所以并发度不是越大越好而是要根据表的WCU配置去估算。假设表的WCU是1000一条Item写一次消费1WCU那么每秒最多写1000条。你并行10个batch每个batch每秒写100条也就是每秒1000条刚刚好。如果并发20个就有可能超限流然后大量重试反而拖慢总吞吐。实用的估算公式大致是wc_units_per_item 1 # 单Item默认1WCU如果Item大于1KB则按KB数向上取整 ideal_workers max(1, table_wcu // (batch_size_per_worker // 25))当然这个公式很粗糙实际效果要看平均Item大小、网络延迟、以及DynamoDB的突发能力。我一般从4个并发开始观察CloudWatch的WriteThrottleEvents指标如果没限流就往上加限流了就往下降。过程比较务实。另外值得一提的优化是如果写入端有比较强的实时性要求比如写入后马上被读取可以考虑用S3加DynamoDB的两级方案先把Pandas结果以Parquet/CSV格式落到S3需要时再批量导入DynamoDB。但这是另一套架构了这里不展开。4. 实测中的翻车现场这些坑每一个都让我排查了半天下面这些坑是我在对接过程中真正踩过的有的非常隐蔽提前分享出来省得大家再去趟一遍。4.1 Decimal的精度问题算钱、算ID时务必小心TypeDeserializer把N类型转成Decimal而Decimal进Pandas以后Pandas会尝试转成float。如果你的字段用得是大整数比如雪花ID、订单号一旦转float就会丢失精度。比如订单号123456789012345678转成float会变成1.2345678901234568e17后几位就乱了。这个问题有两个解法一是在读取转换时判断列内容如果看起来是整数且位数超过15位保留字符串而不是转数字二是在Pandas里统一用pd.read_csv(..., dtype{order_id: str})的思路在构造DataFrame之后用astype(str)保护关键ID列。我的经验是ID类字段在DynamoDB里优先用S类型字符串存储而不是N。虽然浪费一点存储但彻底规避精度问题。如果已经是N类型了读取转换时可以给出一份字段清单指定哪些列按字符串读。4.2 NaN和None怎么优雅地消失写入方向的NaN是一个很容易被忽视的坑。float(nan)直接放到batch_writer.put_item里有时候不报错但DynamoDB会拒收或报ValidationException。因为NaN不是合法的数字表示。我在前面已经写了跳过NaN的代码但想要更多控制的时候会采用另一个策略写入之前先对DataFrame做一次清洗。df df.where(pd.notnull(df), None) # 把NaN统一替换成None这样转换函数里就能统一处理None可以选择跳过该字段不写入Item也可以写入NULL类型。业务上如果下游习惯用attribute_not_exists判断字段是否存在建议跳过如果喜欢统一读取所有字段再判断NULL则写入NULL。4.3 超出400KB单Item限制的巨型属性DynamoDB单Item上限400KB包括属性名和属性值的全部字节数。如果一个表格里有大量文本字段或者Base64图片数据很容易撞上这个限制。我的处理方式是大字段单独拆到S3DynamoDB只存S3的key或者访问路径。读取时按需回源。这不仅是技术上的妥协也是成本上的实际考虑DynamoDB存储是按GB计算的大字段会显著增加费用。4.4 批量写入限流不要只用一条batch_writerbatch_writer虽然自动重试但如果表配置的WCU很低它会在内部反复退避重试造成整体吞吐极低。我测过一个100WCU的表单线程batch_writer写10GB数据速度慢到像蜗牛同时CloudWatch里全是ThrottledWriteEvents。解决方案就是前面提到的并发写入但并发写入也会放大问题如果一开始就把并发开到16结果表只有100WCU每个worker都在等重试整体性能反而比4并发还差。所以一定要先小并发探测再逐步加。另外提醒一句如果你用的是On-demand模式按请求计费而不是Provisioned模式限流相比之下会少很多但也不是完全无限。AWS的On-demand背靠一组共享容量池突发太猛还是可能断流。4.5 Scan读取大表时的ResponseTimeout碰到过大表Scan吗第一次跑的时候我以为aws的API会等全部返回后来发现同一个请求如果执行时间超长会报ReadTimeoutError。尤其是客户端网络环境一般加上表很大默认超时往往不够。应对办法是给boto3配置更大的read_timeout或者从架构上把纯Scan改为按天分区、按天Scan。比如以天为时间分区设计表分析单天数据时用Query配合KeyConditionExpression效率比全表Scan高一个量级。配置超时的写法from botocore.config import Config config Config(read_timeout120, connect_timeout10) dynamodb boto3.client(dynamodb, region_nameus-east-1, configconfig)这个优化在数据量大时是救命稻草之一。5. 工程化落地把对接代码封装成能复用的模块上面这些逻辑如果只是写脚本时用用散落在一堆临时文件里也无所谓。但当我第三个、第四个项目都遇到同样的DynamoDB与Pandas对接需求时我决定做一个简单的封装把所有脏活累活统一到一个类里。5.1 DynamoPandas类的设计思路这个类的目标是一句代码完成读取一句代码完成写入同时暴露必要的参数让调用者控制类型转换、分页和并发。我建议的类接口长这样class DynamoPandas: def __init__(self, region_nameus-east-1, read_timeout120): self.client boto3.client(dynamodb, region_nameregion_name, configConfig(read_timeoutread_timeout)) self.resource boto3.resource(dynamodb, region_nameregion_name) def read_to_df(self, table_name, projection_expressionNone, expand_nestedFalse, max_recordsNone): ... def write_df(self, df, table_name, batch_size2500, workers4, wcu_estimate1000): ...内部实现就是把第二节和第三节的代码搬进去只是增加了一些参数校验和日志。这个封装类的好处是read_to_df统一处理分页、类型转换、分页内存管理。write_df统一处理NaN清洗、类型转换、并发写入和重试。任何新项目只要import这个类就能立即使用不用每次再写一遍踩坑逻辑。5.2 并发写入的细节线程池与分片并发写入部分我的实现大致是先把DataFrame按行索引均匀切n段每一段交给一个线程线程内部用batch_writer批量提交。每个线程自己带一个独立的resource对象boto3.client和resource可以多线程共用但通过分开更稳妥避免共享连接带来的偶发问题。一段示意代码from concurrent.futures import ThreadPoolExecutor def _write_chunk(chunk_df, table_name): table boto3.resource(dynamodb).Table(table_name) with table.batch_writer() as batch: for _, row in chunk_df.iterrows(): batch.put_item(Itemdataframe_row_to_item(row)) def write_df(self, df, table_name, workers4, chunk_size2000): chunks [df.iloc[i:ichunk_size] for i in range(0, len(df), chunk_size)] with ThreadPoolExecutor(max_workersworkers) as executor: futures [executor.submit(_write_chunk, chunk, table_name) for chunk in chunks] for f in futures: f.result()重点在于chunk_size和workers两个参数不能拍脑袋。我一般让workers * chunk_size * 平均Item字节数对应的每秒写入请求数略小于表WCU的80%留20%余量给突刺。5.3 测试方式本地mock还是真表写这类功能测试不能少但每次都连真表很费钱也费时间。我用的方案是类型转换和DataFrame组装相关逻辑用moto库本地mock批量写入的性能测试才到真表上做。moto的用法大概是这样import moto moto.mock_dynamodb def test_write_and_read(): dynamodb boto3.resource(dynamodb, region_nameus-east-1) dynamodb.create_table(...) # 然后走你封装的读写逻辑mock环境里没有真正的容量限制和网络延迟所以性能测试的结果只能用来验证功能不能当作性能基准。我测过mock和真表之间的写入速度差距能到一个数量级这个务必要注意。5.4 一套完整的调用示例最后给一个完整示例涵盖了读、处理、写三个环节。假设我们有一个订单表orders里面存着订单Item现在要把所有订单按用户维度聚合成订单数统计再写回一个配置表order_stats。dp DynamoPandas(region_nameus-east-1) # 读取只取需要的字段展开嵌套的shipping地址 orders_df dp.read_to_df( orders, projection_expressionuser_id, amount, shipping, created_at, expand_nestedTrue, ) # 常规Pandas处理 orders_df[amount] orders_df[amount].astype(float) orders_df[month] pd.to_datetime(orders_df[created_at]).dt.to_period(M) stats orders_df.groupby([user_id, month]).agg( order_count(amount, count), total_amount(amount, sum), ).reset_index() # 写回 dp.write_df(stats, order_stats, workers2, chunk_size500)这段代码看着简单但背后每一环都包含了前面讲的处理逻辑。5.5 维护和扩展方向封装完成之后后续可以考虑几个扩展方向支持从S3导入的离线批量写入模式、支持DynamoDB Streams配合Pandas做增量处理、以及把类型转换规则做成可配置的JSON映射这样不同团队就可以用不同规则处理同一张表的数据。但这些多属于锦上添花核心的读写链条已经能覆盖大多数业务需求了。我实际用下来这个封装稳定运行了大半年读写几亿条数据没有再在类型转换和批量写入上摔过跤。如果你也在纠结Pandas和DynamoDB的对接方案希望这篇梳理能让你少走一些弯路。