Python 多进程数据处理multiprocessing 不如 dask 顺手一张 800 万行的订单表需要做用户分群和交叉统计。用 Pandas 单进程跑要 8 分钟换成 multiprocessing 手写进程池、手动分块、手动合并结果代码写了 80 行结果合并时内存又爆了。后来换成 Dask改了三行代码1 分半就跑完了。今天聊聊为什么在数据处理场景里Dask 比裸写 multiprocessing 更顺手。一、multiprocessing 的三座大山Python 多进程本身不算难Pool.map一行就能并行化。但数据处理这件事远不止把计算分布到多个进程里那么简单。看一个典型场景800 万行 DataFrame按用户做完聚合后还要做全局排序取 Top 100。用 multiprocessing 的实现大概长这样import multiprocessing as mp import pandas as pd def process_chunk(df_chunk: pd.DataFrame) - pd.DataFrame: 单块数据处理按 user_id 聚合金额 return df_chunk.groupby(user_id)[amount].sum().reset_index() def parallel_groupby(df: pd.DataFrame, n_workers: int 4) - pd.DataFrame: 用 multiprocessing 并行分组聚合 设计意图将大 DataFrame 手动切分成 n_workers 份 每个子进程处理一块最后合并结果并二次聚合。 问题分块、序列化、合并全部需要手动处理代码量与出错概率同步增加。 # 1. 手动按行切分 chunk_size len(df) // n_workers chunks [df.iloc[i * chunk_size : (i 1) * chunk_size] for i in range(n_workers)] # 2. 进程池映射 with mp.Pool(n_workers) as pool: results pool.map(process_chunk, chunks) # 3. 手动合并每个子进程的 DataFrame 通过 pickle 序列化传回来 merged pd.concat(results, ignore_indexTrue) # 4. 合并后需要二次聚合因为同一个 user_id 可能跨多个 chunk final merged.groupby(user_id)[amount].sum().reset_index() return final这里暴露了 multiprocessing 的三个核心痛点第一座山手动分块。你得自己决定怎么切数据。按行均匀切如果某个用户的交易记录全在一个块里还好如果分散在多个块里合并后还得二次聚合。这个数据均匀性问题完全落在你头上。第二座山序列化开销。Pool.map把每个 chunk 通过pickle序列化传给子进程子进程算完后再序列化传回来。对于大 DataFrame序列化开销可能比计算本身还大。第三座山内存管理。Pool.map会把所有结果收集到内存再返回。8 个 chunk 的结果同时存在内存里很容易把内存撑爆。这不是 multiprocessing 的 bug而是它的定位 —- 它是一个通用的并行原语不是为数据处理定制的。就像一把瑞士军刀能剪能锯但你要用它来劈柴还是专门的大斧头顺手。二、Dask 的数据分区哲学Dask DataFrame 的核心思路就是把 Pandas DataFrame 当 LEGO 积木。一个大 DataFrame 被自动切分成若干小块partition每个 partition 是标准 Pandas DataFrame。你写的代码几乎跟 Pandas 一毛一样Dask 自动在底层拆分成对每个 partition 的操作。import dask.dataframe as dd # 用 Dask 读大 CSV几乎不需要改动 Pandas 代码 ddf dd.read_csv(large_orders.csv, dtype{user_id: int32, amount: float32}, blocksize128MB) # 自动按 128MB 分成多个 partition # Pandas 风格的语法Dask 自动处理分块和合并 result ddf.groupby(user_id)[amount].sum().nlargest(100) # .compute() 时才真正执行 top_users result.compute()关键区别在于Dask 替你把分块、调度、合并这三件事全包了。你写的是逻辑Dask 负责物理执行。来看看对比graph TB subgraph multiprocessing流程 A1[手动读数据] -- A2[手动切分 DataFrame] A2 -- A3[手动管理进程池] A3 -- A4[手动收集结果] A4 -- A5[手动合并聚合] end subgraph Dask流程 B1[dask.read_csv 自动分块] -- B2[写 Pandas 风格代码] B2 -- B3[.compute 自动调度执行] B3 -- B4[自动合并结果] end style A2 fill:#f99,stroke:#333 style A3 fill:#f99,stroke:#333 style A5 fill:#f99,stroke:#333 style B1 fill:#9f9,stroke:#333 style B3 fill:#9f9,stroke:#333三行 Dask 代码抵过 multiprocessing 的 20 行手动操作代码而且不容易出错。三、Task GraphDask 的懒执行魔法Dask 真正的威力在于它不立即执行。当你写ddf.groupby(...)它只是构建一个任务图Task Graph记录了读文件 → 分组 → 聚合 → 排序这一串操作。调用.compute()时Dask 才开始分析这个图做两件厉害的事# 查看任务图结构不执行计算 dask_result ddf.groupby(user_id)[amount].sum().nlargest(100) print(dask_result.visualize()) # 输出任务依赖图融合算子如果发现filter后面接groupby它会把过滤操作下推到 partition 级别避免不必要的数据传输。动态调度根据每个 partition 的数据量和当前 CPU 负载动态分配任务给 worker。from dask.distributed import Client, LocalCluster # 启动本地分布式集群 cluster LocalCluster( n_workers4, threads_per_worker2, memory_limit4GB # 限制每个 worker 的内存防止 OOM ) client Client(cluster) # 与 multiprocessing 不同Dask 的 worker 有内存管理 # 中间结果在 worker 之间通过零拷贝共享不需要序列化传回主进程 result ddf.groupby(user_id)[amount].sum().compute()这里对比最明显multiprocessing 里子进程的结果必须 pickle 序列化后传回主进程而 Dask 的 worker 之间可以直接传递数据的引用。对于需要多阶段计算的场景比如先过滤再分组再排序Dask 避免了多次序列化反序列化。四、选型决策什么场景坚持用 multiprocessingDask 虽好用但不是万能。有几个场景裸 multiprocessing 反而更合适场景一非 DataFrame 的计算。如果你的并行计算不涉及 Pandas比如大量小文件的内容哈希校验、独立的 API 调用multiprocessing 更轻量from multiprocessing import Pool def hash_file(filepath: str) - str: 对单个文件计算哈希——不涉及 DataFramemultiprocessing 更合适 import hashlib with open(filepath, rb) as f: return hashlib.sha256(f.read()).hexdigest() # 200 个小文件的哈希计算multiprocessing 最简单直接 with Pool(8) as pool: hashes pool.map(hash_file, file_list)场景二需要共享内存的元数据处理。Dask 的任务图有调度开销对于单个计算只有几毫秒的场景比如 100 万行数据每行做一个极快的判断multiprocessing 的直接映射反而更快。# 极轻量级计算每行判断是否合法计算本身在微秒级 # 这种场景 Dask 的调度开销比计算还大 def is_valid(row): return len(row[name]) 0 and row[age] 0 with Pool(8) as pool: # 字典列表而不是 DataFramemultiprocessing 开销小 results pool.map(is_valid, data_rows)决策原则总结场景推荐工具原因Pandas DataFrame 并行Dask自动分块、无缝语法纯 Python 函数并行multiprocessing无额外依赖、调度轻超大数据集 内存Dask分区惰加载不爆内存GPU 计算multiprocessingDask 对 GPU 调度不够好进度条断点续跑Dask自带进度追踪五、总结multiprocessing 是 Python 并行的基石但在数据处理场景它就是一盒散装的螺丝刀配件你得自己组装Dask 是一整套电动工具箱开箱即用。大多数数据分析场景读 CSV → 清洗 → 聚合 → 输出直接用 Dask 改三行代码pd.read_csv→dd.read_csv后面.compute()就能把串行变成并行。除非你的计算完全脱离 DataFrame 范式否则别在 multiprocessing 的手动管理上浪费时间。