python - 如何将 dask 数据帧保存到与 dask 调度程序/工作人员相同的机器上的 Parquet ？

Question

我正在尝试通过 Dask Dataframe 保存到与 dask 调度程序/工作人员所在的同一台计算机上的 Parquet 。但是，我在这期间遇到了麻烦。

我的 Dask 设置:我的 python 脚本在本地计算机(笔记本电脑 16 GB RAM)上执行，但该脚本为在远程计算机(具有 400 GB RAM 用于并行计算的服务器)上运行的 Dask 调度程序创建 Dask 客户端。 Dask 调度程序和工作线程都位于同一服务器上，因此它们都共享相同的文件系统，可供本地使用。由于我团队的所有成员都使用此远程 Dask 调度程序，因此我们正在处理的文件也位于同一服务器上，从而为所有成员提供通过同一 Dask 集群对所有文件的公共(public)访问。

我已经尝试过:

# This saves the parquet files in a folder on my local machine.
ddf.to_parquet(
    '/scratch/dataset_no_dalayed', compression='brotli').compute()

# This delayed call of `ddf.to_parquet` saves the Dask Dataframe chucks
# into individual parquet files (i.e. parts) in the given folder.
# However, I want to persist the Dask dataframe in my workflow, but this
# fails as seen below.
dask.delayed(ddf.to_parquet)(
    '/scratch/dataset_dalayed', compression='brotli').compute()

# If the Dask dataframe is persisted, the `to_parquet` fails with
# a "KilledWorker" error!
ddf = client.persist(ddf)
dask.delayed(ddf.to_parquet)(
    '/scratch/dataset_persist/', compression='brotli').compute()

# In the example below, I can NOT save the Dask dataframe.
# Because the delayed function makes the Dask dataframe
# to a Pandas dataframe on runtime. And this fails as the path is a
# folder and not at file as Pandas requires!
@dask.delayed
def save(new_ddf):
    new_ddf.to_parquet('/scratch/dataset_function/', compression='brotli')

save(ddf).compute()

如何正确执行此操作？

Answer 1

通常要将 dask 数据框保存为 parquet 数据集，人们会执行以下操作:

df.to_parquet(...)

从您的问题来看，您的工作人员可能并非都有权访问 NFS 或 S3 等共享文件系统。如果是这种情况，并且您存储到本地驱动器，那么您的数据将分散在各种计算机上，而没有明显的方法将它们收集在一起。原则上，我鼓励您避免这种情况，并投资共享文件系统。它们在进行分布式计算时非常有帮助。

如果你做不到这一点，那么我个人可能会并行写入本地驱动器，然后将它们传输回一台机器。

如果您的数据集足够小，那么您还可以调用 .compute 返回本地 Pandas 数据帧，然后使用 Pandas 写入

df.compute().to_parquet(...)