执行Dask Merge时遇磁盘空间不足问题求解决方案

一、优先优化自连接逻辑（最有效方案） #

你的需求是找出同一供应商（NiuSup）下的不同客户（NiuCust）配对，直接用全量自连接会生成海量中间数据（比如一个供应商有1000个客户，会产生100万条临时记录），这是磁盘占用爆仓的核心原因。改用分组后生成配对的方式，能大幅削减数据量：

import dask.dataframe as dd
import pandas as pd

def generate_unique_cust_pairs(group):
    # 获取当前供应商下的唯一客户列表
    unique_custs = group['NiuCust'].unique()
    # 生成无序的客户配对（避免重复的(x,y)和(y,x)）
    pairs = [(cust_a, cust_b) for idx, cust_a in enumerate(unique_custs) 
             for cust_b in unique_custs[idx+1:]]
    return pd.DataFrame(pairs, columns=['NiuCust_x', 'NiuCust_y'])

# 调整分区数：500万数据设置10-20个分区即可，无需用NiuSup的唯一值数量
NetworkDD = dd.from_pandas(Network, npartitions=15)
# 分组应用函数，生成结果
NodesSharingSupplier = NetworkDD.groupby('NiuSup').apply(
    generate_unique_cust_pairs,
    meta={'NiuCust_x': int, 'NiuCust_y': int}
).compute()

这种方式只在每个供应商分组内生成必要的配对，中间数据量比全量merge小几个数量级，基本能解决磁盘空间问题。

二、你提出的两个方案的可行性说明 #

1. 将临时数据存储到Dropbox #

Dask默认使用本地磁盘作为临时存储，要切换到Dropbox需要先通过工具（如rclone）将Dropbox挂载为本地目录，再修改Dask的临时存储路径：

import dask
# 替换为你的Dropbox挂载路径
dask.config.set(temporary_directory='/mnt/dropbox/temp')

但Dropbox的IO性能远低于本地SSD，会导致计算速度大幅下降，不推荐作为优先方案。

2. 强制Dask不写入本地内容 #

Dask只有在所有计算数据能完全放入内存时才能避免写入磁盘。可以尝试调整内存相关配置：

# 关闭内存使用检测，让更多数据留在内存
dask.config.set({'dataframe.memory.usage': False})

# 或使用分布式集群指定更大内存限制（根据你的机器内存调整）
from dask.distributed import Client, LocalCluster
cluster = LocalCluster(memory_limit='64GB')
client = Client(cluster)

但如果数据量超出机器内存上限，这种方式依然会报错，所以还是优先优化计算逻辑。

三、额外优化建议 #

• 提前去重：如果Network存在重复的(NiuSup, NiuCust)记录，先执行Network = Network.drop_duplicates()，减少后续计算的数据量。
• 合理设置分区数：分区过多会增加临时文件数量和元数据开销，500万条数据设置10-20个分区是合理范围。

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