dask - 使用 Dask 将大型 csv 拼花 - OOM

Question

我有 7 个 csv 文件，每个 8 GB，需要转换为 parquet。

内存使用量达到 100 GB，我不得不终止它。我也尝试使用 Distributed Dask。内存限制为 12 GB，但长时间没有输出。供引用。我曾经使用 Chunking + Prod consumer 的传统 pandas --> 能够在 30 分钟内转换Dask 处理我缺少什么？

def ProcessChunk(df,...):
    df.to_parquet()     

for factfile in fArrFileList:
   df = dd.read_csv(factfile, blocksize="100MB",
                 dtype=fColTypes, header=None, sep='|',names=fCSVCols)
   result = ProcessChunk(df,output_parquet_file, chunksize, fPQ_Schema, fCSVCols, fColTypes)

Answer 1

感谢大家的建议。 map_partitions 工作。

df = dd.read_csv(filename, blocksize="500MB",
                     dtype=fColTypes, header=None, sep='|',names=fCSVCols)
df.map_partitions(DoWork,output_parquet_file, chunksize, Schema, CSVCols, fColTypes).compute(num_workers=2)

但是对于 Dask 分布式本地集群，同样的方法效果不佳。当 csv 大小 < 100 MB 时，它在本地集群模式下工作。