python - 避免同时读取 dask 数组的多个文件

Question

从库中，我得到一个读取文件并返回 numpy 数组的函数。

我想构建一个包含来自多个文件的多个 block 的 Dask 数组。

每个 block 都是在文件上调用函数的结果。

当我要求Dask计算时，Dask会要求函数同时从硬盘读取多个文件吗？

如果是这种情况，如何避免？我的计算机没有并行文件系统。

示例:

import numpy as np
import dask.array as da
import dask

# Make test data
n = 2
m = 3
x = np.arange(n * m, dtype=np.int).reshape(n, m)
np.save('0.npy', x)
np.save('1.npy', x)

# np.load is a function that reads a file 
# and returns a numpy array.

# Build delayed
y = [dask.delayed(np.load)('%d.npy' % i)
     for i in range(2)]

# Build individual Dask arrays.
# I can get the shape of each numpy array without 
# reading the whole file.
z = [da.from_delayed(a, (n, m), np.int) for a in y]

# Combine the dask arrays
w = da.vstack(z)

print(w.compute())

Answer 1

您可以使用分布式 lock原语 - 以便您的加载器函数执行获取-读取-释放。

read_lock = distributed.Lock('numpy-read')

@dask.delayed
def load_numpy(lock, fn):
    lock.acquire()
    out = np.load(fn)
    lock.release()
    return out

y = [load_numpy(lock, '%d.npy' % i) for i in range(2)]

此外，da.from_array 接受锁，因此您可以从直接提供锁的延迟函数np.load 创建单独的数组。

或者，您可以分配一个单位 resource到工作线程(具有多个线程)，然后根据每个文件读取任务一个单元的要求进行计算(或持久化)，如链接文档中的示例所示。

对评论的回应:问题中没有指定to_hdf，我不确定为什么现在会受到质疑；但是，您可以将 da.store(compute=False) 与 h5py.File 结合使用，然后指定调用compute 时要使用的资源。请注意，这不会将数据具体化到内存中。