06_distributed_advanced.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "source": [
    "<img src=\"images/dask_horizontal.svg\" align=\"right\" width=\"30%\">"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "# 分布式，进阶"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 分布式futures"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from dask.distributed import Client\n",
    "c = Client(n_workers=4)\n",
    "c.cluster"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "在前一章中，我们展示了使用分布式执行器执行计算（使用delayed创建）与任何其他执行器相同。 但是，我们现在可以访问其他功能，并控制内存中保存的数据。\n",
    "\n",
    "首先，`futures` 接口（源自内置的 `concurrent.futures`）允许类似 map-reduce 的功能。 我们可以使用一组输入提交单个函数进行评估，或者使用 `submit()` 和 `map()` 对一系列输入进行评估。 请注意，调用立即返回，给出一个或多个 *futures*，其状态从\"pending\"开始，然后变为\"finished\"。 本地 Python 会话不会被阻塞。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "下面是最简单的`submit`示例："
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "def inc(x):\n",
    "    return x + 1\n",
    "\n",
    "fut = c.submit(inc, 1)\n",
    "fut"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "我们可以根据需要随时重新执行以下单元格，以此作为轮询 `future` 状态的一种方式。 这当然可以在循环中完成，在每次迭代时暂停一小段时间。 我们可以继续我们的工作，或者查看仍在进行的工作进度条，或者强制等待，直到未来准备就绪。\n",
    "\n",
    "与此同时，`status` 仪表板（上面的Cluster插件旁边的链接）在任务流中获得了一个新元素，表明 `inc()` 已完成，并且问题处的进度部分显示一个任务已完成和 保存在记忆中。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "fut"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "您可以调查的可能替代方案：\n",
    "```python\n",
    "from dask.distributed import wait, progress\n",
    "progress(fut)\n",
    "```\n",
    "会在*这个*notebook中显示进度条，而不必转到dashboard。这个进度条也是异步的，同时不会阻塞其他代码的执行。\n",
    "\n",
    "```python\n",
    "wait(fut)\n",
    "```\n",
    "会阻塞并强制笔记本等待，直到`fut`指向的计算完成。但是，请注意`inc()`的结果位于集群中，现在**不需要花费时间**执行计算，因为Dask注意到我们正在请求它已经知道的计算结果。稍后会详细介绍。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 取回信息 - 如果fut没有准备好的话，会阻塞\n",
    "c.gather(fut)\n",
    "# 仅考虑单个`future`时的等效动作\n",
    "# fut.result()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "在这里，我们看到了在集群上执行工作的另一种方式：当您将输入作为`future`提交或映射时，*计算移动到数据*而非相反，客户端在本地 Python 会话中 ，永远不需要看到中间值。 这类似于使用delayed构建图，实际上，delayed可以与future结合使用。 这里我们使用之前的延迟对象 `total`。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 一些需要时间的琐碎工作\n",
    "# 从\"分布式\"章节中复制而来\n",
    "\n",
    "from dask import delayed\n",
    "import time\n",
    "\n",
    "def inc(x):\n",
    "    time.sleep(5)\n",
    "    return x + 1\n",
    "\n",
    "def dec(x):\n",
    "    time.sleep(3)\n",
    "    return x - 1\n",
    "\n",
    "def add(x, y):\n",
    "    time.sleep(7)\n",
    "    return x + y\n",
    "\n",
    "x = delayed(inc)(1)\n",
    "y = delayed(dec)(2)\n",
    "total = delayed(add)(x, y)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 注意与 total.compute() 的区别\n",
    "# 请注意，此单元格立即完成\n",
    "fut = c.compute(total)\n",
    "fut"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "c.gather(fut) # 等，直到结果准备好"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### `Client.submit`\n",
    "\n",
    "`submit` 接受一个函数和参数，将它们推送到集群，返回一个表示要计算的结果的 *Future*。 该函数被传递给一个工作进程进行计算。 请注意，此单元格立即返回，而计算可能仍在集群上进行。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "fut = c.submit(inc, 1)\n",
    "fut"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "这看起来很像上面的`compute()`，除了现在我们是将函数和参数直接传递给集群。 对于习惯了 `concurrent.futures` 的人来说，这看起来很熟悉。 这个新的 `fut` 的行为方式与上面的相同。 请注意，我们现在已经覆盖了之前的 `fut` 定义，它将被垃圾收集(GC)，因此，之前的结果被集群释放。\n",
    "\n",
    "### 练习：使用 `Client.submit` 重建上述延迟计算\n",
    "\n",
    "传递给 `submit` 的参数可以是来自其他提交操作或延迟对象的 `future`。 特别是前者，展示了*将计算转移到数据*的概念，这是使用 Dask 进行编程的最强大元素之一。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 将你的代码写在这里"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "x = c.submit(inc, 1)\n",
    "y = c.submit(dec, 2)\n",
    "total = c.submit(add, x, y)\n",
    "\n",
    "print(total)     # 这仍然是一个future\n",
    "c.gather(total)   # 这会阻塞，直到计算完成"
   ],
   "outputs": [],
   "metadata": {
    "jupyter": {
     "source_hidden": true
    }
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "每个`future`代表集群保留，或在计算中的结果。 因此我们可以控制中间值的缓存 —— 当一个`future`不再被引用时，它的值就会被遗忘。 在上面的解决方案中，为每个函数调用都保留`future`。 如果我们选择提交更多需要它们的工作，那么就不需要重新计算这些结果。\n",
    "\n",
    "我们可以使用`scatter()`显式地将数据从本地会话传递到集群中，但通常更好的做法是，构造函数，在worker内部加载数据，这样就不需要序列化和通信数据。 Dask中的大多数加载函数，比如`dd.read_csv`，都是这样工作的。 同样，我们通常不希望使用`gather()`获取在内存中太大的结果。\n",
    "\n",
    "分布式调度器的[完整 API](http://distributed.readthedocs.io/en/latest/api.html)提供了与集群交互的详细信息，请记住，集群可以是在您的本地机器上，也可以是大型计算资源上。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "futures API提供了一种工作提交风格，可以轻松模拟许多人可能熟悉的 map/reduce 范式（参见 `c.map()`）。 由future表示的中间结果可以传递给新任务，而不必从集群本地拉取，并且可以将新工作分配给尚未开始的先前工作的输出。\n",
    "\n",
    "通常，任何使用 `.compute()` 执行的 Dask 操作都可以使用 `c.compute()` 提交异步执行，这适用于所有集合。 这是之前在 Bag 章节中看到的计算示例。 我们已经用分布式客户端版本替换了那里的 `.compute()` 方法，因此，我们可以继续提交更多工作（可能基于计算结果），或者，在下一个单元格中，跟踪计算的进度。 类似的进度条出现在监控 UI 页面中。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "%run prep.py -d accounts"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "import dask.bag as db\n",
    "import os\n",
    "import json\n",
    "filename = os.path.join('data', 'accounts.*.json.gz')\n",
    "lines = db.read_text(filename)\n",
    "js = lines.map(json.loads)\n",
    "\n",
    "f = c.compute(js.filter(lambda record: record['name'] == 'Alice')\n",
    "       .pluck('transactions')\n",
    "       .flatten()\n",
    "       .pluck('amount')\n",
    "       .mean())"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from dask.distributed import progress\n",
    "# 请注意，进度必须是单元格的最后一行\n",
    "# 才能显示\n",
    "progress(f)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 获取结果\n",
    "c.gather(f)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 通过删除future来释放变量值\n",
    "del f, fut, x, y, total"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### 持久化"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "考虑哪些数据应该由 worker 加载，而不是传递，以及哪些中间值要保留在 worker 内存中，在许多情况下将决定进程的计算效率。\n",
    "\n",
    "在这里的例子中，我们重复 Array 章节中的一个计算——注意每次调用 `compute()` 的速度大致相同，因为每次都包括数据的加载。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "%run prep.py -d random"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "import h5py\n",
    "import os\n",
    "f = h5py.File(os.path.join('data', 'random.hdf5'), mode='r')\n",
    "dset = f['/x']\n",
    "import dask.array as da\n",
    "x = da.from_array(dset, chunks=(1000000,))\n",
    "\n",
    "%time x.sum().compute()\n",
    "%time x.sum().compute()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "相反，如果我们预先将数据持久化到 RAM 中（这需要几秒钟才能完成 - 我们可以在此过程中使用 `wait()`），那么进一步的计算将会更快。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 将 x 从一组delayed变量\n",
    "# 更改为一组指向 RAM 中数据的future。\n",
    "# 在 UI dashboard上查看此内容。\n",
    "x = c.persist(x)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "%time x.sum().compute()\n",
    "%time x.sum().compute()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "自然地，整个过程中持久化每个中间变量是一个坏主意，因为这往往会填满所有可用的 RAM 并使整个系统变慢（或崩溃！）。理想的持久化位置通常是在一组数据清理步骤的末尾，此时数据处于一种经常被查询的形式。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "**练习**：一旦我们知道我们已经完成了`x`，与`x`相关的内存是如何释放的？"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 异步计算\n",
    "<img style=\"float: right;\" src=\"images/450px-Rosenbrock_function.svg.png\" height=200 width=200>\n",
    "\n",
    "使用futures API 的一个好处是您可以进行动态计算，随着事情的进展进行调整。 在这里，我们通过在结果出现时循环遍历结果来实现一个简单的朴素搜索，并在其他点仍在运行时提交新点以进行计算。\n",
    "\n",
    "在运行时观察 [诊断仪表板](../../9002/status)，您可以看到计算正在并发运行，同时正在提交更多计算。 这种灵活性对于需要某种程度的同步的并行算法非常有用。\n",
    "\n",
    "让我们使用动态规划来执行一个非常简单的最小化。 用到的函数被称为Rosenbrock： \n",
    "\n",
    "译者注：Rosenbrock函数是在数学优化领域中，一个用来测试最优化算法性能的非凸函数。其定义为:\n",
    "$$f(x,y)=(1-x)^2+100(y-x^2)^2$$"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 一个具有有趣最小值的简单函数\n",
    "import time\n",
    "\n",
    "def rosenbrock(point):\n",
    "    \"\"\"计算rosenbrock function并返回点和结果\"\"\"\n",
    "    time.sleep(0.1)\n",
    "    score = (1 - point[0])**2 + 2 * (point[1] - point[0]**2)**2\n",
    "    return point, score"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "初始设置，包括创建图形。 我们为此使用 Bokeh，它允许在结果出现时动态更新图形。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from bokeh.io import output_notebook, push_notebook\n",
    "from bokeh.models.sources import ColumnDataSource\n",
    "from bokeh.plotting import figure, show\n",
    "import numpy as np\n",
    "output_notebook()\n",
    "\n",
    "# 设置绘图背景\n",
    "N = 500\n",
    "x = np.linspace(-5, 5, N)\n",
    "y = np.linspace(-5, 5, N)\n",
    "xx, yy = np.meshgrid(x, y)\n",
    "d = (1 - xx)**2 + 2 * (yy - xx**2)**2\n",
    "d = np.log(d)\n",
    "\n",
    "p = figure(x_range=(-5, 5), y_range=(-5, 5))\n",
    "p.image(image=[d], x=-5, y=-5, dw=10, dh=10, palette=\"Spectral11\");"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "我们从 (0, 0) 处的一个点开始，并在其周围随机散布测试点。 每次评估需要大约 100 毫秒，随着结果的出现，我们测试是否有新的最佳点，并在搜索框缩小时在新的最佳点周围选择随机点。\n",
    "\n",
    "每次我们有新的最佳值时，我们都会打印函数值和当前最佳位置。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from dask.distributed import as_completed\n",
    "from random import uniform\n",
    "\n",
    "scale = 5                  # 初始化随机扰动尺度\n",
    "best_point = (0, 0)        # 初始化猜测值\n",
    "best_score = float('inf')  # 迄今为止最好的分数\n",
    "startx = [uniform(-scale, scale) for _ in range(10)]\n",
    "starty = [uniform(-scale, scale) for _ in range(10)]\n",
    "\n",
    "# 设置绘图\n",
    "source = ColumnDataSource({'x': startx, 'y': starty, 'c': ['grey'] * 10})\n",
    "p.circle(source=source, x='x', y='y', color='c')\n",
    "t = show(p, notebook_handle=True)\n",
    "\n",
    "# 初始化10个随机点\n",
    "futures = [c.submit(rosenbrock, (x, y)) for x, y in zip(startx, starty)]\n",
    "iterator = as_completed(futures)\n",
    "\n",
    "for res in iterator:\n",
    "    # 取一个完整的点，它是一个改进吗？\n",
    "    point, score = res.result()\n",
    "    if score < best_score:\n",
    "        best_score, best_point = score, point\n",
    "        print(score, point)\n",
    "\n",
    "    x, y = best_point\n",
    "    newx, newy = (x + uniform(-scale, scale), y + uniform(-scale, scale))\n",
    "    \n",
    "    # 更新绘图\n",
    "    source.stream({'x': [newx], 'y': [newy], 'c': ['grey']}, rollover=20)\n",
    "    push_notebook(document=t)\n",
    "    \n",
    "    # 动态添加新点，以在集群上工作\n",
    "    new_point = c.submit(rosenbrock, (newx, newy))\n",
    "    iterator.add(new_point)  # 也开始跟踪新任务\n",
    "\n",
    "    # 缩小搜索范围并考虑停止\n",
    "    scale *= 0.99\n",
    "    if scale < 0.001:\n",
    "        break\n",
    "point"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 调试"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "当分布式作业出现问题时，很难弄清楚问题出在哪里以及如何处理。 当任务引发异常时，该异常将在收集该结果或其他依赖于它的结果时显示。\n",
    "\n",
    "考虑由集群计算的以下延迟计算。 像往常一样，我们返回一个future，集群继续进行计算（对于不重要的过程，这发生得非常缓慢）。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "@delayed\n",
    "def ratio(a, b):\n",
    "    return a // b\n",
    "\n",
    "ina = [5, 25, 30]\n",
    "inb = [5, 5, 6]\n",
    "out = delayed(sum)([ratio(a, b) for (a, b) in zip(ina, inb)])\n",
    "f = c.compute(out)\n",
    "f"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "我们只有在收集结果时才知道发生了什么（这对于 `out.compute()` 也是如此，除了在此期间我们不能做其他事情）。 对于第一组输入，它工作正常。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "c.gather(f)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "但是如果我们引入了错误的输入，就会引发异常。 异常发生在`ratio`中，但只有在计算总和时才会引起我们的注意。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "ina = [5, 25, 30]\n",
    "inb = [5, 0, 6]\n",
    "out = delayed(sum)([ratio(a, b) for (a, b) in zip(ina, inb)])\n",
    "f = c.compute(out)\n",
    "c.gather(f)"
   ],
   "outputs": [],
   "metadata": {
    "tags": [
     "raises-exception"
    ]
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "这种情况下的显示使异常的来源显而易见，但情况并非总是如此。 这应该如何调试，我们将如何找出导致异常的确切条件？\n",
    "\n",
    "当然，第一步是编写经过良好测试的代码，在出现问题时对其输入做出适当的断言并清除警告和错误消息。 这适用于所有的代码。\n",
    "\n",
    "最典型的做法是在本地线程中执行部分计算，这样我们就可以运行 Python 调试器并查询异常发生时事物的状态。 显然，在集群上处理大数据时，这不能在整个数据集上执行，但即使这样，一个合适的样本也可能会这样做。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "import dask\n",
    "with dask.config.set(scheduler=\"sync\"):\n",
    "    # 不要在这里使用 c.compute(out)\n",
    "    # 我们特别不想要分布式调度器\n",
    "    out.compute()"
   ],
   "outputs": [],
   "metadata": {
    "tags": [
     "raises-exception"
    ]
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# 取消注释以进入事后调试器\n",
    "# %debug"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "这种方法的问题在于，Dask是设计被用于执行大型数据集/计算 - 您可能无法简单地在一个本地线程中运行整个工程，否则您一开始就不会使用 Dask。 所以上面的代码应该只用于一小部分也表现出错误的数据。\n",
    "此外，当您处理future（例如上面的`f`，或在持久化之后）而不是基于延迟的计算时，该方法将不起作用。\n",
    "\n",
    "作为替代方案，您可以要求调度分析您的计算并找到导致错误的特定子任务，并仅在本地拉取它及其依赖项以执行。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "c.recreate_error_locally(f)"
   ],
   "outputs": [],
   "metadata": {
    "tags": [
     "raises-exception"
    ]
   }
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "# uncomment to enter post-mortem debugger\n",
    "# %debug"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "最后，当我们需要查看调度器/工作线程的状态时，还有异常以外的错误。 在我们开始的标准`LocalCluster`中，我们可以直接访问这些错误。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "[(k, v.state) for k, v in c.cluster.scheduler.tasks.items() if v.exception is not None]"
   ],
   "outputs": [],
   "metadata": {}
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "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
}