楔子

现在相信你已经对DAG的工作原理有了基本的理解，那么下面来看看Dask如何使用DAG来创建健壮的、可扩展的workload(控制器)。

下面我们要完成两件事：使用Dask的DataFrame API来分析结构化数据集；研究一些有用的诊断工具，并使用low-level Delayed API来创建一个简单的自定义任务图。

但是我们需要先安装Dask，直接pip install dask -i https://pypi.tuna.tsinghua.edu.cn/simple即可，Dask是Python的一个第三方库。

import sys
import dask

print(dask.__version__)  # 2.28.0
print(sys.version[: 5])  # 3.8.1

但是我们需要先安装Dask，直接pip install dask -i https://pypi.tuna.tsinghua.edu.cn/simple即可，Dask是Python的一个第三方库。

你好Dask：看一下你的DataFrame API

任何数据科学项目的一个基本步骤都是对数据集执行探索性分析，在探索性分析期间，你需要检查数据是否存在缺失值、异常值和任何其他数据质量问题。如果出现了脏数据，需要对其进行清洗，以确保对数据所做的任何结论不会受到错误或异常数据的影响。在使用Dask DataFrame API的第一个示例中，我们将逐步读取数据文件，并扫描数据以查找缺失值，并删除由于丢失太多数据或者对分析没有帮助的列。

检测Dask对象的元数据(metadata)

下面我们首先需要读取数据集，关于数据集，书中使用的是上一篇博客中说的NYC停车罚单数据，但是那个数据集已经没有了。但是我们可以在kaggle上下载其它的数据集，这里我下载的数据集是：https://www.kaggle.com/kenshoresearch/kensho-derived-wikimedia-data里面的item.csv。

没有特殊说明，我们的代码都是在jupyter notebook上面运行。

import dask.dataframe as dd

df = dd.read_csv(r"C:\Users\satori\Desktop\item\item.csv")
df

如果你是一个经验丰富的pandas使用者，那么你会发现上面的代码非常的熟悉，因为它们在语法上是等价的。但是结果却不一定是你熟悉的，pandas的DataFrame对象在打印的时候会尝试将数据的开头和结尾显示出来，最上方是列名，左侧是索引。但是我们看到Dask DataFrame在打印的时候会显示元数据，列名在顶部，而下方则是每个列各自的数据类型，因为Dask会非常努力、并且智能地从数据中推断数据类型，就像pandas所做的那样。但是Dask在这方面的能力也是会受到限制的，Dask是用来处理不能被单机读取的中大型数据集的，而pandas是完全基于内存操作，所以它可以轻松而快速地扫描整个数据集，从而判断每个列的最佳数据类型。另一方面，Dask必须能够很好的处理分散在分布式文件系统中多个机器的数据集，因此Dask DataFrame使用随机抽样的方法从一小部分数据样本中分析和推断数据类型。但是这样有一个问题，如果数百万或数十亿行中只有一个异常行，那么这个异常行就不太可能被随机挑选出来，这将导致Dask选择不兼容的数据类型，可能会导致在后面的执行计算中出现错误。因此避免这种情况的最佳实践是显式地设置数据类型，而不是依赖于Dask的推断。

事实上这一点很好理解，因为pandas是将数据全部读取到内存中，即便是分块，它最终也是可以选择一个合适的类型。但是对于Dask而言，显然是没有办法这么做的，因为数据量就已经决定了光靠随机采样是达不到百分之百的准确率。

或者更好的做法是使用能够表示数据类型的二进制文件(比如：Parquet)，类型是什么在文件中进行体现，这样就完全可以避免类型推断带来的问题。我们将在后续系列来讨论这个问题，目前我们就让Dask自己推断数据类型。

显然我们来看看Dask DataFrame的输出，因为这和pandas DataFrame的输出还是有很大差别的。Dask DataFrame显示的是元数据信息，它告诉了我们Dask的任务调度器在处理该文件时是如何对任务进行分解的。

