将150万张图像加载到dask群集的最快方法

Question

我正在尝试将150万张图像作为快闪阵列保存到快闪集群中，然后获得一些摘要统计信息。我正在追踪image processing tutorial from @mrocklin's blog，并已将我的脚本编辑为最小程度可复制的示例：

import time
import dask
import dask.array as da

import numpy as np
from distributed import Client

client = Client()

def get_imgs(num_imgs):
    def get():
        arr = np.random.randint(2000, size=(3, 120, 120)).flatten()
        return arr

    delayed_get = dask.delayed(get)

    return [da.from_delayed(delayed_get(), shape=(3 * 120 * 120,), dtype=np.uint16) for num in range(num_imgs)]

imgs = get_imgs(1500000)
imgs = da.stack(imgs, axis=0)
client.persist(imgs)

persist步骤导致我的jupyter进程崩溃。是否因为persist步骤导致对集合中的每个对象执行一堆操作，并且集合太大而无法容纳在内存中？所以我改用scatter：

start = time.time()
imgs_future = client.scatter(imgs, broadcast=True)
print(time.time() - start)

但是jupyter进程崩溃，或者与调度程序的网络连接丢失。

所以我尝试分解scatter步骤：

st = time.time()

chunk_size = 50000
chunk_num = 0
chunk_futures = []
start = 0
end = start + chunk_size
is_last_chunk = False

for dataset in client.list_datasets():
    client.unpublish_dataset(dataset)

while True:
    cst = time.time()
    chunk = imgs[start:end]
    cst1 = time.time()

    if start == 0:
        print('loaded chunk in', cst1 - cst)

    if len(chunk) == 0:
        break

    chunk_future = client.scatter(chunk)
    chunk_futures.append(chunk_future)
    dataset_name = "chunk_{}".format(chunk_num)
    client.publish_dataset(**{dataset_name: chunk_future})

    if start == 0:
        print('submitted chunk in', time.time() - cst1)
    start = end

    if is_last_chunk:
        break

    chunk_num += 1
    end = start + chunk_size
    if end > len(image_paths_to_submit):
        is_last_chunk = True
        end = len(image_paths_to_submit)

    if start == end:
        break

    if chunk_num % 5 == 0:
        print('chunk_num', chunk_num, 'start', start)

print('completed in', time.time() - st)

但是这种方法也会导致连接丢失。建议以异步方式将大型图像数据集持久存储在群集中的方法是什么？

我看过delayed best practices，然后突然跳出来的是我可能正在使用太多任务？因此，也许我需要在每个get()调用中进行更多批处理。

Answer 1

是因为持久步骤导致对集合中的每个对象执行一堆操作，并且集合太大而无法容纳在内存中？

找出这种情况的最佳方法是使用Dask的仪表板。https://docs.dask.org/en/latest/diagnostics-distributed.html#dashboard

我正在关注@mrocklin博客中的图像处理教程

那个帖子有些旧。您可能还想看看以下最新帖子：https://blog.dask.org/2019/06/20/load-image-data

我查看了延迟的最佳实践，而让我跳出来的是我可能使用了太多任务？因此，也许我需要在每个get（）调用中进行更多批处理。

是的，这可能是个问题。如果您可以减少任务数量，那就太好了。