export-to-csv - 我应该如何使用 dask.dataframe 高效地编写多个 CSV 文件？

Question

这里是我正在做的事情的总结:

首先，我通过普通的多处理和 pandas 包来实现:

第 1 步。获取我要读取的文件名列表

import os    
files = os.listdir(DATA_PATH + product)

第 2 步. 遍历列表

from multiprocessing import Pool
import pandas as pd    

def readAndWriteCsvFiles(file):
    ### Step 2.1 read csv file into dataframe 
    data = pd.read_csv(DATA_PATH + product + "/" + file, parse_dates=True, infer_datetime_format=False)

    ### Step 2.2 do some calculation
    ### .......

    ### Step 2.3 write the dataframe to csv to another folder
    data.to_csv("another folder/"+file)

if __name__ == '__main__':
    cl = Pool(4)
    cl.map(readAndWriteCsvFiles, files, chunksize=1)
    cl.close()
    cl.join()  

代码工作正常，但速度很慢。

完成任务大约需要 1000 秒。

与使用 library(parallel) 和 parSapply 函数的 R 程序比较。

R 程序只需要大约 160 秒。

然后我尝试使用 dask.delayed 和 dask.dataframe 使用以下代码:

第 1 步。获取我要读取的文件名列表

import os    
files = os.listdir(DATA_PATH + product)

第 2 步. 遍历列表

from dask.delayed import delayed
import dask.dataframe as dd
from dask import compute

def readAndWriteCsvFiles(file):
    ### Step 2.1 read csv file into dataframe 
    data = dd.read_csv(DATA_PATH + product + "/" + file, parse_dates=True, infer_datetime_format=False, assume_missing=True)

    ### Step 2.2 do some calculation
    ### .......

    ### Step 2.3 write the dataframe to csv to another folder
    data.to_csv(filename="another folder/*", name_function=lambda x: file)

compute([delayed(readAndWriteCsvFiles)(file) for file in files])

这一次，我发现如果我在 dask 代码和 pandas 代码中注释掉步骤 2.3，dask 的运行速度会比普通的 pandas 和多处理快得多。

但是如果我调用 to_csv 方法，那么 dask 和 pandas 一样慢。

有什么解决办法吗？

谢谢

Answer 1

读取和写入 CSV 文件通常受 GIL 约束。您可能想尝试使用进程而不是线程进行并行化(dask delayed 的默认设置)。

您可以通过将 scheduler='processes' 关键字添加到您的计算调用中来实现这一点。

compute([delayed(readAndWriteCsvFiles)(file) for file in files], scheduler='processes')

参见 scheduling documentation了解更多信息

另请注意，您在这里使用的不是 dask.dataframe，而是 dask.delayed。