Python dask 分布式计算pandas 跑不动的数据dask 怎么解决一张表 2000 万行pandas 直接 OOM 罢工试试 dask用法跟 pandas 几乎一样但数据再多也不怕。一、pandas 到 dask不是替代是升级做数据分析最崩溃的瞬间是什么对我来说是跑了 20 分钟的 pandas 脚本最后弹出一个MemoryError。pandas 的设计哲学是把数据全部加载到内存里操作。单机内存 16G、32G 的时候处理百万级数据没啥问题。但现在的业务数据动不动就上千万行、几十个字段宽表一张表几个 G 很常见pandas 是真的扛不住了。dask 的思路不一样它不把数据一次性全读进内存而是分块读、分块算用完一块就释放。具体来说dask 做了两件事惰性计算Lazy Evaluation你写操作的时候它不执行只是构建一个计算图。等你调用.compute()的时候它才真正执行。这样就可以在计算前做全局优化。任务分片Task Partitioning把大数据拆成很多小数据块partition每个块独立处理。一个块算完就释放内存再处理下一个。所以 dask 不是 pandas 的替代品而是当你数据大到 pandas 撑不住时的升级方案。日常小数据继续用 pandas大数据场景切 dask。为什么 dask 不是pandas 替代品而是pandas 超时后的升级方案这里有个性能悖论在处理 10 万行数据时pandas 比 dask 快 3-5 倍。原因是 dask 的惰性计算图构建、分片调度、序列化/反序列化都是有固定开销的——处理的数据量越小这些固定开销占的比重就越高。pandas 直接对内存中的 C 数组做向量化操作没有中间商。只有当真的一台机器装不下数据时比如 2000 万行 CSV 文件 16GBdask 的分批加载优势才开始抵消调度开销。所以不要用 dask 处理小数据——那等于抱着消防栓浇花。二、dask DataFrame跟 pandas 几乎一样的 APIdask 最友好的一点是它的 API 高度模仿 pandas。大部分时候你把import pandas as pd改成import dask.dataframe as dd代码基本不用动。import dask.dataframe as dd import pandas as pd # # 1. 读取 CSV —— 跟 pandas 语法一模一样 # # dask 不会立刻读文件只是记录了要读这个文件的元信息 # 用 blocksize 控制每个分片的大小建议 64MB~128MB df dd.read_csv( sales_2024.csv, blocksize64MB, # 每个分片 64MB控制内存峰值 dtype{user_id: str}, # 指定类型避免自动推断出错 parse_dates[order_date] ) # # 2. 基础操作 —— 写法跟 pandas 完全一致 # # 过滤、分组、聚合语法和 pandas 一毛一样 filtered df[df[amount] 100] # 筛选高客单价订单 monthly_sales ( df.groupby(df[order_date].dt.month) # 按月分组 [amount].sum() # 求和 ) # # 3. 触发计算 —— 这是 dask 和 pandas 最大的区别 # # .compute() 会把 Dask DataFrame 转成 pandas DataFrame # 调用之前所有操作都只是计划现在才真正执行 result monthly_sales.compute() # 此时才真正执行计算 print(result)需要注意一个细节dask 的惰性计算意味着每次调用.compute()都会重新从头算。如果你要对同一个数据做多次聚合最好先.persist()把中间结果缓存到内存# 错误做法每次 .compute() 都重算一遍 avg df[amount].mean().compute() # 第1次从头算 std df[amount].std().compute() # 第2次又从头算 !!! # 正确做法persist 缓存中间结果 df_cached df[[amount, order_date, region]].persist() avg df_cached[amount].mean().compute() # 从缓存算 std df_cached[amount].std().compute() # 从缓存算快很多三、dask 调优partition 是核心dask 的性能瓶颈通常出在 partition 数量的选择上。太少每个分片太大内存还是撑不住太多调度开销比计算开销还大。一个实用的经验公式每个 partition 大小控制在 100MB 左右。比如你的数据是 5G那就设 50 个分区。import dask.dataframe as dd df dd.read_csv(big_data.csv, blocksize64MB) # # 查看当前分片数量 —— 调优前先看一眼 # print(f当前分片数: {df.npartitions}) # # 调整分片数repartition 重新分配 # # npartitions 太小 每个分片太大 内存不够 # npartitions 太大 调度开销 计算开销 反而变慢 # 经验值每个 partition 约 100MB file_size_mb 5000 # 假设文件 5G target_partitions file_size_mb // 100 # 每个分片 100MB df_optimized df.repartition(npartitionstarget_partitions)另一个性能杀手是数据倾斜。比如分组聚合时某个 key 的数据量特别大它所在的分片就会成为瓶颈。可以用map_partitions在分片内部做预聚合来缓解# 分片内预聚合 —— 减少 shuffle 的数据量 def pre_agg_in_partition(partition): 在每个分片内先做一次聚合减少后续 shuffle 的数据量 return partition.groupby(user_id)[amount].sum().reset_index() # map_partitions 对每个分片独立执行函数 pre_aggregated df.map_partitions(pre_agg_in_partition) # 再做全局聚合数据量小了很多 final pre_aggregated.groupby(user_id)[amount].sum().compute()四、dask 分布式从单机到集群dask 单机模式可以解决内存不够的问题但要真正加速计算还得上分布式。from dask.distributed import Client, LocalCluster # # 方式一本地模拟分布式开发调试用 # cluster LocalCluster( n_workers4, # 4个worker进程 threads_per_worker2, # 每个worker 2线程 memory_limit4GB # 每个worker内存上限 ) client Client(cluster) # # 方式二连接已有集群生产环境用 # # client Client(scheduler-address:8786) # # 设置后所有 dask 操作自动用集群资源 # import dask.dataframe as dd df dd.read_csv(s3://my-bucket/big_data/*.csv) # 聚合操作会自动分布到各个 worker 上 result df.groupby(category)[sales].sum().compute() # 查看 dashboard默认端口 8787 # 可视化任务执行情况、内存使用、数据传输 print(fDashboard: {client.dashboard_link})分布式场景下有几个实用技巧用client.persist()代替.persist()数据会分散缓存在多个 worker 上避免全局排序.sort_values()需要把所有数据 shuffle 到一个 worker非常慢优先用有损操作比如approx_percentile替代精确百分位速度快很多五、总结 踩坑提醒dd.read_csv的blocksize不是每个分区精确大小如果你设blocksize64MBdask 会从 64MB 的偏移量开始切割文件。但 CSV 的行边界不一定在 64MB 处对齐——切割点可能正好落在某一行中间导致该行被截断。dask 会在每个分片的末尾往前找到下一个换行符所以实际分片大小可能比 64MB 略大。如果你对分片数量有严格要求读取后用.repartition()做二次调整。惰性求值 随机抽样会得到不随机的结果df.sample(frac0.01).compute()每次执行都会重新随机抽样但由于 dask 的惰性计算图是确定的如果你没设random_state两次compute()的抽样结果可能完全不同——这在需要对抽样结果做验证分析的场景下是个坑。建议每次.compute()前重新设一次random_state或者persist()后再抽样。分布式集群中memory_limit不是硬限制你设了memory_limit4GB期望 worker 到了 4GB 就停止接收新任务。但实际上 dask 的 memory_limit 是软限制——它只在 scheduler 调度新任务时检查当前内存使用不会主动 kill 已运行的任务。如果你的一个任务在运行中自己申请了 8GB比如一个大排序dask 不会阻止worker 会直接 OOM 被杀。解决方案是同时设置--memory-limit和--memory-spill4GB 时 spill 到磁盘或者在代码层面控制单任务数据量。dask 的核心价值就一句话用 pandas 的写法处理 pandas 吃不下的数据。它不要求你学一套全新的 API大部分代码从 pandas 迁移过来只是换个 import。但它通过惰性计算和分片机制让单机能处理的场景从百万行升级到十亿行。在实际工作中我现在的默认策略是100 万行以内pandas简单直接100 万 ~ 1000 万行dask 单机模式加个dd.read_csv就行1000 万行以上dask 分布式集群并行加速当然 dask 也有不擅长的场景比如频繁的 join 操作、需要全局排序的复杂计算。这些时候就该 Spark 上场了后面我们会聊到。为什么 dask 不擅长 JOIN 和全局排序JOIN 在 dask 里需要对两个 DataFrame 的 partition 做对齐——如果两个 DataFrame 的分区数不同或分区方式不同比如按不同列做了索引JOIN 时就需要 shuffle 全部数据等于把所有数据在集群里洗一遍牌。Spark 有专门的 Sort-Merge Join 和 Broadcast Join 优化器能根据数据量自动选择最佳策略dask 在这方面远没有 Spark 成熟。全局排序同理——sort_values()需要把所有数据集中到一个 partition如果数据是 100GB那这个 partition 就是 100GB——直接炸。在这些场景下dask 更像是能跑但不保证快而 Spark 提供了生产级的优化保障。本文由朱大喜原创欢迎点赞收藏有问题评论区交流~
Python dask 分布式计算pandas 跑不动的数据dask 怎么解决一张表 2000 万行pandas 直接 OOM 罢工试试 dask用法跟 pandas 几乎一样但数据再多也不怕。一、pandas 到 dask不是替代是升级做数据分析最崩溃的瞬间是什么对我来说是跑了 20 分钟的 pandas 脚本最后弹出一个MemoryError。pandas 的设计哲学是把数据全部加载到内存里操作。单机内存 16G、32G 的时候处理百万级数据没啥问题。但现在的业务数据动不动就上千万行、几十个字段宽表一张表几个 G 很常见pandas 是真的扛不住了。dask 的思路不一样它不把数据一次性全读进内存而是分块读、分块算用完一块就释放。具体来说dask 做了两件事惰性计算Lazy Evaluation你写操作的时候它不执行只是构建一个计算图。等你调用.compute()的时候它才真正执行。这样就可以在计算前做全局优化。任务分片Task Partitioning把大数据拆成很多小数据块partition每个块独立处理。一个块算完就释放内存再处理下一个。所以 dask 不是 pandas 的替代品而是当你数据大到 pandas 撑不住时的升级方案。日常小数据继续用 pandas大数据场景切 dask。为什么 dask 不是pandas 替代品而是pandas 超时后的升级方案这里有个性能悖论在处理 10 万行数据时pandas 比 dask 快 3-5 倍。原因是 dask 的惰性计算图构建、分片调度、序列化/反序列化都是有固定开销的——处理的数据量越小这些固定开销占的比重就越高。pandas 直接对内存中的 C 数组做向量化操作没有中间商。只有当真的一台机器装不下数据时比如 2000 万行 CSV 文件 16GBdask 的分批加载优势才开始抵消调度开销。所以不要用 dask 处理小数据——那等于抱着消防栓浇花。二、dask DataFrame跟 pandas 几乎一样的 APIdask 最友好的一点是它的 API 高度模仿 pandas。大部分时候你把import pandas as pd改成import dask.dataframe as dd代码基本不用动。import dask.dataframe as dd import pandas as pd # # 1. 读取 CSV —— 跟 pandas 语法一模一样 # # dask 不会立刻读文件只是记录了要读这个文件的元信息 # 用 blocksize 控制每个分片的大小建议 64MB~128MB df dd.read_csv( sales_2024.csv, blocksize64MB, # 每个分片 64MB控制内存峰值 dtype{user_id: str}, # 指定类型避免自动推断出错 parse_dates[order_date] ) # # 2. 基础操作 —— 写法跟 pandas 完全一致 # # 过滤、分组、聚合语法和 pandas 一毛一样 filtered df[df[amount] 100] # 筛选高客单价订单 monthly_sales ( df.groupby(df[order_date].dt.month) # 按月分组 [amount].sum() # 求和 ) # # 3. 触发计算 —— 这是 dask 和 pandas 最大的区别 # # .compute() 会把 Dask DataFrame 转成 pandas DataFrame # 调用之前所有操作都只是计划现在才真正执行 result monthly_sales.compute() # 此时才真正执行计算 print(result)需要注意一个细节dask 的惰性计算意味着每次调用.compute()都会重新从头算。如果你要对同一个数据做多次聚合最好先.persist()把中间结果缓存到内存# 错误做法每次 .compute() 都重算一遍 avg df[amount].mean().compute() # 第1次从头算 std df[amount].std().compute() # 第2次又从头算 !!! # 正确做法persist 缓存中间结果 df_cached df[[amount, order_date, region]].persist() avg df_cached[amount].mean().compute() # 从缓存算 std df_cached[amount].std().compute() # 从缓存算快很多三、dask 调优partition 是核心dask 的性能瓶颈通常出在 partition 数量的选择上。太少每个分片太大内存还是撑不住太多调度开销比计算开销还大。一个实用的经验公式每个 partition 大小控制在 100MB 左右。比如你的数据是 5G那就设 50 个分区。import dask.dataframe as dd df dd.read_csv(big_data.csv, blocksize64MB) # # 查看当前分片数量 —— 调优前先看一眼 # print(f当前分片数: {df.npartitions}) # # 调整分片数repartition 重新分配 # # npartitions 太小 每个分片太大 内存不够 # npartitions 太大 调度开销 计算开销 反而变慢 # 经验值每个 partition 约 100MB file_size_mb 5000 # 假设文件 5G target_partitions file_size_mb // 100 # 每个分片 100MB df_optimized df.repartition(npartitionstarget_partitions)另一个性能杀手是数据倾斜。比如分组聚合时某个 key 的数据量特别大它所在的分片就会成为瓶颈。可以用map_partitions在分片内部做预聚合来缓解# 分片内预聚合 —— 减少 shuffle 的数据量 def pre_agg_in_partition(partition): 在每个分片内先做一次聚合减少后续 shuffle 的数据量 return partition.groupby(user_id)[amount].sum().reset_index() # map_partitions 对每个分片独立执行函数 pre_aggregated df.map_partitions(pre_agg_in_partition) # 再做全局聚合数据量小了很多 final pre_aggregated.groupby(user_id)[amount].sum().compute()四、dask 分布式从单机到集群dask 单机模式可以解决内存不够的问题但要真正加速计算还得上分布式。from dask.distributed import Client, LocalCluster # # 方式一本地模拟分布式开发调试用 # cluster LocalCluster( n_workers4, # 4个worker进程 threads_per_worker2, # 每个worker 2线程 memory_limit4GB # 每个worker内存上限 ) client Client(cluster) # # 方式二连接已有集群生产环境用 # # client Client(scheduler-address:8786) # # 设置后所有 dask 操作自动用集群资源 # import dask.dataframe as dd df dd.read_csv(s3://my-bucket/big_data/*.csv) # 聚合操作会自动分布到各个 worker 上 result df.groupby(category)[sales].sum().compute() # 查看 dashboard默认端口 8787 # 可视化任务执行情况、内存使用、数据传输 print(fDashboard: {client.dashboard_link})分布式场景下有几个实用技巧用client.persist()代替.persist()数据会分散缓存在多个 worker 上避免全局排序.sort_values()需要把所有数据 shuffle 到一个 worker非常慢优先用有损操作比如approx_percentile替代精确百分位速度快很多五、总结 踩坑提醒dd.read_csv的blocksize不是每个分区精确大小如果你设blocksize64MBdask 会从 64MB 的偏移量开始切割文件。但 CSV 的行边界不一定在 64MB 处对齐——切割点可能正好落在某一行中间导致该行被截断。dask 会在每个分片的末尾往前找到下一个换行符所以实际分片大小可能比 64MB 略大。如果你对分片数量有严格要求读取后用.repartition()做二次调整。惰性求值 随机抽样会得到不随机的结果df.sample(frac0.01).compute()每次执行都会重新随机抽样但由于 dask 的惰性计算图是确定的如果你没设random_state两次compute()的抽样结果可能完全不同——这在需要对抽样结果做验证分析的场景下是个坑。建议每次.compute()前重新设一次random_state或者persist()后再抽样。分布式集群中memory_limit不是硬限制你设了memory_limit4GB期望 worker 到了 4GB 就停止接收新任务。但实际上 dask 的 memory_limit 是软限制——它只在 scheduler 调度新任务时检查当前内存使用不会主动 kill 已运行的任务。如果你的一个任务在运行中自己申请了 8GB比如一个大排序dask 不会阻止worker 会直接 OOM 被杀。解决方案是同时设置--memory-limit和--memory-spill4GB 时 spill 到磁盘或者在代码层面控制单任务数据量。dask 的核心价值就一句话用 pandas 的写法处理 pandas 吃不下的数据。它不要求你学一套全新的 API大部分代码从 pandas 迁移过来只是换个 import。但它通过惰性计算和分片机制让单机能处理的场景从百万行升级到十亿行。在实际工作中我现在的默认策略是100 万行以内pandas简单直接100 万 ~ 1000 万行dask 单机模式加个dd.read_csv就行1000 万行以上dask 分布式集群并行加速当然 dask 也有不擅长的场景比如频繁的 join 操作、需要全局排序的复杂计算。这些时候就该 Spark 上场了后面我们会聊到。为什么 dask 不擅长 JOIN 和全局排序JOIN 在 dask 里需要对两个 DataFrame 的 partition 做对齐——如果两个 DataFrame 的分区数不同或分区方式不同比如按不同列做了索引JOIN 时就需要 shuffle 全部数据等于把所有数据在集群里洗一遍牌。Spark 有专门的 Sort-Merge Join 和 Broadcast Join 优化器能根据数据量自动选择最佳策略dask 在这方面远没有 Spark 成熟。全局排序同理——sort_values()需要把所有数据集中到一个 partition如果数据是 100GB那这个 partition 就是 100GB——直接炸。在这些场景下dask 更像是能跑但不保证快而 Spark 提供了生产级的优化保障。本文由朱大喜原创欢迎点赞收藏有问题评论区交流~
