具有异步实时并行性的 Dask 分布式

Question

我正在阅读dask.distributed 上的文档，看起来我可以通过client.submit() 向分布式集群提交函数。

我有一个现有的函数some_func 异步获取单个文档（例如，文本文件），我想获取原始文档并获取所有不包含元音的单词并将其推回另一个数据库。此数据处理步骤正在阻塞。

假设有几百万个文档，分布式集群只有 10 个节点和 1 个可用进程（即一次只能处理 10 个文档），dask.distributed 将如何处理它需要的文档流处理？

这里是一些示例代码：

client = dask.distributed('tcp://1.2.3.4:8786')

def some_func():
    doc = retrieve_next_document_asynchronously() 
    client.submit(get_vowelless_words, doc)

def get_vowelless_words(doc):
    vowelless_words = process(doc)
    write_to_database(vowelless_words)

if __name__ == '__main__':
    for i in range(1000000):
        some_func()

由于一个文档的处理是阻塞的，并且集群只能同时处理 10 个文档，如果在集群忙的时候检索到 30 个其他文档会发生什么？我知道client.submit() 是异步的，它会返回一个并发的未来，但在这种情况下会发生什么？它是否会将文档保存在内存中，直到它有 1/10 个内核可用，并可能导致机器在有 1,000 个文档等待之后耗尽内存。

在这种情况下调度程序会做什么？先进先出？我是否应该以某种方式更改代码，以便在检索下一个文档之前等待核心可用？那怎么可能实现？

Answer 1

要在 dask 中使用队列，下面是在分布式集群中使用 dask 队列的修改示例（基于documentation）：

#!/usr/bin/env python

import distributed
from queue import Queue
from threading import Thread

client = distributed.Client('tcp://1.2.3.4:8786')
nprocs = len(client.ncores())

def increment(x):
    return x+1

def double(x):
    return 2*x

input_q = Queue(maxsize=nprocs)
remote_q = client.scatter(input_q)
remote_q.maxsize = nprocs
inc_q = client.map(increment, remote_q)
inc_q.maxsize = nprocs
double_q = client.map(double, inc_q)
double_q.maxsize = nprocs
result_q = client.gather(double_q)

def load_data(q):
    i = 0
    while True:
        q.put(i)
        i += 1

load_thread = Thread(target=load_data, args=(input_q,))
load_thread.start()

while True:
    size = result_q.qsize()
    item = result_q.get()
    print(item, size)

在这种情况下，我们明确限制每个队列的最大大小等于可用的分布式进程的数量。否则，while 循环将使集群过载。当然，您也可以将 maxsize 调整为可用进程数的倍数。对于像增量和加倍这样的简单函数，我发现maxsize = 10*nprocs 仍然是合理的，但这肯定会受到运行自定义函数所需时间的限制。

Answer 2

当您调用提交时，所有参数都被序列化并立即发送到调度程序。另一种方法是获取文档并在集群上处理它们（这假设文档对所有工作人员都是全局可见的）。

for fn in filenames:
    doc = client.submit(retrieve_doc, fn)
    process = client.submit(process_doc, doc)
    fire_and_forget(process)

如果文档仅在您的客户端计算机上可用并且您想限制流量，那么您可以考虑使用 dask Queues 或 as_completed 迭代器。