显然我们来看看Dask DataFrame的输出，因为这和pandas DataFrame的输出还是有很大差别的。Dask DataFrame显示的是元数据信息，它告诉了我们Dask的任务调度器在处理该文件时是如何对任务进行分解的。

npartitions显示了将DataFrame分隔成多少个分区，这里是65个，由于该文件的大小是3.85GB，在65个分区中，每个分区的大小大概是60.65MB。这意味着Dask不是一次性将文件加载到内存中，而是每个Dask工作线程一次处理一个60.65MB的文件块。而且dd.read_csv在读取这个大文件时，几乎是瞬间完成的，这也说明了该函数在读取csv文件时会执行懒加载。

Dask将文件分成可以独立处理的多个小块，而不是马上将整个文件读取到内存中，而这些块就叫做分区。而Dask DataFrame中的每一个分区，都相当于是一个较小的pandas DataFrame。

所以Dask是将大文件分割成多个分区，一次处理一个分区。

图中的Dask DataFrame是由两个分区组成，因此单个Dask DataFrame由两个较小的pandas DataFrame组成，每个分区都可以加载到内存中进行处理。工作节点先拾取分区1进行处理，然后将结果保存在临时存储空间中。然后拾取分区2进行处理，也将结果保存在一个临时空间中，然后将两个结果合并并返回给我们。因为工作节点一次可以处理较小的数据片段，所以可以将工作分配个多个机器，或者在本地的情况下，也可以在非常大的数据集上继续工作，而不会导致内存不足产生的错误(内存溢出)。

注意df返回的元数据，我们还有一个信息没有说，就是最下面的65 tasks，这表示Dask DataFrame由65个任务组成。这表示Dask创建了一个有65个节点的有向无环图来处理数据。而我们恰好也有65个分区，表示每个分区里面有一个任务，注意：分区数和任务数不一定是一致的，因为一个分区可以对应多个任务。我们这里是65个分区，所以如果有65个worker的话，那么可以同时处理整个文件，但如果只有一个worker，那么Dask将依次循环遍历每一个分区。现在让我们尝试计算一下，整个文件中每个列的缺失值。

# 可以看到使用的api和pandas的DataFrame基本是一致的
missing_value = df.isnull().sum()
"""
Dask Series Structure:
npartitions=1
en_description    int64
item_id             ...
dtype: int64
Dask Name: dataframe-sum-agg, 196 tasks
"""

虽然在计算的方法上和pandas是一致的，但是结果和之前一样，返回的Series对象并没有给出我们期望的输出。返回的不是缺失值的统计值，而是一些关于预期结果的元数据信息。而且通过输出信息我们发现返回的结果看起来像是一系列int64，但是实际数据在哪里呢？其实Dask还没有进行处理，因为采用的是惰性计算。这意味着Dask实际上做的是准备一个DAG，然后存储在missing_value变量中，在显式执行任务图之前，不会计算数据。这种行为使得快速构建复杂任务图成为可能，而不必等待每个中间步骤完成，通过返回信息的最后一行我们知道此时的任务数量已经增加到了196。Dask从DAG中获取了前65个任务，这些任务用于读取数据文件创建DataFrame，然后DataFrame又添加了131个任务来检查null值以及计算sum，最终将所有部分收集到一个单一的Series对象中并返回答案。

missing_count = ((missing_value / df.index.size) * 100)
missing_count
"""
Dask Series Structure:
npartitions=1
en_description    float64
item_id               ...
dtype: float64
Dask Name: mul, 329 tasks
"""

在执行计算之前，我们要求Dask将缺失值的数量转成百分比，显然要除以DataFrame的总行数，再乘以100。注意：任务数量增加的同时，返回的Series对象的数据类型也从int64变成了float64。这是因为触发操作的结果不是整数，因此Dask自动将结果转换为浮点数，正如Dask尝试从文件中推断数据类型一样，它也会尝试推断某个操作如何影响输出的数据类型。由于我们已经像DAG中添加了一个用于两个数相除的操作，Dask推断我们可能会从整数移动到浮点数，并相应的改变元数据。

