使用多进程调度程序将Dask阵列并行写入HDF5失败

Question

Dask是一个文档良好的可伸缩库，用于并行处理，使用基于图形的工作流在编写许多具有固有并行性的应用程序时非常有用。然而，当并行写入hdf5文件时，这是相当困难的，特别是在使用多进程调度器时。如果使用默认的多线程调度器，则以下代码可以正常工作，

x = da.arange(25000, chunks = (1000,))
da.to_hdf5('hdfstore.h5', '/store', x)

但如果您全局设置多进程调度器：

dask.set_options(get=dask.multiprocessing.get)

然后再次运行代码，

TypeError: can't pickle _thread.lock objects

多线程调度器还可以，但它在读取单个大型csv文件并将其转换为hdf5文件时速度太慢。使用多进程调度器，它的速度很快，能够在最大负载下使用所有CPU，但hdf写入失败并出现上述错误(我认为hdf5文件支持使用h5py mpi驱动程序进行同步写入访问)。如果您直接这样做

x.compute()

一切都很好，但它将整个数据加载到内存中，也就是说，它在处理大型数组和文件时并不是很好。有没有人遇到过这样的场景？请务必分享宝贵的建议..

conda虚拟环境上的Dask版本'0.13.0‘

Answer 1

我认为这段代码的问题在于，当您使用多进程调度程序时，它会同时将不同的数据块写入同一个hdf5文件。

据我所知，HDF5格式支持HDF5 SWMR。因此，当一个Python进程有权写入1个块时，它通过锁定机制阻止其他进程同时写入。

如果你想即时编写，也许this能帮上忙。