跳至主要内容

用 Dask 提升您的数据科学工作流：完整指南

了解 Dask 如何通过并行计算和惰性求值革新数据处理。
· 11分钟 阅读

使用 AI 探索

当 Wes McKinney 开始编写 pandas 时，他有条经验法则：要让 pandas 最优化运行，机器的内存（RAM）应当是数据集大小的 5–10 倍。这个规则在 2010 年左右还容易遵循，但如今已是 2023 年。

到 2020 年，真实世界的数据集规模已经轻松让普通笔记本和台式机崩溃。对此问题的前瞻性预判，使得一个解决方案在 2015 年就已发布。

Dask 是由 Anaconda 团队开发的开源库，旨在解决在单机内存容量之外的大型数据集上进行可扩展且高效计算的挑战。

本教程将全面介绍 Dask 及其关键特性，包括 DataFrame、Array 和 Bag 等接口。

安装与配置 Dask

与其他库一样，您可以通过三种方式安装 Dask：Conda、Pip 和源码安装。

由于本文是 Dask 的入门介绍，我们不会涉及最后一种安装方式，那更适合维护者使用。

如果您使用 Anaconda，Dask 已包含在默认安装中（这也反映了该库的受欢迎程度）。如果您希望重新安装或升级它，可以使用 install 命令：

conda install dask

PIP 的替代方式如下：

pip install "dask[complete]"

添加 [complete] 扩展还会安装 Dask 所需的依赖项，无需手动安装 NumPy、Pandas 和 Tornado。

您可以通过查看库的版本来检查安装是否成功：

import dask

dask.__version__

输出：

'2023.5.0'

您使用 Dask 的大多数工作将聚焦于三种接口：Dask DataFrame、Array 和 Bag。让我们与 numpy 和 pandas 一起导入它们，供本文其余部分使用：

import dask.array as da
import dask.bag as db
import dask.dataframe as dd
import numpy as np
import pandas as pd

Dask 的基本概念

从高层来看，可以将 Dask 视为一个封装器，它扩展了 pandas、NumPy 和 Spark 等传统工具的能力，以处理超过内存的大型数据集。

当面对大对象，例如大于内存的数组（向量）或矩阵（数据框）时，Dask 会将它们切分为多个块（chunk），也称为分区（partition）。

例如，考虑在 NumPy 和 Dask 中包含 12 个随机数的数组：

narr = np.random.rand(12)

narr
array([0.9261154 , 0.87774082, 0.87078873, 0.22309476, 0.24575174,
      0.04182393, 0.31476305, 0.04599283, 0.62354124, 0.97597454,
      0.23923457, 0.81201211])
darr = da.from_array(narr, chunks=3)
darr

image2.png

上图显示，由于我们将 chunks 设为 3，这个 Dask 数组包含四个块。在底层，每个块本身就是一个 NumPy 数组。

现在，让我们看一个更大的例子。我们将创建两个 1 万行 × 10 万列的数组（10 亿个元素），并在两个库中进行逐元素相乘，同时测量性能：

# Create the NumPy arrays
arr1 = np.random.rand(10_000, 100_000)
arr2 = np.random.rand(10_000, 100_000)

# Create the Dask arrays
darr1 = da.from_array(arr1, chunks=(1_000, 10_000))
darr2 = da.from_array(arr2, chunks=(1_000, 10_000))
%%time

result_np = np.multiply(arr1, arr2)
Wall time: 3.19 s
%%time

result_dask = da.multiply(darr1, darr2)
Wall time: 94.8 ms

上述代码演示了使用 NumPy 和 Dask 对两个大数组进行逐元素相乘。正如输出所示，对于该计算，Dask 大约比 NumPy 快 34 倍。随着计算复杂度和数组规模增加，性能优势会更加显著。

对于其他对象，Dask 也采用类似的分块方式，并将这些块分布到您机器上的所有可用内核上。

Dask DataFrame

要充分体会 Dask 的优势，我们需要一个大型数据集，最好超过 1 GB。不过，为了跟随教程下载如此大的数据集可能并不合适。您可以改用此脚本，它会生成一个包含 1000 万行、10 个数值特征和 10 个类别特征的合成数据集。

