08_machine_learning.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "source": [
    "# 并行和分布式机器学习\n",
    "\n",
    "[Dask-ML](https://dask-ml.readthedocs.io) 上有并行和分布式机器学习的资源。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 扩展类型\n",
    "\n",
    "您可能会面临几个明显的扩展问题。\n",
    "扩展策略取决于您面临的问题。\n",
    "\n",
    "1. CPU限制：数据适合RAM，但训练时间太长。 许多超参数组合，许多模型的大型集成等。\n",
    "2. 内存限制：数据大于 RAM，且不适用采样。\n",
    "\n",
    "![](images/ml-dimensions.png)"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "* 对于内存适宜的问题，只需使用 scikit-learn（或您最喜欢的 ML 库）。\n",
    "* 对于大型模型，使用 `dask_ml.joblib` 和你最喜欢的 scikit-learn 估计器(`estimator`)\n",
    "* 对于大型数据集，使用 `dask_ml` 估计器"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 五分钟浅谈Scikit-Learn\n",
    "\n",
    "Scikit-Learn 有一个很好的、一致的 API。\n",
    "\n",
    "1. 你实例化一个 `Estimator`（例如 `LinearRegression`、`RandomForestClassifier` 等）。 所有模型*超参数*（用户指定的参数，不是估计器学习的参数）在创建时传递给估计器。\n",
    "2. 你调用 `estimator.fit(X, y)` 来训练估计器。\n",
    "3. 使用 `estimator` 来检查属性、进行预测等。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "让我们生成一些随机数据。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from sklearn.datasets import make_classification\n",
    "\n",
    "X, y = make_classification(n_samples=10000, n_features=4, random_state=0)\n",
    "X[:8]"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "y[:8]"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "我们将拟合一个支持向量分类器(SVC)。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from sklearn.svm import SVC"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "创建估算器并拟合它。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "estimator = SVC(random_state=0)\n",
    "estimator.fit(X, y)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "检查学到的属性。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "estimator.support_vectors_[:4]"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "检查准确率。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "estimator.score(X, y)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 超参数"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "大多数模型都有*超参数*。 它们会影响拟合，但他们是预先指定，而不是在训练期间学习。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "estimator = SVC(C=0.00001, shrinking=False, random_state=0)\n",
    "estimator.fit(X, y)\n",
    "estimator.support_vectors_[:4]"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "estimator.score(X, y)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 超参数优化\n",
    "\n",
    "有几种方法可以在训练时学习最佳的*超*参数。 一种是`GridSearchCV`。\n",
    "顾名思义，这会在超参数组合的网格上进行蛮力搜索。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from sklearn.model_selection import GridSearchCV"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "%%time\n",
    "estimator = SVC(gamma='auto', random_state=0, probability=True)\n",
    "param_grid = {\n",
    "    'C': [0.001, 10.0],\n",
    "    'kernel': ['rbf', 'poly'],\n",
    "}\n",
    "\n",
    "grid_search = GridSearchCV(estimator, param_grid, verbose=2, cv=2)\n",
    "grid_search.fit(X, y)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## scikit-learn的单机并行\n",
    "\n",
    "![](images/unmerged_grid_search_graph.svg)\n",
    "\n",
    "通过Joblib，Scikit-Learn具有很好的*单机*并行性。\n",
    "任何可以并行操作的scikit-learn估计器都会公开一个 `n_jobs` 关键字。\n",
    "这控制将使用的CPU内核数。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "%%time\n",
    "grid_search = GridSearchCV(estimator, param_grid, verbose=2, cv=2, n_jobs=-1)\n",
    "grid_search.fit(X, y)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## 使用 Dask 实现多机并行\n",
    "\n",
    "![](images/merged_grid_search_graph.svg)\n",
    "\n",
    "Dask 可以与 scikit-learn（通过 joblib）交互，以便您在 *集群* 上训练模型。\n",
    "\n",
    "如果您在笔记本电脑上运行此程序，将花费相当长的时间，但在此期间 CPU 使用率将令人满意地接近 100%。 要运行得更快，您需要一个分布式集群。 这意味着在调用 `Client` 时放入一些像是\n",
    "\n",
    "```\n",
    "c = Client('tcp://my.scheduler.address:8786')\n",
    "```\n",
    "之类的东西。\n",
    "\n",
    "可以在 [此处](https://docs.dask.org/en/latest/setup/single-distributed.html) 找到有关创建集群的多种方法的详细信息。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "让我们在更大的问题（更多超参数）上尝试一下。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "import joblib\n",
    "import dask.distributed\n",
    "\n",
    "c = dask.distributed.Client()"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "param_grid = {\n",
    "    'C': [0.001, 0.1, 1.0, 2.5, 5, 10.0],\n",
    "    # 取消注释此以在集群上进行更大的网格搜索\n",
    "    # 'kernel': ['rbf', 'poly', 'linear'],\n",
    "    # 'shrinking': [True, False],\n",
    "}\n",
    "\n",
    "grid_search = GridSearchCV(estimator, param_grid, verbose=2, cv=5, n_jobs=-1)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "%%time\n",
    "with joblib.parallel_backend(\"dask\", scatter=[X, y]):\n",
    "    grid_search.fit(X, y)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "grid_search.best_params_, grid_search.best_score_"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "# 大数据集上训练\n",
    "\n",
    "有时您会想要在比内存更大的数据集上进行训练。 `dask-ml` 已经实现了在 dask array和dataframe上运行良好的估计器，这些dataframe可能大于您机器的 RAM。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "import dask.array as da\n",
    "import dask.delayed\n",
    "from sklearn.datasets import make_blobs\n",
    "import numpy as np"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "我们将使用 scikit-learn 在本地创建一个小型（随机）数据集。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "n_centers = 12\n",
    "n_features = 20\n",
    "\n",
    "X_small, y_small = make_blobs(n_samples=1000, centers=n_centers, n_features=n_features, random_state=0)\n",
    "\n",
    "centers = np.zeros((n_centers, n_features))\n",
    "\n",
    "for i in range(n_centers):\n",
    "    centers[i] = X_small[y_small == i].mean(0)\n",
    "    \n",
    "centers[:4]"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "小数据集将成为我们的大型随机数据集的模板。\n",
    "我们将使用 `dask.delayed` 来处理 `sklearn.datasets.make_blobs`，以便在我们的worker上生成实际的数据集。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "n_samples_per_block = 200000\n",
    "n_blocks = 500\n",
    "\n",
    "delayeds = [dask.delayed(make_blobs)(n_samples=n_samples_per_block,\n",
    "                                     centers=centers,\n",
    "                                     n_features=n_features,\n",
    "                                     random_state=i)[0]\n",
    "            for i in range(n_blocks)]\n",
    "arrays = [da.from_delayed(obj, shape=(n_samples_per_block, n_features), dtype=X.dtype)\n",
    "          for obj in delayeds]\n",
    "X = da.concatenate(arrays)\n",
    "X"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "X = X.persist()  # 只在集群上运行这个。"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "Dask-ML 中实现的算法是可扩展的。 它们可以很好地处理大于内存的数据集。\n",
    "\n",
    "它们遵循 scikit-learn API，因此如果您熟悉 scikit-learn，您会对 Dask-ML 感到很熟悉。"
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "from dask_ml.cluster import KMeans"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "clf = KMeans(init_max_iter=3, oversampling_factor=10)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "%time clf.fit(X)"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "clf.labels_"
   ],
   "outputs": [],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "source": [
    "clf.labels_[:10].compute()"
   ],
   "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"
  },
  "nbsphinx": {
   "execute": "never"
  },
  "interpreter": {
   "hash": "3f2e3fb42896b6b47c37375014a159d976fd49965917325c599b06eb4946867a"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}