python - 将本地文件从客户端加载到 dask 分布式集群

Question

有点初学者的问题，但我找不到相关的答案..

基本上我的数据 (7gb) 位于我的本地机器上。我有在本地网络上运行的分布式集群。我怎样才能把这个文件放到集群上？

通常的 dd.read_csv() 或 read_parquet() 会失败，因为工作人员无法在自己的环境中找到文件。

我是否需要手动将文件传输到集群中的每个节点？

注意:由于管理员限制，我只能使用 SFTP...

Answer 1

两种选择

网络文件系统

正如评论中所建议的，使用普通文件系统解决方案，有多种方法可以让集群中的其他机器访问您的本地文件。如果您可以访问，这是一个不错的选择。

本地加载和分散

如果这不起作用，那么您始终可以在本地加载数据并将其分散到集群中的各个工作程序。如果您的文件大于单台计算机的内存，那么您可能需要一 block 一 block 地执行此操作。

单程

如果一切都适合内存，那么我会正常加载数据，然后将其分散给工作人员。如果需要，您可以随后将其拆分并传播给其他工作人员:

import pandas
import dask.dataframe as dd
from dask.distributed import Client

client = Client('scheduler-address:8786')

df = pd.read_csv('myfile.csv')
future = client.scatter(df)  # send dataframe to one worker
ddf = dd.from_delayed([future], meta=df)  # build dask.dataframe on remote data
ddf = ddf.repartition(npartitions=20).persist()  # split
client.rebalance(ddf)  # spread around all of your workers

多位

如果你有多个小文件，那么你可以迭代加载和分散，也许在一个 for 循环中，然后从许多 future 制作一个 dask.dataframe

futures = []
for fn in filenames:
    df = pd.read_csv(fn)
    future = client.scatter(df)
    futures.append(future)

ddf = dd.from_delayed(futures, meta=df)

在这种情况下，您可以跳过重新分区和重新平衡步骤

如果您有一个大文件，那么您可能需要自己对它进行一些拆分，或者使用 pd.read_csv(..., chunksize=...)