请确保您的机器至少有 12 GB RAM 以运行该脚本。

当您在工作区中得到 large_dataset.csv 文件后，可以使用 Dask DataFrame 接口（dd）中的 read_csv 函数加载它：

import dask.dataframe as dd

dask_df = dd.read_csv("data/large_dataset.csv")

dask_df.head()

image4.png

尽管文件很大，您会注意到结果几乎是瞬时返回。对于更大的文件，您可以指定 `blocksize` 参数，它决定将文件按多少字节切分。

类似于 Dask Array 包含多个小的 NumPy 数组块，Dask 也被设计为沿行索引处理多个小的 Pandas DataFrame。

image5.png

正如从 read_csv 函数可以看出，pandas API 中大多数常用的语法和功能在 Dask 中都得以保留。以下代码块应当让您联想到使用 pandas 时的熟悉感。

选择列与逐元素运算

在这个示例中，我们对名为 dask_df 的 Dask DataFrame 进行一些直观的列运算。我们将 Numeric_0 列的值与 Numeric_9 与 Numeric_3 相乘的结果相加。我们把结果存入变量 result。

result = (
   dask_df["Numeric_0"] + dask_df["Numeric_9"] * dask_df["Numeric_3"]
)

result.compute().head()


0    1.301960
1    1.190679
2    1.100955
3    0.758272
4    0.926729
dtype: float64

如前所述，Dask 与传统计算工具略有不同，它不会立即执行这些操作。相反，它会创建一种称为任务图（task graph）的“计划”，以便稍后再执行。这种方式让 Dask 能在需要时优化并并行化计算。compute() 函数会触发 Dask 最终去执行这些计算，而 head() 则用于展示结果的前几行。

条件过滤

现在来看 Dask 如何过滤数据。我们选择 DataFrame 中 "Categorical_5" 列值为 "A" 的行。

这种过滤过程与在 pandas 中的做法类似，但有个不同点——Dask 以惰性方式执行。它会为该操作准备任务图，但会等到我们调用 compute() 时才真正执行。当我们运行 head()，就能看到过滤后 DataFrame 的前几行。

dask_df[dask_df["Categorical_5"] == "A"].compute().head()

image1.png

常见汇总统计

接下来，我们将使用 Dask 的 describe() 函数生成一些常见的汇总统计量。

它会为我们的 DataFrame 提供一系列描述性统计，包括均值、标准差、最小值、最大值等等。与前面的示例一样，当我们调用 describe() 时，Dask 会为该操作准备任务图，但要等到我们调用 compute() 才会执行。

dask_df.describe().compute()

image3.png

dask_df["Categorical_3"].value_counts().compute().head()
Categorical_3
O    386038
C    385804
A    385493
P    385490
K    385116
Name: count, dtype: int64

我们还使用 value_counts() 来统计 "Categorical_3" 列中每个唯一值的出现次数。通过 compute() 触发计算，head() 展示最常见的取值。

分组（Groupby）

最后，我们使用 groupby() 按 "Categorical_8" 列的取值对数据进行分组。然后选择 "Numeric_7" 列并计算每个分组的均值。

这与您在 pandas 中使用 ‘groupby()’ 的方式类似，但如您所料，Dask 以惰性方式进行。我们通过 compute() 触发计算，head() 展示前几个分组中 "Numeric_7" 列的平均值。

dask_df.groupby("Categorical_8")["Numeric_7"].mean().compute().head()


Categorical_8
A    0.498497
B    0.499767
C    0.500622
D    0.500307
E    0.499530
Name: Numeric_7, dtype: float64

关于 pandas 与 Dask 之间更多相通之处，请参阅Dask 用户指南的这一部分。

惰性求值

现在，让我们来看看为何每个代码块的末尾都要使用 compute 函数。

与会立即返回结果的 Pandas 的“急切执行”（eager）模式不同，Dask 以惰性模式（lazy）评估代码块。

打个烹饪方面的比方，惰性求值就像是提前备菜、切好蔬菜，但只在需要时才把它们混合下锅烹饪。compute 函数就起到这个作用。

相对地，急切求值就像食材一备好就立刻下锅开火，确保一切能同时上桌。