用compute方法运行计算

现在我们准备运行并生成结果了。

from dask.diagnostics import ProgressBar  # 绘制进度条
# 在计算的时候会自动显示进度条
with ProgressBar():
    missing_count_pct = missing_count.compute()

missing_count_pct

结果得到的是一个pandas的Series对象，也和我们平时使用pandas得到的结果也是一样的，我们看到总共花费了37.7s。所以当你想要得到计算结果时，需要调用compute方法，这会告诉Dask我们需要你开始真正地执行了。我们看到这个过程类似于Spark中的RDD，每一步的操作都是一个懒加载(transorm)，当遇见action操作时才会计算结果，如果你不了解Spark也没有关系，因为这很好理解。总之df可以经历很多很多的操作，每一个操作的结果也可以使用变量进行保存，但它们都是一个懒加载，不会立即执行。所以相当于是记录了前前后后的血缘关系，"谁"通过"什么操作"得到了"谁"，有人发现了这不就是DAG吗？是的，我们一开始就说了任务调度器就是使用了DAG的概念，而且Spark也是如此，所以这两者在某种程度上是比较相似的。当执行compute方法时，才会真正从头开始计算。我们还使用了ProgressBar来显示任务进度条，以及花费的时间，这是Dask提供的几种跟踪手段之一，用于跟踪运行中的任务，在使用本地任务调度器尤其方便。由于我们目前是在本地、没有使用集群，所以调度器就是本地任务调度器。

select = missing_count_pct[missing_count_pct != 0].index  # 筛选出缺失值百分比不为零的列
with ProgressBar():
    df_select = df[select].persist() 
df_select    

有趣的是select是一个pandas里的对象，但是我们可以将它和Dask DataFrame的方法一起使用，因为Dask DataFrame的每个分区都是一个pandas DataFrame。在这种情况下，pandas里的对象对所有线程都可用，因此它们可以在计算中使用它。在集群上运行时，pandas Series对象会被序列化并广播到所有工作节点。

另外我们看到除了compute之外，还有一个persist。这里调用compute返回是pandas中的对象，调用persist返回的依旧是Dask中的对象，这里是Dask DataFrame。但是这两者确实都发生了计算，从进度条我们也可以看出来，df[select].persist()返回的Dask DataFrame只有两个字段，证明Dask的确真正执行了df[select]逻辑。

另外，persist根据任务调度器的不同还会有不同的表现，如果任务调度器支持异步计算，那么persist会立即返回，返回值包含一个Dask Future对象；但如果任务调度器只支持阻塞式计算，那么persist也会处于阻塞状态，并且在计算之后依旧返回Dask中的对象，这里是Dask DataFrame。我们下面还会啰嗦一下persist，但是相比你已经猜到persist的使用场景了。

使用persist让复杂的计算更高效

这里我们再来啰嗦一下persist，有些时候我们不需要全部的列，因此我们需要将不要的列过滤掉，否则每次计算时都要加载额外的列。但是我们不能使用compute，因为这样就直接得到pandas中的对象了，所以返回的依旧得是Dask中的对象。回想一下，任务图中的节点一旦执行，它的中间结果就会被丢弃，因为要最小化内存使用，没有persist的话，这意味着想对做一些额外的事情(查看DataFrame的前五行)的话，我们将不得不再次重新运行整个转换链。为了避免多次重复的计算，Dask允许我们存储计算的中间结果，以便重用它们，这样的话就不需要重头计算了，而这一步就通过persist来完成。此外，如果Dask需要内存时，可能会从内存中删除一些分区，这些被删除的分区将在需要时被动态计算。尽管重新计算分区需要花些时间，但是它仍然比重新计算整个任务图要快的多。所以如果你有一个需要多次重用、并且非常大非常复杂的DAG，那么适当地使用persist进行持久化对于加速计算是非常有用的。

