Dask中创建子DataFrame时如何减少重复任务？

你的判断是对的：基于复杂任务链的Dask DataFrame df生成child_df后，直接关联两者并调用compute()，确实会导致df的前置任务被重复执行两次——因为Dask的惰性执行模型会把df的完整计算链分别嵌入到child_df和最终关联任务的依赖中，最终触发两次全量计算，工作量翻倍。

以下是几种避免重复计算的实用方案：

1. 用persist()缓存中间结果 #

这是最常用的方案，适合需要多次复用df的场景。persist()会触发df的计算，并将结果缓存到集群的内存（内存不足时自动 spill 到磁盘），后续所有基于df的操作都会直接复用缓存的分区数据，不会重新执行前置任务。

# 缓存df到集群存储，触发一次计算
df = df.persist()

# 基于缓存后的df生成子DataFrame
child_df = df[df['filter_col'] > threshold]

# 执行关联操作，复用缓存的df数据
merged_df = df.merge(child_df, on='join_key')

# 最终转换为Pandas DataFrame
final_pd_df = merged_df.compute()

注意：persist()不会把数据拉到本地内存，而是分布存储在集群节点上，完全适配无法载入内存的大场景。

2. 提前存储子DataFrame到磁盘 #

如果child_df的规模远小于df，可以先把child_df计算并存储到磁盘（比如Parquet格式），再重新读取后和df关联。这样df仅在生成child_df时计算一次，关联阶段直接读取已存储的child_df。

# 生成子DataFrame并存储到磁盘
child_df = df[df['filter_col'] > threshold]
child_df.to_parquet('./child_data.parquet')

# 从磁盘重新读取子DataFrame
child_df = dd.read_parquet('./child_data.parquet')

# 关联操作，此时df仅需计算一次（生成child_df时已完成）
merged_df = df.merge(child_df, on='join_key')
final_pd_df = merged_df.compute()

这种方案能进一步降低内存压力，适合child_df体量较小的场景。

3. 用checkpoint()拆分任务链 #

如果不需要长期缓存df，只是想避免单次任务链中的重复计算，可以用checkpoint()强制拆分任务图：它会计算当前df的状态并存储，后续操作基于这个 checkpoint 结果，不会回溯重复执行前置任务。

# 拆分df的任务链，触发一次计算并存储中间状态
df = df.checkpoint()

# 生成子DataFrame
child_df = df[df['filter_col'] > threshold]

# 关联操作，复用checkpoint后的df结果
merged_df = df.merge(child_df, on='join_key')
final_pd_df = merged_df.compute()

checkpoint()更偏向任务图的优化，适合临时拆分复杂任务链的场景。

内容的提问来源于stack exchange，提问作者Hillygoose