01_dask.delayed.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "source": [
    "<img src=\"http://dask.readthedocs.io/en/latest/_images/dask_horizontal.svg\"\n",
    "     align=\"right\"\n",
    "     width=\"30%\"\n",
    "     alt=\"Dask logo\\\">\n",
    "\n",
    "# 使用 `dask.delayed` 并行化代码\n",
    "\n",
    "在本节中，我们使用Dask和`Dask.delayed`,来并行化简单的for循环样式代码。通常，这是您将函数转换为与Dask一起使用的唯一函数。\n",
    "\n",
    "这是一种使用 `dask` 并行化现有代码库或构建 [复杂系统](https://blog.dask.org/2018/02/09/credit-models-with-dask) 的简单方法。 这也将有助于我们对后面的部分进行理解。 \n",
    "\n",
    "**相关文档**\n",
    "\n",
    "* [Delayed 文档](https://docs.dask.org/en/latest/delayed.html)\n",
    "* [Delayed 屏幕录像](https://www.youtube.com/watch?v=SHqFmynRxVU)\n",
    "* [Delayed API](https://docs.dask.org/en/latest/delayed-api.html)\n",
    "* [Delayed 样例](https://examples.dask.org/delayed.html)\n",
    "* [Delayed 最佳实践](https://docs.dask.org/en/latest/delayed-best-practices.html)"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "正如在[分布式调度器notebook](05_distributed.ipynb) 中看到的那样，Dask有几种并行执行代码的方法。我们将通过创建一个 `dask.distributed.Client`来使用分布式调度器。现在，这将为我们提供一些不错的诊断。稍后我们将深入讨论调度器。 "
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from dask.distributed import Client\n",
    "\n",
    "client = Client(n_workers=4)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 基础\n",
    "\n",
    "首先让我们制作一些玩具功能，例如`inc`和`add`，它们会休眠一段时间以模拟工作。 然后，我们将正常运行这些功能。\n",
    "\n",
    "在下一节中，我们将并行化此代码。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from time import sleep\n",
    "\n",
    "def inc(x):\n",
    "    sleep(1)\n",
    "    return x + 1\n",
    "\n",
    "def add(x, y):\n",
    "    sleep(1)\n",
    "    return x + y"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "我们使用 `%%time` 魔术命令来计时这段普通代码的执行时间，这是Jupyter Notebook的一个特殊功能。 "
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "%%time\n",
    "# 这需要三秒钟才能运行，\n",
    "# 因为我们一个接一个地按顺序调用每个函数\n",
    "\n",
    "x = inc(1)\n",
    "y = inc(2)\n",
    "z = add(x, y)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### 使用`dask.delayed`装饰器并行\n",
    "\n",
    "这两个增量调用*可以*并行调用，因为它们完全相互独立。\n",
    "\n",
    "我们将使用 `dask.delayed` 函数转换 `inc` 和 `add` 函数。 当我们通过传递参数调用延迟版本时，与以前完全一样，原始函数实际上还没有被调用 —— 这就是单元执行很快完成的原因。\n",
    "取而代之的是，生成了一个*延迟对象*，该对象跟踪要调用的函数以及要传递给该函数的参数。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from dask import delayed"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "%%time\n",
    "# 这会立即运行，它所做的只是构建一个图\n",
    "\n",
    "x = delayed(inc)(1)\n",
    "y = delayed(inc)(2)\n",
    "z = delayed(add)(x, y)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "这立即运行，因为还没有真正发生任何事情。\n",
    "\n",
    "要获得结果，请调用 `compute`。 请注意，这比原始代码运行得更快。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "%%time\n",
    "# 这里才事实上地使用本地线程池运行我们的计算 \n",
    "\n",
    "z.compute()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 刚才发生了什么？\n",
    "\n",
    "`z`对象是一个惰性(`lazy`)的`Delayed`对象。 该对象包含我们计算最终结果所需的一切，包括对所有必需功能的引用以及它们的输入以及与另一个函数的关系。 我们可以如上所述使用`.compute()`评估结果，也可以使用`.visualize()`可视化该值的任务图。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "z"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 查看`z`的任务图\n",
    "# 译者注:除了适用conda或pip安装python-graphviz，还需要在系统上安装graphviz。以ubuntu为例:sudo apt install graphviz\n",
    "z.visualize()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "注意，这包括之前的函数名称，以及`inc`函数的输出到`add`的输入的逻辑流程。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### 需要考虑的一些问题：\n",
    "\n",
    "-为什么从3秒变成2秒？ 为什么我们不能并行化到1s？  \n",
    "-如果inc和add函数不包含`sleep(1)`会发生什么？ Dask仍然可以加快此代码的速度吗？  \n",
    "-如果我们有多个输出或者还想访问`x`或`y`，该怎么办？  "
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 练习：并行化for循环\n",
    "\n",
    "`for`循环是我们要并行化的最常见的事物之一。 在`inc`和`sum`上使用`dask.delayed`来并行化以下计算："
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "data = [1, 2, 3, 4, 5, 6, 7, 8]"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "%%time\n",
    "# 串行代码\n",
    "\n",
    "results = []\n",
    "for x in data:\n",
    "    y = inc(x)\n",
    "    results.append(y)\n",
    "    \n",
    "total = sum(results)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "total"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "%%time\n",
    "# 将你的并行代码写在这儿"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "results = []\n",
    "\n",
    "for x in data:\n",
    "    y = delayed(inc)(x)\n",
    "    results.append(y)\n",
    "    \n",
    "total = delayed(sum)(results)\n",
    "print(\"Before computing:\", total)  # 让我们看看total是什么变量类型的\n",
    "result = total.compute()\n",
    "print(\"After computing :\", result)  # 计算完成后"
   ],
   "outputs": [],
   "metadata": {
    "jupyter": {
     "source_hidden": true
    }
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "与直接使用 `sum` 函数而不是用 `delayed` 包裹的版本相比，图形可视化与给定的解决方案相比如何？ 你能解释一下后面的版本吗？ 您可能会发现以下表达式的结果很有启发性：\n",
    "```python\n",
    "delayed(inc)(1) + delayed(inc)(2)\n",
    "```"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 练习：使用控制流并行化 for 循环代码\n",
    "\n",
    "通常，我们只希望延迟*某些*功能，立即运行其中一些功能。当这些函数快速运行并帮助我们确定应该调用哪些其他较慢的函数时，这特别有用。这个决定，延迟或不延迟，通常是我们在使用`dask.delayed`时需要考虑的地方。\n",
    "\n",
    "在下面的示例中，我们遍历输入列表。 如果输入是偶数，那么我们要调用`inc`。 如果输入为奇数，则我们要调用`double`。 必须立即（而不是懒惰地）做出决定调用`inc`或`double`的`is_even`，以便我们的图形构建Python代码继续进行。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "def double(x):\n",
    "    sleep(1)\n",
    "    return 2 * x\n",
    "\n",
    "def is_even(x):\n",
    "    return not x % 2\n",
    "\n",
    "data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "%%time\n",
    "# 串行代码\n",
    "\n",
    "results = []\n",
    "for x in data:\n",
    "    if is_even(x):\n",
    "        y = double(x)\n",
    "    else:\n",
    "        y = inc(x)\n",
    "    results.append(y)\n",
    "    \n",
    "total = sum(results)\n",
    "print(total)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "%%time\n",
    "# 将你的并行代码写在这儿。。。\n",
    "# 目标: 使用 dask.delayed 并行化上面的串行代码\n",
    "# 你需要延迟某些功能，但不是全部"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "results = []\n",
    "for x in data:\n",
    "    if is_even(x):  # even\n",
    "        y = delayed(double)(x)\n",
    "    else:          # odd\n",
    "        y = delayed(inc)(x)\n",
    "    results.append(y)\n",
    "    \n",
    "total = delayed(sum)(results)"
   ],
   "outputs": [],
   "metadata": {
    "jupyter": {
     "source_hidden": true
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "%time total.compute()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "total.visualize()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### 需要考虑的一些问题：\n",
    "\n",
    "- 还有哪些我们不能使用延迟执行的控制流示例？\n",
    "- 如果在上面的示例中延迟执行了对`is_even（x）`的求值，将会发生什么？\n",
    "- 您对延迟执行`sum`有何想法？ 这个函数既是计算，又运行速度也很快。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 练习：并行化Pandas Groupby归约\n",
    "\n",
    "在本练习中，我们读取了多个CSV文件，并且并行执行groupby操作。 我们获得了串行代码来执行此操作，并将其与`dask.delayed`并行化。\n",
    "\n",
    "我们将并行进行的计算是根据一些历史飞行数据来计算每个机场的平均离港延误。 我们将通过使用`dask.delayed`和`pandas`来做到这一点。 在以后的部分中，我们将使用`dask.dataframe`进行相同的练习。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 创建数据\n",
    "\n",
    "运行此代码以准备一些数据。\n",
    "\n",
    "此下载并提取了1990年到2000年之间纽约市以外航班的一些历史航班数据。原始数据来自[这里](http://stat-computing.org/dataexpo/2009/the-data.html)。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "%run prep.py -d flights"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### 检查数据"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "import os\n",
    "sorted(os.listdir(os.path.join('data', 'nycflights')))"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### 用`pandas.read_csv`读取一个文件并计算平均出发延迟"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "import pandas as pd\n",
    "df = pd.read_csv(os.path.join('data', 'nycflights', '1990.csv'))\n",
    "df.head()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 是什么样的数据类型？\n",
    "df.dtypes"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 数据中包含哪些始发机场？\n",
    "df.Origin.unique()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 每个机场的年平均离港延误\n",
    "df.groupby('Origin').DepDelay.mean()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### 顺序代码：每个机场的平均离港延误\n",
    "\n",
    "上面的单元格计算了每个机场一年的平均起飞延迟。 在这里，我们使用顺序for循环将其扩展到所有年份。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from glob import glob\n",
    "filenames = sorted(glob(os.path.join('data', 'nycflights', '*.csv')))"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "%%time\n",
    "\n",
    "sums = []\n",
    "counts = []\n",
    "for fn in filenames:\n",
    "    # 读取文件\n",
    "    df = pd.read_csv(fn)\n",
    "    \n",
    "    # Groupby出发地机场\n",
    "    by_origin = df.groupby('Origin')\n",
    "    \n",
    "    # 所有始发航班延误的总和\n",
    "    total = by_origin.DepDelay.sum()\n",
    "    \n",
    "    # 始发航班数\n",
    "    count = by_origin.DepDelay.count()\n",
    "    \n",
    "    # 保存中间变量\n",
    "    sums.append(total)\n",
    "    counts.append(count)\n",
    "\n",
    "# 合并中间量以获得总的平均始发延迟\n",
    "total_delays = sum(sums)\n",
    "n_flights = sum(counts)\n",
    "mean = total_delays / n_flights"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "mean"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### 并行化上面的代码\n",
    "\n",
    "使用`dask.delayed`并行化上面的代码。 您需要了解一些额外的信息。\n",
    "\n",
    "1.  对delayed对象的方法和属性访问会自动工作，因此，如果您有delayed对象，则可以对其执行常规的算术，切片和方法调用，它将产生正确的延迟调用。\n",
    "\n",
    "    ```python\n",
    "    x = delayed(np.arange)(10)\n",
    "    y = (x + 1)[::2].sum()  # 这里的一切都延迟执行了\n",
    "    ```\n",
    "2.  当您只有一个输出时，调用`.compute()`方法效果很好。 当您有多个输出时，您可能想使用`dask.compute`函数：\n",
    "\n",
    "    ```python\n",
    "    >>> from dask import compute\n",
    "    >>> x = delayed(np.arange)(10)\n",
    "    >>> y = x ** 2\n",
    "    >>> min_, max_ = compute(y.min(), y.max())\n",
    "    >>> min_, max_\n",
    "    (0, 81)\n",
    "    ```\n",
    "    \n",
    "    这样Dask可以共享中间值（例如`y = x ** 2`）\n",
    "    \n",
    "因此，您的目标是使用`dask.delayed`并行化上面的代码（已在下面复制）。 您可能还需要可视化一些计算，以查看是否正确执行了计算。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from dask import compute"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "%%time\n",
    "\n",
    "# 复制的串行代码\n",
    "\n",
    "sums = []\n",
    "counts = []\n",
    "for fn in filenames:\n",
    "    # 读取文件\n",
    "    df = pd.read_csv(fn)\n",
    "    \n",
    "    # Groupby出发地机场\n",
    "    by_origin = df.groupby('Origin')\n",
    "    \n",
    "    # 所有始发航班延误的总和\n",
    "    total = by_origin.DepDelay.sum()\n",
    "    \n",
    "    # 始发航班数\n",
    "    count = by_origin.DepDelay.count()\n",
    "    \n",
    "    # 保存中间变量\n",
    "    sums.append(total)\n",
    "    counts.append(count)\n",
    "\n",
    "# 合并中间量以获得总的平均始发延迟\n",
    "total_delays = sum(sums)\n",
    "n_flights = sum(counts)\n",
    "mean = total_delays / n_flights"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "mean"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "%%time\n",
    "# 将你的代码写在这儿。。。"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "如果您加载解决方案，请在单元格顶部添加`%%time`以测量运行时间。 "
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "%%time\n",
    "\n",
    "# 这只是一种可能的解决方案，有好几种方法可以使用`delayed`来做到这一点 \n",
    "\n",
    "sums = []\n",
    "counts = []\n",
    "for fn in filenames:\n",
    "    # 读取文件\n",
    "    df = delayed(pd.read_csv)(fn)\n",
    "\n",
    "    # Groupby 始发机场\n",
    "    by_origin = df.groupby('Origin')\n",
    "\n",
    "    # 所有始发地延误的总和\n",
    "    total = by_origin[\"DepDelay\"].sum()\n",
    "\n",
    "    # 始发地的延误的航班数量\n",
    "    count = by_origin[\"DepDelay\"].count()\n",
    "\n",
    "    # 保存中间变量\n",
    "    sums.append(total)\n",
    "    counts.append(count)\n",
    "\n",
    "# 计算中间变量\n",
    "sums, counts = compute(sums, counts)\n",
    "\n",
    "# 组合中间体以获得每个始发地的总平均延迟\n",
    "total_delays = sum(sums)\n",
    "n_flights = sum(counts)\n",
    "mean = total_delays / n_flights"
   ],
   "outputs": [],
   "metadata": {
    "jupyter": {
     "source_hidden": true
    }
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 确保结果仍然匹配\n",
    "mean"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 要考虑的一些问题：\n",
    "\n",
    "- 您获得了多少提速？ 这是您期望的加速速度吗？\n",
    "- 尝试在何处调用`compute`。 当您用`sums`和`counts`调用它时会发生什么？ 如果您等待并按`mean`调用该怎么办？\n",
    "- 尝试delaying call `sum`。 如果`sum`被延迟，图形将是什么样？ 如果不是，则图形看起来像什么？\n",
    "- 您能想出为什么要以一种方式比另一种方式进行归约的任何原因吗？\n",
    "\n",
    "### 了解更多 \n",
    "\n",
    "访问 [Delayed 文档](https://docs.dask.org/en/latest/delayed.html). 特别是，这个[delayed 屏幕录像](https://www.youtube.com/watch?v=SHqFmynRxVU) 将强化您在这里学到的概念和[delayed 最佳实践](https://docs.dask.org/en/latest/delayed-best-practices.html) 文档收集了关于如何使用 `dask.delayed` 的建议。 "
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 关闭客户端 \n",
    "\n",
    "在继续下一个练习之前，请确保关闭您的客户端或停止此内核。 "
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "client.close()"
   ],
   "outputs": [],
   "metadata": {}
  }
 ],
 "metadata": {
  "kernelspec": {
   "name": "python3",
   "display_name": "Python 3.8.10 64-bit ('dask-tutorial': conda)"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.10"
  },
  "interpreter": {
   "hash": "3f2e3fb42896b6b47c37375014a159d976fd49965917325c599b06eb4946867a"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}