惰性求值是 Dask 优秀性能的关键，因为它能够：

  1. 减少计算量。 只有在需要（调用 compute）时才会对表达式求值，避免产生不必要的中间结果。
  2. 优化资源分配。 惰性求值避免将内存或算力分配给可能最终不需要的中间结果。
  3. 支持大型数据集。 这种方法按需或以更小的块处理数据元素，更高效地利用内存资源。

当 compute 的结果返回时，类型将是 Pandas 的 Series/DataFrame 或 NumPy 数组，而不是原生的 Dask DataFrame。

>>> type(dask_df)
dask.dataframe.core.DataFrame


>>> type(
   dask_df[["Numeric_5", "Numeric_6", "Numeric_7"]].mean().compute()
)

pandas.core.series.Series

其原因在于，大多数数据处理操作只会返回原始数据框的一小部分，所占空间要小得多。因此，此时无需再使用 Dask 的并行能力，您可以在后续工作中继续使用 pandas 或 NumPy。

适用于非结构化数据的 Dask Bag 与 Dask Delayed

Dask Bag 和 Dask Delayed 是 Dask 库中的两个组件，为处理非结构化或半结构化数据并实现惰性求值提供了强大工具。

过去表格数据最为常见，而如今的数据集常包含非结构化文件，如图像、文本、视频和音频。Dask Bag 提供了并行且可扩展地处理此类非结构化文件的功能与 API。

例如，来看一个简单示例：

import dask.bag as db

# Create a Dask Bag from a list of strings
b = db.from_sequence(["apple", "banana", "orange", "grape", "kiwi"])

# Filter the strings that start with the letter 'a'
filtered_strings = b.filter(lambda x: x.startswith("a"))

# Map a function to convert each string to uppercase
uppercase_strings = filtered_strings.map(lambda x: x.upper())

# Compute the result as a list
result = uppercase_strings.compute()

print(result)
['APPLE']

在该示例中，我们从字符串列表创建了一个 Dask Bag b。随后在 Bag 上应用操作，筛选出以字母 a 开头的字符串，并分别通过 filter() 和 map() 将其转换为大写。最后使用 compute() 将结果计算为列表并打印输出。

现在想象一下，您可以对存储在文本文件中的数十亿个类似字符串执行更复杂的操作。若没有 Dask Bag 提供的惰性求值和并行能力，您将面临巨大的挑战。（在Dask 文档中阅读更多关于 Bag 的内容）。

至于 Dask Delayed，它提供了更强的灵活性，并将惰性求值与并行能力引入到更多场景。借助 Dask Delayed，您可以通过 @dask.delayed 装饰器将任意原生 Python 函数转换为惰性对象。

下面是一个简单示例：

%%time

import time
import dask


@dask.delayed
def process_data(x):
   # Simulate some computation
   time.sleep(1)
   return x**2


# Generate a list of inputs
inputs = range(1000)

# Apply the delayed function to each input
results = [process_data(x) for x in inputs]

# Compute the results in parallel
computed_results = dask.compute(*results)



Wall time: 42.1 s

在该示例中，我们定义了用 @dask.delayed 装饰的函数 process_data。该函数通过 sleep 1 秒来模拟计算工作，然后返回输入值的平方。

如果没有并行化，对 1000 个输入执行此计算将耗时超过 1000 秒。然而，借助 Dask Delayed 和并行执行，计算仅耗时约 42.1 秒。

该示例展示了并行化如何通过高效地在多个内核或工作进程之间分配负载，从而减少计算时间。

这就是并行的意义所在。

结语与延伸资源

Dask 是数据生态中的基石库之一。它扩展了诸多广受喜爱的库（如 NumPy、Pandas 和 Spark）的功能，使处理超出内存的大型数据集变得顺畅自如。

借助 Dask Bag 和 Dask Delayed，它将并行与惰性求值带入了诸如处理非结构化数据或原生 Python 对象等非常规场景。

若想了解更细致的实现细节，请务必仔细阅读Dask 文档。

如果您在寻找系统掌握 Dask 的资源，请查看我们的课程——Parallel Programming With Dask in Python。

主题

Python