Python的分布式计算框架——Dask调度器简介

Dask是Python的分布式计算框架，它支持分布式的DataFrame，也就是pandas的DataFrame，二者接口完美兼容，但Dask是分布式计算的框架，可以支持内存无法装载的数据，进行计算，它也支持对一般的python程序进行分布式计算。是非常优秀的Python框架。本文主要介绍Dask的几种不同的调度器的使用。

Dask支持多种调度器，从单线程、多线程、多进程到本地分布式和集群分布式，各种调度器在不同情况下有不同的作用，本文来源于Dask官方文档的翻译，主要向大家介绍这五种调度器的使用情景和方式。最后提供了如何在不同情境下设置Dask调度器的方法。

• 一、本地线程（Local Threads）
• 二、本地进程（Local Process）
• 三、单线程（Single Thread）
• 四、Dask本地分布式（Dask Distributed (local)）
• 五、Dask集群分布式（Dask Distributed (Cluster)）
• 六、调度器配置

所有大型的Dask集合变量（例如Dask Array，Dask DataFrame和Dask Bag）以及细粒度的API（例如Delay和Future）都会生成任务图，其中图中的每个节点都是常规的Python函数，而节点之间的边缘是常规的Python对象，由一个任务创建为输出，并在另一任务中用作输入。 在Dask生成这些任务图之后，它需要在并行硬件上执行它们。这就是任务调度。Dask存在不同的任务调度，每个调度程序将使用一个任务图并计算得到相同的结果，但是它们的性能差别很大。

Dask有两类任务调度：

1. 单机任务调度：单机任务调度在本地进程或线程池上提供基本功能。该调度是默认提供的。尽管它只能在单台机器上使用并且无法弹性伸缩，但使用起来简单且门槛很低。
2. 分布式任务调度：分布式任务调度功能更复杂，有更好的性能，但设置起来也需要更多的精力。 它可以通过集群在本地运行或分布式执行。

对于不同的计算任务来说，使用不同的任务调度有更好的性能。本文档可帮助您了解如何在不同的任务调度之间进行选择和配置。

一、本地线程（Local Threads）

import daskdask.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对象，那么您可能要尝试以下基于进程的调度程序（我们目前建议在本地计算机上使用分布式调度程序）。

二、本地进程（Local Process）

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}

对于更复杂的工作负载，其中较大的中间结果可能被多个下游任务依赖，我们通常建议在本地计算机上使用分布式调度程序。 分布式调度程序对于处理较大的中间结果更为智能。

三、单线程（Single Thread）

import dask
dask.config.set(scheduler='synchronous')  # overwrite default with single-threaded scheduler

单线程同步调度器在本地线程中执行所有计算，而完全没有并行性。 这对于调试和概要分析特别有价值，而调试和概要分析在使用线程或进程时更加困难。

例如，当使用IPython或Jupyter笔记本时，使用并行Dask调度程序时，％debug，％pdb或％prun的技巧将无法正常工作（它们并非设计用于并行计算上下文中）。 但是，如果遇到异常并想进入调试器，则可能希望在单线程调度程序下重新运行计算，这些工具将在此正常运行。

四、Dask本地分布式（Dask Distributed (local)）

from dask.distributed import Client
client = Client() # or client = Client(processes=False)

Dask分布式调度器既可以在群集上运行，也可以在个人计算机上本地运行。 尽管名称为“ distributed”，但出于以下几个原因，它在本地计算机上通常是实用的：

1. 它提供对异步API的访问，例如Futures
2. 它提供了一个诊断仪表板（diagnostic dashboard），可以提供有关性能和进度的见解
3. 它可以在本地执行更复杂地数据处理，因此在需要多个流程的工作负载上，它比多进程调度器（multiprocess）更有效。

您可以在这篇博客中了解有关在单台计算机上使用Dask分布式调度程序的更多信息。

五、Dask集群分布式（Dask Distributed (Cluster)）

您也可以在分布式集群上运行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 ThreadPoolwith dask.config.set(pool=ThreadPool(4)):    ...with dask.config.set(num_workers=4):    ...