dask-futures.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Dask Futures\n",
    "\n",
    "Dask 支持扩展 Python 的concurrent.futures接口的实时任务框架\n",
    "\n",
    "它是即时的而不是惰性的，这在计算可能随时间演变的情况下提供了更多的灵活性。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table style=\"border: 2px solid white;\">\n",
       "<tr>\n",
       "<td style=\"vertical-align: top; border: 0px solid white\">\n",
       "<h3 style=\"text-align: left;\">Client</h3>\n",
       "<ul style=\"text-align: left; list-style: none; margin: 0; padding: 0;\">\n",
       "  <li><b>Scheduler: </b>tcp://127.0.0.1:60400</li>\n",
       "  <li><b>Dashboard: </b><a href='http://127.0.0.1:8787/status' target='_blank'>http://127.0.0.1:8787/status</a></li>\n",
       "</ul>\n",
       "</td>\n",
       "<td style=\"vertical-align: top; border: 0px solid white\">\n",
       "<h3 style=\"text-align: left;\">Cluster</h3>\n",
       "<ul style=\"text-align: left; list-style:none; margin: 0; padding: 0;\">\n",
       "  <li><b>Workers: </b>4</li>\n",
       "  <li><b>Cores: </b>16</li>\n",
       "  <li><b>Memory: </b>16.00 GiB</li>\n",
       "</ul>\n",
       "</td>\n",
       "</tr>\n",
       "</table>"
      ],
      "text/plain": [
       "<Client: 'tcp://127.0.0.1:60400' processes=4 threads=16, memory=16.00 GiB>"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from dask.distributed import Client\n",
    "client = Client(n_workers=4, threads_per_worker=4)\n",
    "client"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 简单使用"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "import random\n",
    "\n",
    "def inc(x):\n",
    "    time.sleep(random.random())\n",
    "    return x + 1\n",
    "\n",
    "def double(x):\n",
    "    time.sleep(random.random())\n",
    "    return 2 * x\n",
    "\n",
    "def add(x, y):\n",
    "    time.sleep(random.random())\n",
    "    return x + y"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# 本地运行方式\n",
    "inc(1)\n",
    "double(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<b>Future: int</b> <font color=\"gray\">status: </font><font color=\"black\">pending</font>, <font color=\"gray\">key: </font>int-3a498ea6e2a6c00f58dc190be9078b58"
      ],
      "text/plain": [
       "<Future: pending, key: int-3a498ea6e2a6c00f58dc190be9078b58>"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Dask远程执行\n",
    "future = client.submit(int, 1)\n",
    "future"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<b>Future: int</b> <font color=\"gray\">status: </font><font color=\"black\">finished</font>, <font color=\"gray\">type: </font>builtins.int, <font color=\"gray\">key: </font>int-3a498ea6e2a6c00f58dc190be9078b58"
      ],
      "text/plain": [
       "<Future: finished, type: builtins.int, key: int-3a498ea6e2a6c00f58dc190be9078b58>"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "future"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "future.result()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 链式依赖计算"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<b>Future: add</b> <font color=\"gray\">status: </font><font color=\"black\">pending</font>, <font color=\"gray\">key: </font>add-2edc5b140d75a0e6ef51ef775a4d3ebe"
      ],
      "text/plain": [
       "<Future: pending, key: add-2edc5b140d75a0e6ef51ef775a4d3ebe>"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# add函数依赖inc、double计算完后的值作为参数传入\n",
    "x = client.submit(inc, 1)\n",
    "y = client.submit(double, 2)\n",
    "z = client.submit(add, x, y)\n",
    "z"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "6"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "z.result()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 多个任务同时提交"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 329 ms, sys: 36.9 ms, total: 366 ms\n",
      "Wall time: 349 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "\n",
    "zs = []\n",
    "\n",
    "for i in range(256):\n",
    "    x = client.submit(inc, i)     # x = inc(i)\n",
    "    y = client.submit(double, x)  # y = inc(x)\n",
    "    z = client.submit(add, x, y)  # z = inc(y)\n",
    "    zs.append(z)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "total = client.submit(sum, zs)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "from dask.distributed import wait, progress"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "13285c188eb4451691289254ca7bfd46",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# 异步化进度条\n",
    "progress(total)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "98688"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "total.result()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 树求和"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "finish           total             single output\n",
    "    ^          /        \\\n",
    "    |        c1          c2        neighbors merge\n",
    "    |       /  \\        /  \\\n",
    "    |     b1    b2    b3    b4     neighbors merge\n",
    "    ^    / \\   / \\   / \\   / \\\n",
    "start   a1 a2 a3 a4 a5 a6 a7 a8    many inputs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "L = zs\n",
    "while len(L) > 1:\n",
    "    new_L = []\n",
    "    for i in range(0, len(L), 2):\n",
    "        future = client.submit(add, L[i], L[i + 1])  # add neighbors\n",
    "        new_L.append(future)\n",
    "    L = new_L    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "del zs, L, future, new_L, x, y"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "distributed.client - ERROR - Failed to reconnect to scheduler after 10.00 seconds, closing client\n",
      "_GatheringFuture exception was never retrieved\n",
      "future: <_GatheringFuture finished exception=CancelledError()>\n",
      "asyncio.exceptions.CancelledError\n"
     ]
    }
   ],
   "source": [
    "client.shutdown()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "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.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}