以上我们便结束了对Dask DataFrame的基本了解，现在你已经知道了如何通过几行代码读取数据并开始为探索性分析做准备。而上面的代码有一个最让人兴奋的特点，那就是无论你在单台机器处理还是在多台机器处理，无论分析的是几MB的数据还是几PB的数据，它们的工作原理都是一样的。另外由于它和pandas的代码非常相似，你可以对之前的代码进行很少的修改即可实现通过Dask执行并行计算。在后续系列我们将更深入地分析，但是目前我们有一个当务之急，我们需要挖掘一下Dask如何使用DAG来管理我们刚才说的任务分发。

可视化DAG

到目前为止，你已经知道了有向无环图(DAG)是如何工作的，并且了解了Dask是使用DAG来安排DataFrame的分布式计算。只不过我们还没有看到调度器创建DAG的具体过程，然而我们可以通过某些手段实现，Dask可以使用第三方库graphviz对任务调度器创建的DAG可视化，安装了这个库之后，我们便可以检查所有支持Dask Delayed对象的DAG，直接调用一个visualize函数即可。当然在安装这个库之后还不够，我们还需要去https://graphviz.org/download/网站下载相应的graphviz程序，如果是Windows直接进入到https://www2.graphviz.org/Packages/stable/windows/里面下载即可，我这里是一个zip包，解压之后将bin目录配置到环境变量即可(最好重启一下机器，我这里是需要重启的)。

使用Dask Delayed对象可视化一个简单的DAG

我们之前使用的是Dask DataFrame，现在我们后退一步，降低一个级别：Dask Delayed对象。之所以这么做的原因是，即使是简单的DataFrame对象，Dask为其创建的DAG中的节点也会有很多，这会加大可视化的难度。

import dask.delayed as delayed

def inc(i):
    return i + 1

def add(x, y):
    return x + y

# 关于delayed我们后面会说，这里可以认为将函数变成一个Delayed对象
# 然后传参方式依旧不变
x = delayed(inc)(1)
y = delayed(inc)(2)
z = delayed(add)(x, y)
# 进行可视化
z.visualize()

这图是程序帮我们画的，自下而上分析的话，还是很形象的，两个inc函数执行得到的结果传递给add函数。逻辑很简单，重点是里面的delayed函数，它是负责将普通的函数转化为一个Delayed对象。而且我们看到Delayed对象在调用时，可以接收普通的值，也可以接收别的Delayed对象在调用之后的返回值(当然最终得到的也是一个普通的值)。

type(x)  # dask.delayed.Delayed

这图是程序帮我们画的，自下而上分析的话，还是很形象的，两个inc函数执行得到的结果传递给add函数。逻辑很简单，重点是里面的delayed函数，它是负责将普通的函数转化为一个Delayed对象。而Delayed对象代表了DAG中的节点(task)，像我们之前的DataFrame是在Delayed之上的更高一级的对象，它对应多个Delayed对象。代码中的x表示函数inc的延迟求值，因为它不是被立刻执行的，当然Delayed对象还可以引用其它的Delayed对象，这一点从z的定义上也能看出来。这些Delayed对象连接在一起最终构成了一个图，对于要求值的z，首先要求x和y被计算出来。所以这便是一个简单的DAG示意图，对z的求值有一个很明显的依赖链，需要按照顺序求值，并且有一个准确定义的起点和终点。

从图中我们看到add函数是有依赖的，但是inc函数没有，因此如果inc函数执行时间比较的话，那么并行计算就很有意义。

使用循环和容器对象可视化一个复杂的DAG

让我们看一个稍微复杂的栗子。

import dask.delayed as delayed

def add_two(x):
    return x + 2

data = [1, 5, 8, 10]
# step_1是一个列表, 里面是对add_two函数的延迟求值
step_1 = [delayed(add_two)(i) for i in data]
# 这里将内置函数sum也变成了Delayed对象
total = delayed(sum)(step_1)
total.visualize()

整体没什么难度，现在我们弄得再复杂一些。

import dask.delayed as delayed

def add_two(x):
    return x + 2

def sum_two_numbers(x,y):
    return x + y

def multiply_four(x):
    return x * 4


