将单个大文件从客户端加载到 dask 工作人员答案

【问题标题】：Load a single large file from client to dask workers将单个大文件从客户端加载到 dask 工作人员
【发布时间】：2018-07-25 11:09:35
【问题描述】：

如何让 dask 中的所有其他工作节点都可以访问一个 8 GB 的大文件？我试过pd.read_csv() 和chunksize 和client.scatter，但这需要很长时间。我在 macOS 上运行它。

这是我的代码：

import time

import pandas as pd
import dask as dask
import dask.distributed as distributed
import dask.dataframe as dd
import dask.delayed as delayed
from dask.distributed import Client, progress


client = Client(IP:PORT)
print client
print client.scheduler_info()
f = []
chunksize = 10 ** 6

for chunk in pd.read_csv('file.csv', chunksize=chunksize):
    f_in = client.scatter(chunk)
    f.append(f_in)
print "read"

ddf = dd.from_delayed(f)        
ddf = ddf.groupby(['col1'])[['col2']].sum()

future = client.compute(ddf)
print future
progress(future)

result = client.gather(future)
print result

坚持下去。提前致谢！

【问题讨论】：

标签： python dask dask-distributed

【解决方案1】：

只要文件是 .csv 文件（未压缩），Dask 就会对其进行分块，不知道您为什么要尝试自己对其进行分块。做吧：

将 dask.dataframe 导入为 dd df = dd.read_csv('data*.csv')

【讨论】：

这行不通，因为工人无权访问原始数据文件。

【解决方案2】：

在您的工作流程中，您在本地加载 CSV 数据，解析成数据帧，然后将这些数据帧的序列化版本传输给工作人员，一次一个。

一些可能的解决方案：

将文件复制给每个工作人员（这在空间方面很浪费），或者将其放在他们都可以看到的某个位置，例如共享文件系统或云存储
使用client.upload_file，它并不是真正为大负载设计的，而且还会复制到每个工作人员
像以前一样使用dask.bytes.read_bytes读取串行数据块，并将这些数据持久化给工作人员，因此至少您不会遭受序列化成本，并且工作人员之间共享解析工作。

【讨论】：

我不确定 client.upload_file 是否会将 csv 文件作为输入。