arrays - 多个图像意味着 dask.delayed 与 dask.array

Question

背景
我有一个列表，其中包含经过预处理并保存为 .npy 二进制文件的数千个图像堆栈(3D numpy 数组)的路径。

案例研究 我想计算所有图像的平均值，为了加快分析速度，我想并行处理。

方法使用 dask.delayed

# List with the file names
flist_img_to_filter

# I chunk the list of paths in sublists. The number of chunks correspond to 
# the number of cores used for the analysis
chunked_list
# Scatter the images sublists to be able to process in parallel
futures = client.scatter(chunked_list)

# Create dask processing graph
output = []
for future in futures:
    ImgMean = delayed(partial_image_mean)(future)
    output.append(ImgMean)
    ImgMean_all = delayed(sum)(output)
    ImgMean_all = ImgMean_all/len(futures)

 # Compute the graph
 ImgMean = ImgMean_all.compute()

方法使用 dask.arrays
修改自 Matthew Rocklin blog

imread = delayed(np.load, pure=True)  # Lazy version of imread
# Lazily evaluate imread on each path
lazy_values = [imread(img_path) for img_path in flist_img_to_filter]     

arrays = [da.from_delayed(lazy_value, dtype=np.uint16,shape=shape) for 
lazy_value in lazy_values]

# Stack all small Dask arrays into one
stack = da.stack(arrays, axis=0)

ImgMean = stack.mean(axis=0).compute()               

问题

1. 在 dask.delayed方法是否有必要预先分块列表？如果我分散原始列表，我将获得每个元素的 future 。有没有办法告诉 worker 处理它可以访问的 future ？
2. dask.arrays方法明显更慢并且内存使用量更高。这是使用 dask.arrays 的“坏方法”吗？
3. 有没有更好的方法来解决这个问题？

谢谢!

Answer 1

In the dask.delayed approach is it necessary to pre-chunk the list? If I scatter the original list I obtain a future for each element. Is there a way to tell a worker to process the futures it has access to?

简单的答案是否定的，从 Dask 版本 0.15.4 开始，没有非常健壮的方法来提交对“当前存在于该工作线程上的某种类型的所有任务”的计算。

但是，您可以使用 who_has 轻松询问调度程序哪些键存在于调度程序中。或 has_what客户端方法。

from dask.distributed import wait
import wait

futures = dask.persist(futures)
wait(futures)
client.who_has(futures)

The dask.arrays approach is significantly slower and with higher memory usage. Is this a 'bad way' to use dask.arrays?

你可能想玩 split_every= mean 的关键字功能否则rechunk在调用 mean 之前将图像分组在一起(可能类似于上面的操作)以进行并行/内存权衡。

Is there a better way to approach the issue?

你也可以试试 as_completed并在数据完成时计算运行手段。您必须从延迟切换到 futures为了这。