data = [1, 5, 8, 10]
step_1 = [delayed(add_two)(i) for i in data]
step_2 = [delayed(multiply_four)(j) for j in step_1]
total = delayed(sum)(step_1)
total.visualize()

所以我们看到可以将多个计算连在一起，而无需立即计算中间结果。

使用persist降低DAG的复杂性

我们继续，将上一步计算出的结果和自己本身再加起来，然后再求和。

import dask.delayed as delayed

def add_two(x):
    return x + 2

def multiply_four(x):
    return x * 4

def sum_two_numbers(x,y):
    return x + y

data = [1, 5, 8, 10]
step_1 = [delayed(add_two)(i) for i in data]
step_2 = [delayed(multiply_four)(j) for j in step_1]
total = delayed(sum)(step_1)
# 将data中的每一个值都和total进行相加，得到data2
data2 = [delayed(sum_two_numbers)(k, total) for k in data]
# 然后对data2再进行求和
total2 = delayed(sum)(data2)
total2.visualize()

当然我相信逻辑依旧是很简单的，但问题是如果我们重复几次的话，那么这个DAG就会变得很大。类似的，如果我们的原始列表中有100个数字，不是4个，那么DAG也会变得非常非常的大，可以尝试一下。不过问题是为什么大型DAG难以处理，原因就是持久性。

正如之前提到的，每次在延迟对象上调用compute方法时，Dask都会逐步遍历完整的DAG来生成结果，这对于简单的计算来说是可以的。但如果处理的是非常大的分布式数据集， 那么一次又一次的重复计算很快会变得效率低下，而解决的一种办法就是持久化想要重用的中间结果，但是这对DAG来说会有什么影响呢。

import dask.delayed as delayed

def add_two(x):
    return x + 2

def multiply_four(x):
    return x * 4

def sum_two_numbers(x,y):
    return x + y

data = [1, 5, 8, 10]
step_1 = [delayed(add_two)(i) for i in data]
step_2 = [delayed(multiply_four)(j) for j in step_1]
total = delayed(sum)(step_1)
# 将total持久化
total_persist = total.persist()
total_persist.visualize()

某个节点想要开始执行，必须要保证所有指向它的节点都完成，total_persist是由total持久化得到的，所以没有任何节点指向它，对于此时得到的DAG是一个空的。

import dask.delayed as delayed

def add_two(x):
    return x + 2

def multiply_four(x):
    return x * 4

def sum_two_numbers(x,y):
    return x + y

data = [1, 5, 8, 10]
step_1 = [delayed(add_two)(i) for i in data]
step_2 = [delayed(multiply_four)(j) for j in step_1]
total = delayed(sum)(step_1)
# 将total持久化
total_persist = total.persist()
data2 = [delayed(sum_two_numbers)(k, total_persist) for k in data]
total2 = delayed(sum)(data2)
total2.visualize()

我们看懂这个DAG只有一半部分，这是因为total_persist已经被计算、并且持久化了，所以Dask可以使用持久化数据，而不是重新计算整个DAG，从而减少生成结果所需要的计算数量。

当然了你可以尝试绘制出更大的DAG，尽管不适合显示，但是却能让你知道：Dask可以非常优雅地处理复杂的应用。

任务调度

正如我们之前提到的，Dask在其API中使用了惰性计算的概念，而它带来的效果我们也已经看到了，当我们想看到结果时，必须调用compute方法。考虑到处理pb级数据会花费很长时间，这是非常有利的，因为在请求结果之前不会发生任何的计算，我们可以连续定义很多的操作，而不必等待一个操作完成后再定义下一个操作。

惰性计算

惰性(延迟)计算还允许Dask将工作分割为更小的逻辑块，这可以避免将操作的整个数据结构都加载到内存中，正如我们之前读取的item.csv文件，大小是3.85GB，分成了65个分区，平均每个分区处理的大小是60.65MB。

