11_Natural_Language_Processing.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Tools for NLP\n",
    "\n",
    "There are lots of feature transformations that need to be done on text data to get it to a point that machine learning algorithms can understand. Luckily, Spark has placed the most important ones in convienent Feature Transformer calls. \n",
    "\n",
    "Let's go over them before jumping into the project."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "# May take a little while on a local computer\n",
    "spark = SparkSession.builder.appName(\"nlp\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# check (try) if Spark session variable (spark) exists and print information about the Spark context\n",
    "try:\n",
    "    spark\n",
    "except NameError:\n",
    "    print(\"Spark session does not context exist. Please create Spark session first (run cell above).\")\n",
    "else:\n",
    "    configurations = spark.sparkContext.getConf().getAll()\n",
    "    for item in configurations: print(item)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Tokenizer\n",
    "<p><a href=\"http://en.wikipedia.org/wiki/Lexical_analysis#Tokenization\">Tokenization</a> is the process of taking text (such as a sentence) and breaking it into individual terms (usually words).  A simple <a href=\"api/scala/index.html#org.apache.spark.ml.feature.Tokenizer\">Tokenizer</a> class provides this functionality.  The example below shows how to split sentences into sequences of words.</p>\n",
    "\n",
    "<p><a href=\"api/scala/index.html#org.apache.spark.ml.feature.RegexTokenizer\">RegexTokenizer</a> allows more\n",
    " advanced tokenization based on regular expression (regex) matching.\n",
    " By default, the parameter &#8220;pattern&#8221; (regex, default: <code>\"\\\\s+\"</code>) is used as delimiters to split the input text.\n",
    " Alternatively, users can set parameter &#8220;gaps&#8221; to false indicating the regex &#8220;pattern&#8221; denotes\n",
    " &#8220;tokens&#8221; rather than splitting gaps, and find all matching occurrences as the tokenization result.</p>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import Tokenizer, RegexTokenizer\n",
    "from pyspark.sql.functions import col, udf\n",
    "from pyspark.sql.types import IntegerType"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sentenceDataFrame = spark.createDataFrame([\n",
    "    (0, \"Hi I heard about Spark\"),\n",
    "    (1, \"I wish Java could use case classes\"),\n",
    "    (2, \"Logistic,regression,models,are,neat\")\n",
    "], [\"id\", \"sentence\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sentenceDataFrame.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tokenizer = Tokenizer(inputCol=\"sentence\", outputCol=\"words\")\n",
    "\n",
    "regexTokenizer = RegexTokenizer(inputCol=\"sentence\", outputCol=\"words\", pattern=\"\\\\W\")\n",
    "# alternatively, pattern=\"\\\\w+\", gaps(False)\n",
    "\n",
    "countTokens = udf(lambda words: len(words), IntegerType())\n",
    "\n",
    "tokenized = tokenizer.transform(sentenceDataFrame)\n",
    "tokenized.select(\"sentence\", \"words\")\\\n",
    "    .withColumn(\"tokens\", countTokens(col(\"words\"))).show(truncate=False)\n",
    "\n",
    "regexTokenized = regexTokenizer.transform(sentenceDataFrame)\n",
    "regexTokenized.select(\"sentence\", \"words\") \\\n",
    "    .withColumn(\"tokens\", countTokens(col(\"words\"))).show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "## Stop Words Removal\n",
    "\n",
    "<p><a href=\"https://en.wikipedia.org/wiki/Stop_words\">Stop words</a> are words which\n",
    "should be excluded from the input, typically because the words appear\n",
    "frequently and don&#8217;t carry as much meaning.</p>\n",
    "\n",
    "<p><code>StopWordsRemover</code> takes as input a sequence of strings (e.g. the output\n",
    "of a <a href=\"ml-features.html#tokenizer\">Tokenizer</a>) and drops all the stop\n",
    "words from the input sequences. The list of stopwords is specified by\n",
    "the <code>stopWords</code> parameter. Default stop words for some languages are accessible \n",
    "by calling <code>StopWordsRemover.loadDefaultStopWords(language)</code>, for which available \n",
    "options are &#8220;danish&#8221;, &#8220;dutch&#8221;, &#8220;english&#8221;, &#8220;finnish&#8221;, &#8220;french&#8221;, &#8220;german&#8221;, &#8220;hungarian&#8221;, \n",
    "&#8220;italian&#8221;, &#8220;norwegian&#8221;, &#8220;portuguese&#8221;, &#8220;russian&#8221;, &#8220;spanish&#8221;, &#8220;swedish&#8221; and &#8220;turkish&#8221;. \n",
    "A boolean parameter <code>caseSensitive</code> indicates if the matches should be case sensitive \n",
    "(false by default).</p>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import StopWordsRemover\n",
    "\n",
    "sentenceData = spark.createDataFrame([\n",
    "    (0, [\"I\", \"saw\", \"the\", \"red\", \"balloon\"]),\n",
    "    (1, [\"Mary\", \"had\", \"a\", \"little\", \"lamb\"])\n",
    "], [\"id\", \"raw\"])\n",
    "\n",
    "remover = StopWordsRemover(inputCol=\"raw\", outputCol=\"filtered\")\n",
    "remover.transform(sentenceData).show(truncate=False)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## n-grams\n",
    "\n",
    "An n-gram is a sequence of nn tokens (typically words) for some integer nn. The NGram class can be used to transform input features into nn-grams.\n",
    "\n",
    "<p><code>NGram</code> takes as input a sequence of strings (e.g. the output of a <a href=\"ml-features.html#tokenizer\">Tokenizer</a>).  The parameter <code>n</code> is used to determine the number of terms in each $n$-gram. The output will consist of a sequence of $n$-grams where each $n$-gram is represented by a space-delimited string of $n$ consecutive words.  If the input sequence contains fewer than <code>n</code> strings, no output is produced.</p>\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import NGram\n",
    "\n",
    "wordDataFrame = spark.createDataFrame([\n",
    "    (0, [\"Hi\", \"I\", \"heard\", \"about\", \"Spark\"]),\n",
    "    (1, [\"I\", \"wish\", \"Java\", \"could\", \"use\", \"case\", \"classes\"]),\n",
    "    (2, [\"Logistic\", \"regression\", \"models\", \"are\", \"neat\"])\n",
    "], [\"id\", \"words\"])\n",
    "\n",
    "ngram = NGram(n=2, inputCol=\"words\", outputCol=\"ngrams\")\n",
    "\n",
    "ngramDataFrame = ngram.transform(wordDataFrame)\n",
    "ngramDataFrame.select(\"ngrams\").show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "_______\n",
    "# Feature Extractors\n",
    "_______"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<h2 id=\"tf-idf\">TF-IDF</h2>\n",
    "\n",
    "<p><a href=\"http://en.wikipedia.org/wiki/Tf%E2%80%93idf\">Term frequency-inverse document frequency (TF-IDF)</a> \n",
    "is a feature vectorization method widely used in text mining to reflect the importance of a term \n",
    "to a document in the corpus. Denote a term by <code>$t$</code>, a document by  d , and the corpus by D.\n",
    "Term frequency <code>$TF(t, d)$</code> is the number of times that term <code>$t$</code> appears in document <code>$d$</code>, while \n",
    "document frequency <code>$DF(t, D)$</code> is the number of documents that contains term <code>$t$</code>. If we only use \n",
    "term frequency to measure the importance, it is very easy to over-emphasize terms that appear very \n",
    "often but carry little information about the document, e.g. &#8220;a&#8221;, &#8220;the&#8221;, and &#8220;of&#8221;. If a term appears \n",
    "very often across the corpus, it means it doesn&#8217;t carry special information about a particular document.\n",
    "Inverse document frequency is a numerical measure of how much information a term provides:\n",
    "\n",
    "$$ IDF(t, D) = \\log \\frac{|D| + 1}{DF(t, D) + 1} $$\n",
    "\n",
    "where |D| is the total number of documents in the corpus. Since logarithm is used, if a term \n",
    "appears in all documents, its IDF value becomes 0. Note that a smoothing term is applied to avoid \n",
    "dividing by zero for terms outside the corpus. The TF-IDF measure is simply the product of TF and IDF:\n",
    "$$ TFIDF(t, d, D) = TF(t, d) \\cdot IDF(t, D). $$\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import HashingTF, IDF, Tokenizer\n",
    "\n",
    "sentenceData = spark.createDataFrame([\n",
    "    (0.0, \"Hi I heard about Spark\"),\n",
    "    (0.0, \"I wish Java could use case classes\"),\n",
    "    (1.0, \"Logistic regression models are neat\")\n",
    "], [\"label\", \"sentence\"])\n",
    "\n",
    "sentenceData.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tokenizer = Tokenizer(inputCol=\"sentence\", outputCol=\"words\")\n",
    "wordsData = tokenizer.transform(sentenceData)\n",
    "wordsData.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "hashingTF = HashingTF(inputCol=\"words\", outputCol=\"rawFeatures\", numFeatures=20)\n",
    "featurizedData = hashingTF.transform(wordsData)\n",
    "# alternatively, CountVectorizer can also be used to get term frequency vectors\n",
    "\n",
    "idf = IDF(inputCol=\"rawFeatures\", outputCol=\"features\")\n",
    "idfModel = idf.fit(featurizedData)\n",
    "rescaledData = idfModel.transform(featurizedData)\n",
    "\n",
    "rescaledData.select(\"label\", \"features\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## CountVectorizer\n",
    "CountVectorizer and CountVectorizerModel aim to help convert a collection of text documents to vectors of token counts. When an a-priori dictionary is not available, CountVectorizer can be used as an Estimator to extract the vocabulary, and generates a CountVectorizerModel. The model produces sparse representations for the documents over the vocabulary, which can then be passed to other algorithms like LDA.\n",
    "\n",
    "During the fitting process, CountVectorizer will select the top vocabSize words ordered by term frequency across the corpus. An optional parameter minDF also affects the fitting process by specifying the minimum number (or fraction if < 1.0) of documents a term must appear in to be included in the vocabulary. Another optional binary toggle parameter controls the output vector. If set to true all nonzero counts are set to 1. This is especially useful for discrete probabilistic models that model binary, rather than integer, counts."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import CountVectorizer\n",
    "\n",
    "# Input data: Each row is a bag of words with a ID.\n",
    "df = spark.createDataFrame([\n",
    "    (0, \"a b c\".split(\" \")),\n",
    "    (1, \"a b b c a\".split(\" \"))\n",
    "], [\"id\", \"words\"])\n",
    "\n",
    "# fit a CountVectorizerModel from the corpus.\n",
    "cv = CountVectorizer(inputCol=\"words\", outputCol=\"features\", vocabSize=3, minDF=2.0)\n",
    "\n",
    "model = cv.fit(df)\n",
    "\n",
    "result = model.transform(df)\n",
    "result.show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# NLP Code Along\n",
    "\n",
    "For this code along we will build a spam filter! We'll use the various NLP tools we learned about as well as a new classifier, Naive Bayes.\n",
    "\n",
    "We'll use a classic dataset for this - UCI Repository SMS Spam Detection: https://archive.ics.uci.edu/ml/datasets/SMS+Spam+Collection"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = spark.read.csv(\"data/smsspamcollection/SMSSpamCollection\",inferSchema=True,sep='\\t')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = data.withColumnRenamed('_c0','class').withColumnRenamed('_c1','text')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Clean and Prepare the Data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Create a new length feature:**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import length"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = data.withColumn('length',length(data['text']))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Pretty Clear Difference\n",
    "data.groupby('class').mean().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Feature Transformations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import Tokenizer,StopWordsRemover, CountVectorizer,IDF,StringIndexer\n",
    "\n",
    "tokenizer = Tokenizer(inputCol=\"text\", outputCol=\"token_text\")\n",
    "stopremove = StopWordsRemover(inputCol='token_text',outputCol='stop_tokens')\n",
    "count_vec = CountVectorizer(inputCol='stop_tokens',outputCol='c_vec')\n",
    "idf = IDF(inputCol=\"c_vec\", outputCol=\"tf_idf\")\n",
    "ham_spam_to_num = StringIndexer(inputCol='class',outputCol='label')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import VectorAssembler\n",
    "from pyspark.ml.linalg import Vector"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "clean_up = VectorAssembler(inputCols=['tf_idf','length'],outputCol='features')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### The Model\n",
    "\n",
    "We'll use Naive Bayes, but feel free to play around with this choice!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.classification import NaiveBayes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Use defaults\n",
    "nb = NaiveBayes()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml import Pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data_prep_pipe = Pipeline(stages=[ham_spam_to_num,tokenizer,stopremove,count_vec,idf,clean_up])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cleaner = data_prep_pipe.fit(data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "clean_data = cleaner.transform(data)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Training and Evaluation!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "clean_data = clean_data.select(['label','features'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "clean_data.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "(training,testing) = clean_data.randomSplit([0.7,0.3])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spam_predictor = nb.fit(training)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "data.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "test_results = spam_predictor.transform(testing)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_results.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.ml.evaluation import MulticlassClassificationEvaluator"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "acc_eval = MulticlassClassificationEvaluator()\n",
    "acc = acc_eval.evaluate(test_results)\n",
    "print(\"Accuracy of model at predicting spam was: {}\".format(acc))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Not bad considering we're using straight math on text data! Try switching out the classification models! Or even try to come up with other engineered features!\n",
    "\n",
    "## Great Job!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Stop The Spark Session"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# stop the underlying SparkContext.\n",
    "try:\n",
    "    spark\n",
    "except NameError:\n",
    "    print(\"Spark session does not context exist - nothing to stop.\")\n",
    "else:\n",
    "    spark.stop()"
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "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.9.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}