spacy_multiprocess.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Turbo-charge your spaCy NLP pipeline\n",
    "> Tips and tricks to significantly speed up text processing using multi-core spaCy custom pipelines.\n",
    "\n",
    "Consider you have a large tabular dataset on which you want to apply some non-trivial NLP transformations, such as stopword removal followed by lemmatizing (i.e. reducing to root form) the words in a text. [spaCy](https://spacy.io/usage) is an industrial strength NLP library designed for just such a task.\n",
    "\n",
    "In the example shown below, the [New York Times Kaggle dataset](https://www.kaggle.com/nzalake52/new-york-times-articles) is used to showcase how to significantly speed up a spaCy NLP pipeline. the goal is to take in an article's text, and speedily return a list of lemmas with unnecessary words, called \"stopwords\", removed.\n",
    "\n",
    "Pandas DataFrames provide a convenient interface to work with tabular data of this nature. First, import the necessary modules shown."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import re\n",
    "import pandas as pd\n",
    "import spacy\n",
    "from tqdm.notebook import tqdm\n",
    "tqdm.pandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Input variables\n",
    "The tabular data is stored in a tab-separated file obtained by running the preprocessing notebook `preprocessing.ipynb` on the raw text data from Kaggle and stored in the `data/` directory. A curated stopword file is also provided in this location.\n",
    "\n",
    "Additionally, during initial testing, we can limit the size of the DataFrame being worked on (to around $2000$ samples) for faster execution. For the final run, disable the limit by setting it to zero."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "inputfile = \"data/nytimes.tsv\"\n",
    "stopwordfile = \"data/stopwords/stopwords.txt\"\n",
    "limit = 0"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Load spaCy model\n",
    "For lemmatization, a simple spaCy model can be initialized. Since we will not be doing any specialized tasks such as dependency parsing and named entity recognition in this exercise, these components are disabled.\n",
    "\n",
    "spaCy has a `sentencizer` component that can instead be enabled - this simply performs tokenization and sentence boundary detection, following which lemmas can be extracted as token properties. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "nlp = spacy.load('en_core_web_sm', disable=['tagger', 'parser', 'ner'])\n",
    "nlp.add_pipe(nlp.create_pipe('sentencizer'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A method is defined to read in stopwords from a text file and convert it to a set in Python (for efficient lookup)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_stopwords():\n",
    "    \"Return a set of stopwords read in from a file.\"\n",
    "    with open(stopwordfile) as f:\n",
    "        stopwords = []\n",
    "        for line in f:\n",
    "            stopwords.append(line.strip(\"\\n\"))\n",
    "    # Convert to set for performance\n",
    "    stopwords_set = set(stopwords)\n",
    "    return stopwords_set\n",
    "\n",
    "stopwords = get_stopwords()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Read in New York Times Dataset\n",
    "The pre-processed version of the NYT news dataset is read in as a Pandas DataFrame. The columns are named `date`, `headline` and `content` - the text present in the content column is what will be preprocessed to remove stopwords and generate token lemmas."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "def read_data(inputfile):\n",
    "    \"Read in a tab-separated file with date, headline and news content\"\n",
    "    df = pd.read_csv(inputfile, sep='\\t', header=None,\n",
    "                     names=['date', 'headline', 'content'])\n",
    "    df['date'] = pd.to_datetime(df['date'], format=\"%Y-%m-%d\")\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<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>date</th>\n",
       "      <th>headline</th>\n",
       "      <th>content</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>2016-06-30</td>\n",
       "      <td>washington nationals max scherzer baffles mets...</td>\n",
       "      <td>Stellar pitching kept the Mets afloat in the f...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2016-06-30</td>\n",
       "      <td>mayor de blasios counsel to leave next month t...</td>\n",
       "      <td>Mayor Bill de Blasio’s counsel and chief legal...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>2016-06-30</td>\n",
       "      <td>three men charged in killing of cuomo administ...</td>\n",
       "      <td>In the early morning hours of Labor Day last y...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>2016-06-30</td>\n",
       "      <td>tekserve precursor to the apple store to close...</td>\n",
       "      <td>It was the Apple Store in New York City before...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>2016-06-30</td>\n",
       "      <td>once at michael phelpss feet and still chasing...</td>\n",
       "      <td>The United States Olympic swimming trials are ...</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "        date                                           headline  \\\n",
       "0 2016-06-30  washington nationals max scherzer baffles mets...   \n",
       "1 2016-06-30  mayor de blasios counsel to leave next month t...   \n",
       "2 2016-06-30  three men charged in killing of cuomo administ...   \n",
       "3 2016-06-30  tekserve precursor to the apple store to close...   \n",
       "4 2016-06-30  once at michael phelpss feet and still chasing...   \n",
       "\n",
       "                                             content  \n",
       "0  Stellar pitching kept the Mets afloat in the f...  \n",
       "1  Mayor Bill de Blasio’s counsel and chief legal...  \n",
       "2  In the early morning hours of Labor Day last y...  \n",
       "3  It was the Apple Store in New York City before...  \n",
       "4  The United States Olympic swimming trials are ...  "
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df = read_data(inputfile)\n",
    "df.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define text cleaner\n",
    "Since the news article data comes from a raw HTML dump, it is very messy and contains a host of unnecessary symbols, social media handles, URLs and other artifacts. An easy way to clean it up is to use a regex that parses only alphanumeric strings and hyphens (so as to include hyphenated words) that are between a given length (3 and 50). This filters each document down to only meaningful text for the lemmatizer. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "def cleaner(df):\n",
    "    \"Extract relevant text from DataFrame using a regex\"\n",
    "    # Regex pattern for only alphanumeric, hyphenated text with 3 or more chars\n",
    "    pattern = re.compile(r\"[A-Za-z0-9\\-]{3,50}\")\n",
    "    df['clean'] = df['content'].str.findall(pattern).str.join(' ')\n",
    "    if limit > 0:\n",
    "        return df.iloc[:limit, :].copy()\n",
    "    else:\n",
    "        return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<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>date</th>\n",
       "      <th>headline</th>\n",
       "      <th>content</th>\n",
       "      <th>clean</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>2016-06-30</td>\n",
       "      <td>washington nationals max scherzer baffles mets...</td>\n",
       "      <td>Stellar pitching kept the Mets afloat in the f...</td>\n",
       "      <td>Stellar pitching kept the Mets afloat the firs...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2016-06-30</td>\n",
       "      <td>mayor de blasios counsel to leave next month t...</td>\n",
       "      <td>Mayor Bill de Blasio’s counsel and chief legal...</td>\n",
       "      <td>Mayor Bill Blasio counsel and chief legal advi...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>2016-06-30</td>\n",
       "      <td>three men charged in killing of cuomo administ...</td>\n",
       "      <td>In the early morning hours of Labor Day last y...</td>\n",
       "      <td>the early morning hours Labor Day last year gr...</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "        date                                           headline  \\\n",
       "0 2016-06-30  washington nationals max scherzer baffles mets...   \n",
       "1 2016-06-30  mayor de blasios counsel to leave next month t...   \n",
       "2 2016-06-30  three men charged in killing of cuomo administ...   \n",
       "\n",
       "                                             content  \\\n",
       "0  Stellar pitching kept the Mets afloat in the f...   \n",
       "1  Mayor Bill de Blasio’s counsel and chief legal...   \n",
       "2  In the early morning hours of Labor Day last y...   \n",
       "\n",
       "                                               clean  \n",
       "0  Stellar pitching kept the Mets afloat the firs...  \n",
       "1  Mayor Bill Blasio counsel and chief legal advi...  \n",
       "2  the early morning hours Labor Day last year gr...  "
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_preproc = cleaner(df)\n",
    "df_preproc.head(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now that we have just the clean, alphanumeric tokens left over, these can be further cleaned up by removing stopwords before proceeding to lemmatization."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Option 1. Work directly on the data using `pandas.Series.apply`\n",
    "The straightforward way to process this text is to use an existing method, in this case the `lemmatize` method shown below, and apply it to the `clean` column of the DataFrame. Lemmatization is done using the spaCy's underlying [`Doc` representation](https://spacy.io/usage/spacy-101#annotations) of each token, which contains a `lemma_` property. Stopwords are removed simultaneously with the lemmatization process, as each of these steps involves iterating through the same list of tokens."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "def lemmatize(text):\n",
    "    \"\"\"Perform lemmatization and stopword removal in the clean text\n",
    "       Returns a list of lemmas\n",
    "    \"\"\"\n",
    "    doc = nlp(text)\n",
    "    lemma_list = [str(tok.lemma_).lower() for tok in doc\n",
    "                  if tok.is_alpha and tok.text.lower() not in stopwords]\n",
    "    return lemma_list"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The resulting lemmas as stored as a list in a separate column `preproc` as shown below."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "72d5c0c929b14f28ac2644c7f71fb7ff",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "HBox(children=(FloatProgress(value=0.0, max=8821.0), HTML(value='')))"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "CPU times: user 56.1 s, sys: 441 ms, total: 56.5 s\n",
      "Wall time: 56.1 s\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<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>date</th>\n",
       "      <th>content</th>\n",
       "      <th>preproc</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>2016-06-30</td>\n",
       "      <td>Stellar pitching kept the Mets afloat in the f...</td>\n",
       "      <td>[stellar, pitch, keep, mets, afloat, half, sea...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2016-06-30</td>\n",
       "      <td>Mayor Bill de Blasio’s counsel and chief legal...</td>\n",
       "      <td>[mayor, bill, blasio, counsel, chief, legal, a...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>2016-06-30</td>\n",
       "      <td>In the early morning hours of Labor Day last y...</td>\n",
       "      <td>[early, labor, group, gunman, street, gang, cr...</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "        date                                            content  \\\n",
       "0 2016-06-30  Stellar pitching kept the Mets afloat in the f...   \n",
       "1 2016-06-30  Mayor Bill de Blasio’s counsel and chief legal...   \n",
       "2 2016-06-30  In the early morning hours of Labor Day last y...   \n",
       "\n",
       "                                             preproc  \n",
       "0  [stellar, pitch, keep, mets, afloat, half, sea...  \n",
       "1  [mayor, bill, blasio, counsel, chief, legal, a...  \n",
       "2  [early, labor, group, gunman, street, gang, cr...  "
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "df_preproc['preproc'] = df_preproc['clean'].progress_apply(lemmatize)\n",
    "df_preproc[['date', 'content', 'preproc']].head(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Applying this method to the `clean` column of the DataFrame and timing it shows that it takes almost a minute to run on $8800$ news articles."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Option 2. Use `nlp.pipe`\n",
    "Can we do better? in the [spaCy documentation](https://spacy.io/api/language#pipe), it is stated that \"processing texts as a stream is usually more efficient than processing them one-by-one\". This is done by calling a language pipe, which internally divides the data into batches to reduce the number of pure-Python function calls. This means that the larger the data, the better the performance gain that can be obtained by `nlp.pipe`.\n",
    "\n",
    "To use the language pipe to stream texts, a separate lemmatizer method is defined that directly works on a spaCy `Doc` object. This method is then called in batches to work on a *sequence* of `Doc` objects that are streamed through the pipe as shown below."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "def lemmatize_pipe(doc):\n",
    "    lemma_list = [str(tok.lemma_).lower() for tok in doc\n",
    "                  if tok.is_alpha and tok.text.lower() not in stopwords] \n",
    "    return lemma_list\n",
    "\n",
    "def preprocess_pipe(texts):\n",
    "    preproc_pipe = []\n",
    "    for doc in tqdm(nlp.pipe(texts, batch_size=20), total=len(df_preproc)):\n",
    "        preproc_pipe.append(lemmatize_pipe(doc))\n",
    "    return preproc_pipe"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Just as before, a new column is created by passing data from the `clean` column of the existing DataFrame. Note that unlike in workflow $#1$, we do not use the `apply` method - instead, the column of data (an iterable) is directly passed as an argument to the preprocessor pipe method."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "8fc66f49e9a24cc2bf21e8da29b59093",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "HBox(children=(FloatProgress(value=0.0, max=8821.0), HTML(value='')))"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "CPU times: user 56.8 s, sys: 362 ms, total: 57.1 s\n",
      "Wall time: 56.8 s\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<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>date</th>\n",
       "      <th>content</th>\n",
       "      <th>preproc_pipe</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>2016-06-30</td>\n",
       "      <td>Stellar pitching kept the Mets afloat in the f...</td>\n",
       "      <td>[stellar, pitch, keep, mets, afloat, half, sea...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2016-06-30</td>\n",
       "      <td>Mayor Bill de Blasio’s counsel and chief legal...</td>\n",
       "      <td>[mayor, bill, blasio, counsel, chief, legal, a...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>2016-06-30</td>\n",
       "      <td>In the early morning hours of Labor Day last y...</td>\n",
       "      <td>[early, labor, group, gunman, street, gang, cr...</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "        date                                            content  \\\n",
       "0 2016-06-30  Stellar pitching kept the Mets afloat in the f...   \n",
       "1 2016-06-30  Mayor Bill de Blasio’s counsel and chief legal...   \n",
       "2 2016-06-30  In the early morning hours of Labor Day last y...   \n",
       "\n",
       "                                        preproc_pipe  \n",
       "0  [stellar, pitch, keep, mets, afloat, half, sea...  \n",
       "1  [mayor, bill, blasio, counsel, chief, legal, a...  \n",
       "2  [early, labor, group, gunman, street, gang, cr...  "
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "df_preproc['preproc_pipe'] = preprocess_pipe(df_preproc['clean'])\n",
    "df_preproc[['date', 'content', 'preproc_pipe']].head(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Timing this workflow shows barely any improvement, but it still takes almost a minute on the entire set of $8800$ news articles. One would expect that as we work on bigger and bigger datasets, the timing gain using `nlp.pipe` would become more noticeable (on average)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Option 3. Parallelize the work using joblib\n",
    "We can do still better! The previous workflows sequentially worked through each news document to produce the lemma lists, which were then appended to the DataFrame as a new column. Because each row's output is completely independent of the other, this is an *embarassingly parallel* problem, making it ideal for using multiple cores.\n",
    "\n",
    "The `joblib` library is recommended by spaCy for processing blocks of an NLP pipeline in parallel. Make sure that you `pip install joblib` before running the below section.\n",
    "\n",
    "To parallelize the workflow, a few more helper methods must be defined. \n",
    "\n",
    "* **Chunking:** The news article content is a list of (long) strings where each document represents a single article's text. This data must be fed in \"chunks\" to each worker process started by `joblib`. Each call of the `chunker` method returns a generator that only contains that particular chunk's text as a list of strings. During lemmatization, each new chunk is retrieved based on the iterator index (with the previous chunks being \"forgotten\").\n",
    "\n",
    "\n",
    "* **Flattening:** Once joblib creates a set of worker processes that work on each chunk, each worker returns a \"list of list\" containing lemmas for each document. These lists are then combined by the executor to provide a deeply nested final \"list of list of lists\". To ensure that the length of the output from the executor is the same as the actual number of articles, a \"flatten\" method is defined to combine the result into a list of lists containing lemmas. For example, if the executor returns a final result `[[[a, b, c], [d, e, f]], [[g, h, i], [j, k, l]]]`, a flattened version of this result would be `[[a, b, c], [d, e, f], [g, h, i], [j, k, l]]`.\n",
    "\n",
    "In addition to the above methods, a similar `nlp.pipe` method is used as in workflow $#2$, on each chunk of texts. Each of these methods is wrapped into a `preprocess_parallel` method that defines the number of worker processes to be used ($7$ in this case), breaks the input data into chunks and returns a flattened result that can then be appended to the DataFrame."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "from joblib import Parallel, delayed\n",
    "from functools import partial\n",
    "\n",
    "def chunker(iterable, total_length, chunksize):\n",
    "    return (iterable[pos: pos + chunksize] for pos in range(0, total_length, chunksize))\n",
    "\n",
    "def flatten(list_of_lists):\n",
    "    \"Flatten a list of lists to a combined list\"\n",
    "    return [item for sublist in list_of_lists for item in sublist]\n",
    "\n",
    "def process_chunk(texts):\n",
    "    preproc_pipe = []\n",
    "    for doc in nlp.pipe(texts, batch_size=20):\n",
    "        preproc_pipe.append(lemmatize_pipe(doc))\n",
    "    return preproc_pipe\n",
    "\n",
    "def preprocess_parallel(texts, chunksize=100):\n",
    "    executor = Parallel(n_jobs=7, backend='multiprocessing', prefer=\"processes\")\n",
    "    do = delayed(process_chunk)\n",
    "    tasks = (do(chunk) for chunk in chunker(texts, len(df_preproc), chunksize=chunksize))\n",
    "    result = executor(tasks)\n",
    "    return flatten(result)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 692 ms, sys: 232 ms, total: 925 ms\n",
      "Wall time: 17.6 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "df_preproc['preproc_parallel'] = preprocess_parallel(df_preproc['clean'], chunksize=1000)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<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>date</th>\n",
       "      <th>content</th>\n",
       "      <th>preproc_parallel</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>2016-06-30</td>\n",
       "      <td>Stellar pitching kept the Mets afloat in the f...</td>\n",
       "      <td>[stellar, pitch, keep, mets, afloat, half, sea...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2016-06-30</td>\n",
       "      <td>Mayor Bill de Blasio’s counsel and chief legal...</td>\n",
       "      <td>[mayor, bill, blasio, counsel, chief, legal, a...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>2016-06-30</td>\n",
       "      <td>In the early morning hours of Labor Day last y...</td>\n",
       "      <td>[early, labor, group, gunman, street, gang, cr...</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "        date                                            content  \\\n",
       "0 2016-06-30  Stellar pitching kept the Mets afloat in the f...   \n",
       "1 2016-06-30  Mayor Bill de Blasio’s counsel and chief legal...   \n",
       "2 2016-06-30  In the early morning hours of Labor Day last y...   \n",
       "\n",
       "                                    preproc_parallel  \n",
       "0  [stellar, pitch, keep, mets, afloat, half, sea...  \n",
       "1  [mayor, bill, blasio, counsel, chief, legal, a...  \n",
       "2  [early, labor, group, gunman, street, gang, cr...  "
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_preproc[['date', 'content', 'preproc_parallel']].head(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Timing this parallelized workflow shows significant performance gains (almost **3x** reduction in run time)! As the number of documents becomes larger, the additional overhead of starting multiple worker threads with `joblib` is quickly paid for, and this method can significantly outperform the sequential methods."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Effect of chunk size and batch size\n",
    "Note that in the parallelized workflow, two parameters need to be specified - the optimum number can vary depending on the dataset. The `chunksize` controls the number of chunks being worked on by each process. In this example, for $8800$ documents, a chunksize of $1000$ is used. Too small a chunksize would mean that a large number of worker threads would spawn (each one waiting for other threads to complete), which can slow down execution. Generally, a chunksize of around $1/10^{th}$ of the total number of documents can be used as a starting point (assuming that all chunks fit into memory at any given time).\n",
    "\n",
    "The batch size is parameter specific to `nlp.pipe`, and again, a good value depends on the data being worked on. For reasonably long-sized text such as news articles, it makes sense to keep the batch size reasonably small (so that each batch doesn't contain *really* long texts), so in this case $20$ was chosen for the batch size. For other cases (e.g. Tweets) where each document is much shorter in length, a larger batch size can be used.\n",
    "\n",
    "**It is recommended to experiment with either parameter to see which combination produces the best performance**."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Bonus: Use sets over lists for lookups wherever possible\n",
    "Note that in the `get_stopwords()` method defined earlier on, the list of stopwords read in from the stopword file was converted to a set before using it in the lemmatizer method for stopword removal via lookups. This is a very useful trick in general, but specifically for stopword removal, the use of sets becomes **all the more important**. Why? \n",
    "\n",
    "In any realistic stopword list, such as this one for a news dataset, it's reasonable to expect *several hundred* stopwords. This is because for downstream tasks such as topic modelling or sentiment analysis, there are a number of domain-specific words that need to be removed (very common verbs, useless abbreviations such as timezones, days of the week, etc.). Each word in each and every document needs to be compared against every word in the stopword list, which is an expensive operation over tens of thousands of documents.\n",
    "\n",
    "It's well known that sets have $O(1)$ (i.e. consant) lookup time as opposed to lists, which have $O(n)$ lookup time. In the `lemmatize()` method, since we're checking each word for membership in the set of stopwords, we would expect sets to be much better than lists. To test this, we can rerun workflow $#1$ but this time, use a stopword *list* instead."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "stopwords = list(stopwords)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "ec0e5f0419ea4a82a050fac8512a1561",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "HBox(children=(FloatProgress(value=0.0, max=8821.0), HTML(value='')))"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "CPU times: user 1min 29s, sys: 561 ms, total: 1min 29s\n",
      "Wall time: 1min 29s\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<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>date</th>\n",
       "      <th>content</th>\n",
       "      <th>preproc_stopword_list</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>2016-06-30</td>\n",
       "      <td>Stellar pitching kept the Mets afloat in the f...</td>\n",
       "      <td>[stellar, pitch, keep, mets, afloat, half, sea...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>2016-06-30</td>\n",
       "      <td>Mayor Bill de Blasio’s counsel and chief legal...</td>\n",
       "      <td>[mayor, bill, blasio, counsel, chief, legal, a...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>2016-06-30</td>\n",
       "      <td>In the early morning hours of Labor Day last y...</td>\n",
       "      <td>[early, labor, group, gunman, street, gang, cr...</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "        date                                            content  \\\n",
       "0 2016-06-30  Stellar pitching kept the Mets afloat in the f...   \n",
       "1 2016-06-30  Mayor Bill de Blasio’s counsel and chief legal...   \n",
       "2 2016-06-30  In the early morning hours of Labor Day last y...   \n",
       "\n",
       "                               preproc_stopword_list  \n",
       "0  [stellar, pitch, keep, mets, afloat, half, sea...  \n",
       "1  [mayor, bill, blasio, counsel, chief, legal, a...  \n",
       "2  [early, labor, group, gunman, street, gang, cr...  "
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "df_preproc['preproc_stopword_list'] = df_preproc['clean'].progress_apply(lemmatize)\n",
    "df_preproc[['date', 'content', 'preproc_stopword_list']].head(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This method now takes **58% longer** than it did before (when using a stopword set), which is almost a **1.5x** increase in run time! This makes sense because in this case, the stopword list is about 500 words long, and *each and every word* in the corpus needs to be checked for membership in this reasonable-sized list."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Conclusions\n",
    "In this exercise, a news article dataset (NY Times) was processed using a spaCy pipeline to output a list of lemmas representing the useful tokens present in each article's content. Because real-world news datasets are almost certainly bigger than this one, and can be almost unbounded in size, a fast, efficient NLP pipeline is necessary to perform any meaningful analysis on the data. The following steps are very useful in speeding up the spaCy pipeline.\n",
    "\n",
    "**Disable unnecessary components in spaCy model:** The standard spaCy model's pipeline contains the tagger (to assign part-of-speech tags), the parser (to generate a dependency parse) and named entity recognition components. If any or none of these actions are desired, these components *must* be disabled immediately after loading the model (as shown above).\n",
    "\n",
    "**Use sets over lists for lookups:** When performing lookups to compare one set of tokens against another, always check for membership using sets - lists are significantly slower for lookups! The larger the list/set of stopwords, the larger the performance gain seen when using sets.\n",
    "\n",
    "**Use custom language pipes when possible:** Setting up a language pipe using `nlp.pipe` is an extremely flexible and efficient way to process large blocks of text. Even better, spaCy allows you to individually disable components for each specific sub-task, for example, when we need to separately perform part-of-speech tagging and named entity recognition (NER). [See the spaCy docs](https://spacy.io/usage/processing-pipelines#disabling) for examples on how to disable pipeline components during model loading, processing or handling custom blocks.\n",
    "\n",
    "**Use multiple cores when possible:** When processing individual documents completely independent of one another, consider parallelizing the workflow by passing the computation to multiple cores. As the number of documents becoms higher and higher, the performance gains can be tremendous. All you need to ensure is that the documents are divided up into chunks, all of which must fit into memory at any given time.\n",
    "\n",
    "Test thest out in your next NLP project!"
   ]
  }
 ],
 "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.6.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}