但是当我们像Dask请求结果时究竟发生了什么呢？首先你定义的所有计算都是由DAG表示，它是对你想要的结果进行计算的每一步计划。然而这些计划并没有指定到底该使用哪些资源执行计算，因此我们必须要考虑两件事：计算要在哪里发生，以及必要时计算的结果又要发送到哪里。和关系型数据库不同，在工作开始前，Dask不会预先确定每个任务的精确运行时位置。相反，任务调度器会动态地评估哪些工作已经完成，哪些工作还没有完成，以及哪些资源可以实时地接收额外的工作。这将允许Dask优雅地处理分布式计算中出现的大量问题，包括worker故障恢复，网络不可靠性，以及worker完成任务的速度不一致等问题。此外，任务调度器可以跟踪中间结果的存储位置，减少数据的传输以及从头计算，在集群上操作Dask，这将大大的提高效率。

数据本地性

由于Dask可以很容易地将我们的代码从单机扩展到数百、数千个服务器上，因此任务调度器必须做出明智的决定，判断哪些任务应该在哪些机器上执行。Dask使用一个集中的任务调度器来协调这些工作，为此每个Dask工作节点都要像任务调度器报告它有哪些可用数据以及它正在做哪些工作，然后任务调度器不断评估地集群的状态，为用户提交的计算提供公平、高效的执行计划。在大多数情况下，如果任务调度器在集群中的机器之间可以均匀地分配工作，那么计算就可以很快速和高效地完成。但很多时候却并不是这样的，比如一台服务器比其他服务器的任务更重，或者硬件不如其他服务器强大，或者不能快速地访问数据等等，这些情况都会导致该服务器的任务执行落后于其它服务器，有可能别人服务器都执行完毕了，该服务器还在执行中，那么为了保证高效，应该将该服务器的任务适当的减少，以避免成为瓶颈。而任务调度器的动态特性便允许它在无法避免的情况下，对这些情况作出反应。

为了获得最佳性能，Dask集群应该使用分布式文件系统，比如：S3或者HDFS来负责数据存储。那么这是为什么呢？来考虑一下反例，如果一个文件只存储在一台机器上，那么另一台机器如果想计算的话是不是要将数据发送到该服务器中呢？而且如果你用过Spark的话，那么你应该听过一句话：移动数据不如移动计算，因为数据(大型)的移动是非常耗费时间的。

所以两台服务器，如果数据都在一台服务器上，那么另一台服务器肯定要先将数据传输到本地，然后才可以读取，而一旦涉及到网络间的数据传输，那么耗时是少不了的。

而解决这个问题的办法就是将文件进行分割，切分成多个小块，不同机器上存储不同的块，而这也是分布式文件系统所做的。而且Dask的任务调度器也会考虑数据局部性，或者数据的物理位置，判断计算应该在哪里进行。

但是完全地避免数据间的移动也是不太可能的，一些数据必须通过广播到集群中的所有机器，不过任务调度器会尽最大努力让移动的数据量达到最小化。因为当数据量比较小的时候，这可能没有太大影响，但是当数据量非常大的时候，在网络中移动数据的影响就会大很多，所以最小化数据移动通常会带来更高的计算性能。

希望你现在对DAG在Dask将大量工作分解为更易于管理的部分时所扮演的角色，有一个更好的理解。我们在后续系列中还会回顾Delayed API，但请记住：整个系列中，我们看到的Dask的每一部分都是由Dealyed对象支持的，并且你还可以随时对DAG进行可视化。尽管在实践中，你可能不需要如此详细地对计算进行故障排除，但是理解Dask的底层机制将会帮助你更好地识别任务中的潜在问题和瓶颈。

总结

• Dask DataFrame上的计算是由使用DAG的任务调度器所支持的。
• 计算是惰性的，需要调用compute方法来执行计算并检验结果。
• 可以调用任何Dask对象上的visualize方法来对可视化底层的DAG。
• 可以通过persist对计算的中间结果进行存储和重用，从而简化计算。
• 应该将计算往数据靠近，而不是将数据往计算靠近，因为移动数据所以造成的网络和IO延迟对整个程序的影响是很大的。