Dask调度器简介

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

[TOC]

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

<center>![](http://www.datalearner.com/resources/blog_images/398f269d-474a-4efb-818e-768db57e9e85.png)</center>

Dask有两类任务调度：

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

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

#### 一、本地线程（Local Threads）

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

#### 二、本地进程（Local Process）

```python
import dask.multiprocessing
dask.config.set(scheduler='processes')  # overwrite default with multiprocessing scheduler
```

多进程调度器使用本地multiprocessing.Pool执行计算。它使用起来很轻巧，不需要任何设置。 每个任务及其所有依赖项都被传送到本地进程执行，然后其结果被传送回主进程。 这意味着它能够绕过GIL的问题，甚至在纯Python代码主导的计算（例如处理字符串，字典和列表的计算）上提供并行性。

但是，将数据移至远程进程并返回会带来性能损失，特别是在进程之间传输的数据较大时。 当工作流相对线性，因此不涉及大量的任务间数据传输以及输入和输出都较小（如文件名和计数）时，多处理调度程序是一个很好的选择。

这在基本数据摄取工作负载中很常见，例如在Dask Bag中是常见的工作负载，其中默认为多进程调度处理：

```python
>>> import dask.bag as db
>>> db.read_text('*.json').map(json.loads).pluck('name').frequencies().compute()
{'alice': 100, 'bob': 200, 'charlie': 300}
```

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

#### 三、单线程（Single Thread）

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

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

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

#### 四、Dask本地分布式（Dask Distributed (local)）
```python
from dask.distributed import Client
client = Client()
# or
client = Client(processes=False)
```
Dask分布式调度器既可以在群集上运行，也可以在个人计算机上本地运行。 尽管名称为“ distributed”，但出于以下几个原因，它在本地计算机上通常是实用的：

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

您可以在[这篇博客](https://www.datalearner.com/blog/1051585635910844)中了解有关在单台计算机上使用Dask分布式调度程序的更多信息。

#### 五、Dask集群分布式（Dask Distributed (Cluster)）
您也可以在分布式集群上运行Dask。有多种方法可根据您的集群进行设置。我们建议参考安装文档以获取更多信息。

#### 六、调度器配置
您可以使用dask.config.set（scheduler ...）命令配置全局默认调度器。

全局设置：
```python
dask.config.set(scheduler='threads')
x.compute()
```

根据情景设置：
```python
with dask.config.set(scheduler='threads'):
    x.compute()
```

在具体某一个程序中使用：
```python
x.compute(scheduler='threads')
```

此外，某些调度器还支持其他关键字参数。 例如，基于池的单机调度器允许您提供自定义池或指定所需的工作线程数：

```python
from multiprocessing.pool import ThreadPool
with dask.config.set(pool=ThreadPool(4)):
    ...

with dask.config.set(num_workers=4):
    ...
```

Dask调度器简介

欢迎大家关注DataLearner官方微信，接受最新的AI技术推送

相关博客

最热博客