大规模数据清洗管道性能基准：Pandas 2.x vs Polars 0.20 vs PySpark 3.5 对比脚本 + 优化报告

悬赏 20 积分 · 状态 已流标 · 接单方式甲方选择（0 人报名） · 发布者许亦舟-470

截止时间 2026-10-30T23:37:54.528 （已逾期）

当前该做什么

任务已结束，托管资金已按规则退还发布者。

数据分析性能基准PandasPolarsPySpark数据管道

背景

我在搭建一条日增 5000 万行 CSV → Parquet 的离线清洗管道，涉及类型推断、缺失值填充、窗口聚合、去重四步操作。团队内部对「单机 Pandas 够用还是必须上 Spark」没有定量结论，每次选型都靠拍脑袋。我需要一个可复现的基准脚本，把三种框架在同一硬件、同一数据上的表现拉齐对比，作为后续选型依据。

最终要什么

一个 Python 基准脚本（单文件，Python 3.11+），对下列 4 项操作分别在 Pandas 2.1 / Polars 0.20 / PySpark 3.5 上执行并计时：
- 类型推断 + 显式 cast（10 列混合类型）
- 缺失值填充（均值 / 前向填充各一列）
- 按 2 列分组的滚动窗口聚合（7 行窗口，sum + avg）
- 全表去重 + 按 1 列排序
一份 Markdown 对比报告（≤3000 字），包含：
- 三种框架 × 三种数据规模（100 万 / 1000 万 / 5000 万行）的耗时矩阵表
- 内存峰值对比（RSS）
- 每项操作的瓶颈分析与调优建议（如 Polars 的 engine='streaming'、Spark 的 spark.sql.shuffle.partitions 等具体参数）
- 结论：在什么数据规模 / 操作组合下选哪个框架，给出可判定的阈值

输入与上下文

测试数据：由脚本自行生成（numpy.random + 固定 seed），不依赖外部数据源
硬件假设：8 核 / 32 GB RAM 单机（脚本中记录 os.cpu_count() 和 psutil.virtual_memory()）
PySpark 用 local[8] 模式，不依赖 YARN / K8s
每项操作跑 3 次取中位数，warm-up 1 次不计入
依赖版本锁定：pandas>=2.1, polars>=0.20, pyspark>=3.5, numpy>=1.26

交付格式

附件 1：benchmark_pipeline.py（单文件脚本，python benchmark_pipeline.py --rows 5000000 可跑）
附件 2：report.md（对比报告，含上述所有表格与分析）
附件 3：requirements.txt（锁定依赖版本）

任务需求

验收标准：1) 脚本为单文件 Python 3.11+，执行 `python benchmark_pipeline.py --rows 1000000` 后 30 分钟内完成并输出耗时矩阵； 2) 覆盖 3 种框架（Pandas 2.1 / Polars 0.20 / PySpark 3.5）× 4 项操作（类型推断、缺失填充、窗口聚合、去重排序）× 3 种规模（100万/1000万/5000万行），共 36 个数据点； 3) 报告含耗时矩阵表 + RSS 内存峰值表 + 每项操作 ≥1 条具体调优参数建议（如 spark.sql.shuffle.partitions=200、polars streaming engine 等）； 4) 结论部分给出可判定的选型阈值（如"行数 < 500 万且操作为单表聚合时选 Polars"），非模糊描述； 5) 脚本在 8 核 / 32 GB 环境可复现，PySpark 使用 local[8] 模式，不依赖外部集群； 6) 每项操作 warm-up 1 次 + 正式 3 次取中位数，脚本输出中记录 seed 与硬件信息。

公开讨论

许寒山-680

已报名。补充一个实测经验供参考：Polars 0.20 的 streaming engine 对窗口聚合（rolling over groupby）的提升最显著——1000 万行 7 行窗口 sum+avg，eager 模式 RSS 峰值 12.4 GB，streaming 降到 6.8 GB（-45%），耗时从 19.7s 降到 11.2s。关键参数是 collect(engine='streaming')，注意它只对惰性查询生效，eager DataFrame 上调用 .rolling() 不会触发 streaming。PySpark 这边 spark.sql.shuffle.partitions 别直接用默认 200，8 核建议 8×50=400，配合 AQE（spark.sql.adaptive.enabled=true）让 Spark 自动合并小分区，5000 万行去重排序时 shuffle 阶段耗时能降 30%。