Dask调度器简介
6,891 阅读
Dask支持多种调度器，从单线程、多线程、多进程到本地分布式和集群分布式，各种调度器在不同情况下有不同的作用，本文来源于Dask官方文档的翻译，主要向大家介绍这五种调度器的使用情景和方式。最后提供了如何在不同情境下设置Dask调度器的方法。
加载中...
加载中...
所有大型的Dask集合变量（例如Dask Array，Dask DataFrame和Dask Bag）以及细粒度的API（例如Delay和Future）都会生成任务图，其中图中的每个节点都是常规的Python函数，而节点之间的边缘是常规的Python对象，由一个任务创建为输出，并在另一任务中用作输入。 在Dask生成这些任务图之后，它需要在并行硬件上执行它们。这就是任务调度。Dask存在不同的任务调度，每个调度程序将使用一个任务图并计算得到相同的结果，但是它们的性能差别很大。
Dask有两类任务调度：
对于不同的计算任务来说，使用不同的任务调度有更好的性能。本文档可帮助您了解如何在不同的任务调度之间进行选择和配置。
import dask
dask.config.set(scheduler='threads') # overwrite default with threaded scheduler
线程调度程序使用本地multiprocessing.pool.ThreadPool执行计算。它是轻量级的，不需要任何设置。它引入了很少的任务开销（每个任务大约50us），并且由于所有事情都在同一进程中发生，因此在任务之间传输数据不会产生任何成本。 但是，由于Python的全局解释器锁定（GIL），此调度程序仅在您的计算由非Python代码主导时才能提供并行性，如NumPy数组、Pandas DataFrames或者是任何其他基于C / C ++ / Cython的项目。
线程调度器是Dask Array，Dask DataFrame和Dask Delayed的默认选择。但是，如果你处理的是如字符串，字典或列表之类的纯Python对象，那么您可能要尝试以下基于进程的调度程序（我们目前建议在本地计算机上使用分布式调度程序）。
import dask.multiprocessing
dask.config.set(scheduler='processes') # overwrite default with multiprocessing scheduler
多进程调度器使用本地multiprocessing.Pool执行计算。它使用起来很轻巧，不需要任何设置。 每个任务及其所有依赖项都被传送到本地进程执行，然后其结果被传送回主进程。 这意味着它能够绕过GIL的问题，甚至在纯Python代码主导的计算（例如处理字符串，字典和列表的计算）上提供并行性。
但是，将数据移至远程进程并返回会带来性能损失，特别是在进程之间传输的数据较大时。 当工作流相对线性，因此不涉及大量的任务间数据传输以及输入和输出都较小（如文件名和计数）时，多处理调度程序是一个很好的选择。
这在基本数据摄取工作负载中很常见，例如在Dask Bag中是常见的工作负载，其中默认为多进程调度处理：
>>> import dask.bag as db
>>> db.read_text('*.json').map(json.loads).pluck('name').frequencies().compute()
{'alice': 100, 'bob': 200, 'charlie': 300}
对于更复杂的工作负载，其中较大的中间结果可能被多个下游任务依赖，我们通常建议在本地计算机上使用分布式调度程序。 分布式调度程序对于处理较大的中间结果更为智能。
import dask
dask.config.set(scheduler='synchronous') # overwrite default with single-threaded scheduler
单线程同步调度器在本地线程中执行所有计算，而完全没有并行性。 这对于调试和概要分析特别有价值，而调试和概要分析在使用线程或进程时更加困难。
例如，当使用IPython或Jupyter笔记本时，使用并行Dask调度程序时，％debug，％pdb或％prun的技巧将无法正常工作（它们并非设计用于并行计算上下文中）。 但是，如果遇到异常并想进入调试器，则可能希望在单线程调度程序下重新运行计算，这些工具将在此正常运行。
from dask.distributed import Client
client = Client()
# or
client = Client(processes=False)
Dask分布式调度器既可以在群集上运行，也可以在个人计算机上本地运行。 尽管名称为“ distributed”，但出于以下几个原因，它在本地计算机上通常是实用的：
您可以在这篇博客中了解有关在单台计算机上使用Dask分布式调度程序的更多信息。
您也可以在分布式集群上运行Dask。有多种方法可根据您的集群进行设置。我们建议参考安装文档以获取更多信息。
您可以使用dask.config.set（scheduler ...）命令配置全局默认调度器。
全局设置：
dask.config.set(scheduler='threads')
x.compute()
根据情景设置：
with dask.config.set(scheduler='threads'):
x.compute()
在具体某一个程序中使用：
x.compute(scheduler='threads')
此外，某些调度器还支持其他关键字参数。 例如，基于池的单机调度器允许您提供自定义池或指定所需的工作线程数：
from multiprocessing.pool import ThreadPool
with dask.config.set(pool=ThreadPool(4)):
...
with dask.config.set(num_workers=4):
...