{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Text Vectorization Pipeline\n",
    "\n",
    "This example illustrates how Dask-ML can be used to classify large textual datasets in parallel.\n",
    "It is adapted from [this scikit-learn example](https://scikit-learn.org/stable/auto_examples/applications/plot_out_of_core_classification.html#sphx-glr-auto-examples-applications-plot-out-of-core-classification-py).\n",
    "\n",
    "The primary differences are that\n",
    "\n",
    "* We fit the entire model, including text vectorization, as a pipeline.\n",
    "* We use dask collections like [Dask Bag](https://docs.dask.org/en/latest/bag.html), [Dask Dataframe](https://docs.dask.org/en/latest/dataframe.html), and [Dask Array](https://docs.dask.org/en/latest/array.html)\n",
    "  rather than generators to work with larger than memory datasets."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-14T16:50:49.117996Z",
     "iopub.status.busy": "2021-10-14T16:50:49.117107Z",
     "iopub.status.idle": "2021-10-14T16:50:51.354540Z",
     "shell.execute_reply": "2021-10-14T16:50:51.355525Z"
    }
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/share/miniconda3/envs/dask-examples/lib/python3.8/site-packages/distributed/node.py:151: UserWarning: Port 8787 is already in use.\n",
      "Perhaps you already have a cluster running?\n",
      "Hosting the HTTP server on port 37341 instead\n",
      "  warnings.warn(\n"
     ]
    },
    {
     "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:45301</li>\n",
       "  <li><b>Dashboard: </b><a href='http://127.0.0.1:37341/status' target='_blank'>http://127.0.0.1:37341/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>2</li>\n",
       "  <li><b>Cores: </b>4</li>\n",
       "  <li><b>Memory: </b>4.00 GB</li>\n",
       "</ul>\n",
       "</td>\n",
       "</tr>\n",
       "</table>"
      ],
      "text/plain": [
       "<Client: 'tcp://127.0.0.1:45301' processes=2 threads=4, memory=4.00 GB>"
      ]
     },
     "execution_count": 1,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from dask.distributed import Client, progress\n",
    "\n",
    "client = Client(n_workers=2, threads_per_worker=2, memory_limit='2GB')\n",
    "client"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Fetch the data\n",
    "\n",
    "Scikit-Learn provides a utility to fetch the newsgroups dataset."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-14T16:50:51.359296Z",
     "iopub.status.busy": "2021-10-14T16:50:51.358210Z",
     "iopub.status.idle": "2021-10-14T16:50:52.065317Z",
     "shell.execute_reply": "2021-10-14T16:50:52.065743Z"
    }
   },
   "outputs": [],
   "source": [
    "import sklearn.datasets\n",
    "\n",
    "bunch = sklearn.datasets.fetch_20newsgroups()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The data from scikit-learn isn't *too* large, so the data is just\n",
    "returned in memory. Each document is a string. The target we're predicting\n",
    "is an integer, which codes the topic of the post.\n",
    "\n",
    "We'll load the documents and targets directly into a dask DataFrame.\n",
    "In practice, on a larger than memory dataset, you would likely load the\n",
    "documents from disk or cloud storage using `dask.bag` or `dask.delayed`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-14T16:50:52.078277Z",
     "iopub.status.busy": "2021-10-14T16:50:52.074108Z",
     "iopub.status.idle": "2021-10-14T16:50:52.596350Z",
     "shell.execute_reply": "2021-10-14T16:50:52.597433Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div><strong>Dask DataFrame Structure:</strong></div>\n",
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>text</th>\n",
       "      <th>target</th>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>npartitions=25</th>\n",
       "      <th></th>\n",
       "      <th></th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>object</td>\n",
       "      <td>int64</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>453</th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>...</th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>10872</th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>11313</th>\n",
       "      <td>...</td>\n",
       "      <td>...</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>\n",
       "<div>Dask Name: from_pandas, 25 tasks</div>"
      ],
      "text/plain": [
       "Dask DataFrame Structure:\n",
       "                  text target\n",
       "npartitions=25               \n",
       "0               object  int64\n",
       "453                ...    ...\n",
       "...                ...    ...\n",
       "10872              ...    ...\n",
       "11313              ...    ...\n",
       "Dask Name: from_pandas, 25 tasks"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import dask.dataframe as dd\n",
    "import pandas as pd\n",
    "\n",
    "df = dd.from_pandas(pd.DataFrame({\"text\": bunch.data, \"target\": bunch.target}),\n",
    "                    npartitions=25)\n",
    "\n",
    "df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Each row in the `text` column has a bit of metadata and the full text of a post."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-14T16:50:52.604269Z",
     "iopub.status.busy": "2021-10-14T16:50:52.602844Z",
     "iopub.status.idle": "2021-10-14T16:50:53.247146Z",
     "shell.execute_reply": "2021-10-14T16:50:53.245686Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "From: lerxst@wam.umd.edu (where's my thing)\n",
      "Subject: WHAT car is this!?\n",
      "Nntp-Posting-Host: rac3.wam.umd.edu\n",
      "Organization: University of Maryland, College Park\n",
      "Lines: 15\n",
      "\n",
      " I was wondering if anyone out there could enlighten me on this car I saw\n",
      "the other day. It was a 2-door sports car, looked to be from the late 60s/\n",
      "early 70s. It was called a Bricklin. The doors were really small. In addition,\n",
      "the front bumper was separate from the rest of the body. This is \n",
      "all I know. If anyone can tellme a m\n"
     ]
    }
   ],
   "source": [
    "print(df.head().loc[0, 'text'][:500])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Feature Hashing"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Dask's [HashingVectorizer](https://ml.dask.org/modules/generated/dask_ml.feature_extraction.text.HashingVectorizer.html#dask_ml.feature_extraction.text.HashingVectorizer) provides a similar API to [scikit-learn's implementation](https://scikit-learn.org/stable/modules/generated/sklearn.feature_extraction.text.HashingVectorizer.html). In fact, Dask-ML's implementation uses scikit-learn's, applying it to each partition of the input `dask.dataframe.Series` or `dask.bag.Bag`.\n",
    "\n",
    "Transformation, once we actually compute the result, happens in parallel and returns a dask Array."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-14T16:50:53.251771Z",
     "iopub.status.busy": "2021-10-14T16:50:53.251260Z",
     "iopub.status.idle": "2021-10-14T16:50:53.336716Z",
     "shell.execute_reply": "2021-10-14T16:50:53.337593Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr>\n",
       "<td>\n",
       "<table>\n",
       "  <thead>\n",
       "    <tr><td> </td><th> Array </th><th> Chunk </th></tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "\n",
       "    <tr><th> Shape </th><td> (nan, 1048576) </td> <td> (nan, 1048576) </td></tr>\n",
       "    <tr><th> Count </th><td> 75 Tasks </td><td> 25 Chunks </td></tr>\n",
       "    <tr><th> Type </th><td> float64 </td><td> scipy.csr_matrix </td></tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</td>\n",
       "<td>\n",
       "\n",
       "</td>\n",
       "</tr>\n",
       "</table>"
      ],
      "text/plain": [
       "dask.array<_transformer, shape=(nan, 1048576), dtype=float64, chunksize=(nan, 1048576), chunktype=scipy.csr_matrix>"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import dask_ml.feature_extraction.text\n",
    "\n",
    "vect = dask_ml.feature_extraction.text.HashingVectorizer()\n",
    "X = vect.fit_transform(df['text'])\n",
    "X"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The output array `X` has unknown chunk sizes becase the input dask Series or Bags don't know their own length.\n",
    "\n",
    "Each block in `X` is a `scipy.sparse` matrix."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-14T16:50:53.344174Z",
     "iopub.status.busy": "2021-10-14T16:50:53.342782Z",
     "iopub.status.idle": "2021-10-14T16:50:54.057525Z",
     "shell.execute_reply": "2021-10-14T16:50:54.058035Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<453x1048576 sparse matrix of type '<class 'numpy.float64'>'\n",
       "\twith 64357 stored elements in Compressed Sparse Row format>"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "X.blocks[0].compute()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This is a document-term matrix. Each row is the hashed representation of the original post."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Classification Pipeline\n",
    "\n",
    "We can combine the [HashingVectorizer](https://ml.dask.org/modules/generated/dask_ml.feature_extraction.text.HashingVectorizer.html#dask_ml.feature_extraction.text.HashingVectorizer) with [Incremental](https://ml.dask.org/modules/generated/dask_ml.wrappers.Incremental.html#dask_ml.wrappers.Incremental) and a classifier like scikit-learn's `SGDClassifier` to\n",
    "create a classification pipeline.\n",
    "\n",
    "We'll predict whether the topic was in the `comp` category."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-14T16:50:54.064315Z",
     "iopub.status.busy": "2021-10-14T16:50:54.063759Z",
     "iopub.status.idle": "2021-10-14T16:50:54.071134Z",
     "shell.execute_reply": "2021-10-14T16:50:54.070424Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['alt.atheism',\n",
       " 'comp.graphics',\n",
       " 'comp.os.ms-windows.misc',\n",
       " 'comp.sys.ibm.pc.hardware',\n",
       " 'comp.sys.mac.hardware',\n",
       " 'comp.windows.x',\n",
       " 'misc.forsale',\n",
       " 'rec.autos',\n",
       " 'rec.motorcycles',\n",
       " 'rec.sport.baseball',\n",
       " 'rec.sport.hockey',\n",
       " 'sci.crypt',\n",
       " 'sci.electronics',\n",
       " 'sci.med',\n",
       " 'sci.space',\n",
       " 'soc.religion.christian',\n",
       " 'talk.politics.guns',\n",
       " 'talk.politics.mideast',\n",
       " 'talk.politics.misc',\n",
       " 'talk.religion.misc']"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "bunch.target_names"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-14T16:50:54.078592Z",
     "iopub.status.busy": "2021-10-14T16:50:54.078045Z",
     "iopub.status.idle": "2021-10-14T16:50:54.089641Z",
     "shell.execute_reply": "2021-10-14T16:50:54.090701Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Dask Series Structure:\n",
       "npartitions=25\n",
       "0        int64\n",
       "453        ...\n",
       "         ...  \n",
       "10872      ...\n",
       "11313      ...\n",
       "Name: target, dtype: int64\n",
       "Dask Name: astype, 101 tasks"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import numpy as np\n",
    "\n",
    "positive = np.arange(len(bunch.target_names))[['comp' in x for x in bunch.target_names]]\n",
    "y = df['target'].isin(positive).astype(int)\n",
    "y"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-14T16:50:54.094399Z",
     "iopub.status.busy": "2021-10-14T16:50:54.093146Z",
     "iopub.status.idle": "2021-10-14T16:50:54.098083Z",
     "shell.execute_reply": "2021-10-14T16:50:54.098841Z"
    }
   },
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import sklearn.linear_model\n",
    "import sklearn.pipeline\n",
    "\n",
    "import dask_ml.wrappers"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Because the input comes from a dask Series, with unknown chunk sizes, we need to specify `assume_equal_chunks=True`. This tells Dask-ML that we know that each partition in `X`\n",
    "matches a partition in `y`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-14T16:50:54.102701Z",
     "iopub.status.busy": "2021-10-14T16:50:54.101598Z",
     "iopub.status.idle": "2021-10-14T16:50:54.107310Z",
     "shell.execute_reply": "2021-10-14T16:50:54.108114Z"
    }
   },
   "outputs": [],
   "source": [
    "sgd = sklearn.linear_model.SGDClassifier(\n",
    "    tol=1e-3\n",
    ")\n",
    "clf = dask_ml.wrappers.Incremental(\n",
    "    sgd, scoring='accuracy', assume_equal_chunks=True\n",
    ")\n",
    "pipe = sklearn.pipeline.make_pipeline(vect, clf)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`SGDClassifier.partial_fit` needs to know the full set of classes up front.\n",
    "Because our `sgd` is wrapped inside an `Incremental`, we need to pass it through\n",
    "as the `incremental__classes` keyword argument in `fit`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-14T16:50:54.111882Z",
     "iopub.status.busy": "2021-10-14T16:50:54.110657Z",
     "iopub.status.idle": "2021-10-14T16:50:58.755197Z",
     "shell.execute_reply": "2021-10-14T16:50:58.754436Z"
    }
   },
   "outputs": [],
   "source": [
    "pipe.fit(df['text'], y,\n",
    "         incremental__classes=[0, 1]);"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As usual, `Incremental.predict` lazily returns the predictions as a dask Array."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-14T16:50:58.760065Z",
     "iopub.status.busy": "2021-10-14T16:50:58.759485Z",
     "iopub.status.idle": "2021-10-14T16:50:58.774701Z",
     "shell.execute_reply": "2021-10-14T16:50:58.775492Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<table>\n",
       "<tr>\n",
       "<td>\n",
       "<table>\n",
       "  <thead>\n",
       "    <tr><td> </td><th> Array </th><th> Chunk </th></tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr><th> Bytes </th><td> unknown </td> <td> unknown </td></tr>\n",
       "    <tr><th> Shape </th><td> (nan,) </td> <td> (nan,) </td></tr>\n",
       "    <tr><th> Count </th><td> 100 Tasks </td><td> 25 Chunks </td></tr>\n",
       "    <tr><th> Type </th><td> int64 </td><td> numpy.ndarray </td></tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</td>\n",
       "<td>\n",
       "\n",
       "</td>\n",
       "</tr>\n",
       "</table>"
      ],
      "text/plain": [
       "dask.array<_predict, shape=(nan,), dtype=int64, chunksize=(nan,), chunktype=numpy.ndarray>"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "predictions = pipe.predict(df['text'])\n",
    "predictions"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can compute the predictions and score in parallel with `dask_ml.metrics.accuracy_score`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "execution": {
     "iopub.execute_input": "2021-10-14T16:50:58.779419Z",
     "iopub.status.busy": "2021-10-14T16:50:58.778203Z",
     "iopub.status.idle": "2021-10-14T16:51:02.043205Z",
     "shell.execute_reply": "2021-10-14T16:51:02.044230Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.9642920275764539"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "dask_ml.metrics.accuracy_score(y, predictions)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This simple combination of a HashingVectorizer and SGDClassifier is\n",
    "pretty effective at this prediction task."
   ]
  }
 ],
 "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.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}