使用Dask合并大数据库时遭遇磁盘空间不足问题求助

我需要处理两个大型数据集：Network（800MB）和SecondOrder（33GB），执行以下合并操作时，无论全程延迟计算后用.to_csv保存，还是调用.compute()，都会触发OS Error: No Space left on Device错误，推测是Dask生成的临时文件占满了磁盘空间。请问有没有解决办法（比如强制Dask不生成临时文件）？

操作代码如下：

NetworkDD = dd.from_pandas(Network, npartitions=Network['NiuSup'].nunique())

NodesSharingSupplier = dd.merge(NetworkDD, NetworkDD, on='NiuSup').query('NiuCust_x != NiuCust_y')

### WORKS UNTIL HERE

NodesSharingSupplier=NodesSharingSupplier.drop('NiuSup', axis=1)
NodesSharingSupplier=NodesSharingSupplier.drop_duplicates()


NodesSharingSupplier=NodesSharingSupplier.rename(columns={"NiuCust_x": "NiuSup", "NiuCust_y": "NiuCust"})

NodesSharingSupplier.to_csv("NodesSharingSupplier.csv")

NodesSharingSupplier=NodesSharingSupplier.drop('NiuSup', axis=1)
NodesSharingSupplier=NodesSharingSupplier.drop_duplicates()

NodesSharingSupplier=NodesSharingSupplier.rename(columns={"NiuCust_x": "NiuSup", "NiuCust_y": "NiuCust"})

SecondOrder=pd.read_csv("/home/francesco.serti/SecondOrder_line55.csv")
SecondOrderDD = dd.from_pandas(SecondOrder, npartitions=SecondOrder['NiuSup'].nunique())

SecondOrderDD_all = SecondOrderDD.merge(NodesSharingSupplier, on=['NiuCust','NiuSup'], how='left', indicator=True)
SecondOrderDD=SecondOrderDD_all.loc[SecondOrderDD_all._merge=='left_only',SecondOrderDD_all.columns!='_merge']

del SecondOrder
del NodesSharingSupplier

##HERE IF I DO .compute() an OS Error: No Space left on Device
SecondOrderDD.to_csv('/home/francesco.serti//SecondOrderDD_line75.csv', single_file=True)

Dask在执行shuffle类操作（比如merge、drop_duplicates、groupby等需要跨分区交换数据的操作）时，会默认在系统临时目录（通常是/tmp）生成大量临时文件存储中间结果。另外，当使用single_file=True时，Dask需要先将所有分区数据写入临时文件，再合并成单个文件，这会进一步占用磁盘空间，当磁盘剩余不足时就会触发空间不足错误。

1. 更换Dask临时文件存储路径 #

将临时文件转移到空间充足的磁盘目录，可通过代码或环境变量设置：

import dask
# 替换为你有充足空间的路径
dask.config.set(temporary_directory='/path/to/large-storage/dask-tmp')

或者在启动Python前设置环境变量：

export DASK_TEMPORARY_DIRECTORY=/path/to/large-storage/dask-tmp

2. 优化中间步骤，减少不必要的磁盘写入 #

• 若不需要保留NodesSharingSupplier.csv文件，直接删除NodesSharingSupplier.to_csv("NodesSharingSupplier.csv")这一步，避免额外的磁盘写入开销。
• 合并重复的drop/rename操作，减少中间结果的生成：

# 合并操作链，减少中间变量的磁盘占用
NodesSharingSupplier = (
    NodesSharingSupplier
    .drop('NiuSup', axis=1)
    .drop_duplicates()
    .rename(columns={"NiuCust_x": "NiuSup", "NiuCust_y": "NiuCust"})
)

3. 优化分区策略，降低shuffle数据量 #

你当前用npartitions=df['NiuSup'].nunique()设置分区数，可能导致分区过多，增加shuffle的磁盘开销。建议按数据大小设置分区（每个分区100-200MB为宜）：

# 直接用Dask读取大文件，自动优化分区
SecondOrderDD = dd.read_csv("/home/francesco.serti/SecondOrder_line55.csv", blocksize='150MB')

# 对Network数据集设置合理分区数
NetworkDD = dd.from_pandas(Network, npartitions=5)  # 800MB数据设为5个分区，每个约160MB

4. 避免生成单个大文件（若业务允许） #

如果不需要输出单个CSV文件，去掉single_file=True，让Dask生成多个分区文件，跳过合并临时文件的步骤，大幅减少磁盘占用：

# 生成多个分区文件，无需合并
SecondOrderDD.to_csv('/home/francesco.serti/SecondOrderDD_line75_*.csv')

后续若需单个文件，可通过命令行工具合并（效率更高）：

cat /home/francesco.serti/SecondOrderDD_line75_*.csv > /home/francesco.serti/SecondOrderDD_line75.csv

5. 手动清理Dask临时文件 #

Dask不会自动清理所有临时文件，可手动删除系统临时目录下的Dask文件：

rm -rf /tmp/dask-*

或者通过代码自动清理临时文件：

from dask.utils import tmpfile

with tmpfile.tempdir():
    # 在临时目录上下文内执行操作，结束后自动清理
    SecondOrderDD.to_csv('/home/francesco.serti/SecondOrderDD_line75.csv', single_file=True)

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