Spark.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "# Introduction to Spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this notebook, you will be introduced to the Apache Spark libary for big data processing. There's a now a python package called `pyspark` which will load Spark for you. Tthe variable `sc` is a Spark context that lets you interact with the Spark runtime. Check that it is correctly initialised:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style scoped>\n",
       "  .ansiout {\n",
       "    display: block;\n",
       "    unicode-bidi: embed;\n",
       "    white-space: pre-wrap;\n",
       "    word-wrap: break-word;\n",
       "    word-break: break-all;\n",
       "    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n",
       "    font-size: 13px;\n",
       "    color: #555;\n",
       "    margin-left: 4px;\n",
       "    line-height: 19px;\n",
       "  }\n",
       "</style>\n",
       "<div class=\"ansiout\">&lt;SparkContext master=local[8] appName=Databricks Shell&gt;\n",
       "Ready to go!\n",
       "</div>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "from pyspark import SparkContext\n",
    "sc = SparkContext.getOrCreate()\n",
    "print(sc)\n",
    "print(\"Ready to go!\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Learning activity: Create RDDs"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To analyse large datasets using Spark you will load them into Resilient Distributed Datasets (RDDs). There are a number of ways in which you can create RDDs. Use the `parallelize()` function to create one from a Python collection of elements [\"Hello\", \"World\", \"!\"]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style scoped>\n",
       "  .ansiout {\n",
       "    display: block;\n",
       "    unicode-bidi: embed;\n",
       "    white-space: pre-wrap;\n",
       "    word-wrap: break-word;\n",
       "    word-break: break-all;\n",
       "    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n",
       "    font-size: 13px;\n",
       "    color: #555;\n",
       "    margin-left: 4px;\n",
       "    line-height: 19px;\n",
       "  }\n",
       "</style>\n",
       "<div class=\"ansiout\"><span class=\"ansired\">Out[</span><span class=\"ansired\">2</span><span class=\"ansired\">]: </span>ParallelCollectionRDD[0] at parallelize at PythonRDD.scala:267\n",
       "</div>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "sc.parallelize(['Hello', 'World', '!'])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Use the `textFile()` function to create an RDD from the file `war-and-peace.txt` and store it in a variable called `lines`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "ename": "NameError",
     "evalue": "name 'sc' is not defined",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mNameError\u001b[0m                                 Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-1-629e20971d5b>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[0;32m----> 1\u001b[0;31m \u001b[0mlines\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0msc\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtextFile\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m'/FileStore/tables/war_and_peace-14a2f.txt'\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m      2\u001b[0m \u001b[0;31m# use line of code below if the notebook was run locally\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mNameError\u001b[0m: name 'sc' is not defined"
     ]
    }
   ],
   "source": [
    "lines = sc.textFile('/FileStore/tables/war_and_peace-14a2f.txt')\n",
    "\n",
    "# use line of code below if the notebook was run locally\n",
    "lines = sc.textFile('war_and_peace.txt')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Learning activity: Basic RDD manipulation"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Print the number of lines in War and Peace using the functions `count()`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style scoped>\n",
       "  .ansiout {\n",
       "    display: block;\n",
       "    unicode-bidi: embed;\n",
       "    white-space: pre-wrap;\n",
       "    word-wrap: break-word;\n",
       "    word-break: break-all;\n",
       "    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n",
       "    font-size: 13px;\n",
       "    color: #555;\n",
       "    margin-left: 4px;\n",
       "    line-height: 19px;\n",
       "  }\n",
       "</style>\n",
       "<div class=\"ansiout\"><span class=\"ansired\">Out[</span><span class=\"ansired\">6</span><span class=\"ansired\">]: </span>54223\n",
       "</div>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "lines.count()\n",
    "#return integer"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Print the first 15 lines using the functions `take()`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style scoped>\n",
       "  .ansiout {\n",
       "    display: block;\n",
       "    unicode-bidi: embed;\n",
       "    white-space: pre-wrap;\n",
       "    word-wrap: break-word;\n",
       "    word-break: break-all;\n",
       "    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n",
       "    font-size: 13px;\n",
       "    color: #555;\n",
       "    margin-left: 4px;\n",
       "    line-height: 19px;\n",
       "  }\n",
       "</style>\n",
       "<div class=\"ansiout\"><span class=\"ansired\">Out[</span><span class=\"ansired\">7</span><span class=\"ansired\">]: </span>\n",
       "[&apos;                                      1869&apos;,\n",
       " &apos;                                 WAR AND PEACE&apos;,\n",
       " &apos;                                 by Leo Tolstoy&apos;,\n",
       " &apos;BK1&apos;,\n",
       " &apos;                                 BOOK ONE: 1805&apos;,\n",
       " &apos;BK1|CH1&apos;,\n",
       " &apos;  CHAPTER I&apos;,\n",
       " &apos;&apos;,\n",
       " &apos;  &quot;Well, Prince, so Genoa and Lucca are now just family estates of the&apos;,\n",
       " &quot;Buonapartes. But I warn you, if you don&apos;t tell me that this means war,&quot;,\n",
       " &apos;if you still try to defend the infamies and horrors perpetrated by&apos;,\n",
       " &apos;that Antichrist- I really believe he is Antichrist- I will have&apos;,\n",
       " &apos;nothing more to do with you and you are no longer my friend, no longer&apos;,\n",
       " &quot;my &apos;faithful slave,&apos; as you call yourself! But how do you do? I see&quot;,\n",
       " &apos;I have frightened you- sit down and tell me all the news.&quot;&apos;]\n",
       "</div>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "lines.take(15)\n",
    "#return a list"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Learning activity: `filter()` and `map()` and `distinct()`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Lets apply some transformations onto RDDs. The following helper function will be useful to select the words from a line."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style scoped>\n",
       "  .ansiout {\n",
       "    display: block;\n",
       "    unicode-bidi: embed;\n",
       "    white-space: pre-wrap;\n",
       "    word-wrap: break-word;\n",
       "    word-break: break-all;\n",
       "    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n",
       "    font-size: 13px;\n",
       "    color: #555;\n",
       "    margin-left: 4px;\n",
       "    line-height: 19px;\n",
       "  }\n",
       "</style>\n",
       "<div class=\"ansiout\">[&apos;This&apos;, &apos;is&apos;, &apos;a&apos;, &apos;test&apos;]\n",
       "</div>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# A helper function to compute the list of words in a line of text\n",
    "import re\n",
    "def get_words(line):\n",
    "    return re.compile('\\w+').findall(line)\n",
    "\n",
    "print(get_words(\"This, is a test!\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Use `filter()` to count the number of lines which mention `war` and the number of lines which mention `peace`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style scoped>\n",
       "  .ansiout {\n",
       "    display: block;\n",
       "    unicode-bidi: embed;\n",
       "    white-space: pre-wrap;\n",
       "    word-wrap: break-word;\n",
       "    word-break: break-all;\n",
       "    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n",
       "    font-size: 13px;\n",
       "    color: #555;\n",
       "    margin-left: 4px;\n",
       "    line-height: 19px;\n",
       "  }\n",
       "</style>\n",
       "<div class=\"ansiout\">265\n",
       "104\n",
       "</div>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# How often are war and peace mentioned?\n",
    "\n",
    "#if doesn't contain war, it returns false and get ignored\n",
    "#the get_words function avoid the counting of word warning in such a war\n",
    "war_lines = lines.filter(lambda l: 'war' in get_words(l))\n",
    "print(war_lines.count())\n",
    "\n",
    "peace_lines = lines.filter(lambda l: 'peace' in get_words(l))\n",
    "print(peace_lines.count())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Use `map()` to capitalise each line in the RDD, and print the first 15 capitalized lines."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style scoped>\n",
       "  .ansiout {\n",
       "    display: block;\n",
       "    unicode-bidi: embed;\n",
       "    white-space: pre-wrap;\n",
       "    word-wrap: break-word;\n",
       "    word-break: break-all;\n",
       "    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n",
       "    font-size: 13px;\n",
       "    color: #555;\n",
       "    margin-left: 4px;\n",
       "    line-height: 19px;\n",
       "  }\n",
       "</style>\n",
       "<div class=\"ansiout\"><span class=\"ansired\">Out[</span><span class=\"ansired\">12</span><span class=\"ansired\">]: </span>\n",
       "[&apos;                                      1869&apos;,\n",
       " &apos;                                 WAR AND PEACE&apos;,\n",
       " &apos;                                 BY LEO TOLSTOY&apos;,\n",
       " &apos;BK1&apos;,\n",
       " &apos;                                 BOOK ONE: 1805&apos;,\n",
       " &apos;BK1|CH1&apos;,\n",
       " &apos;  CHAPTER I&apos;,\n",
       " &apos;&apos;,\n",
       " &apos;  &quot;WELL, PRINCE, SO GENOA AND LUCCA ARE NOW JUST FAMILY ESTATES OF THE&apos;,\n",
       " &quot;BUONAPARTES. BUT I WARN YOU, IF YOU DON&apos;T TELL ME THAT THIS MEANS WAR,&quot;,\n",
       " &apos;IF YOU STILL TRY TO DEFEND THE INFAMIES AND HORRORS PERPETRATED BY&apos;,\n",
       " &apos;THAT ANTICHRIST- I REALLY BELIEVE HE IS ANTICHRIST- I WILL HAVE&apos;,\n",
       " &apos;NOTHING MORE TO DO WITH YOU AND YOU ARE NO LONGER MY FRIEND, NO LONGER&apos;,\n",
       " &quot;MY &apos;FAITHFUL SLAVE,&apos; AS YOU CALL YOURSELF! BUT HOW DO YOU DO? I SEE&quot;,\n",
       " &apos;I HAVE FRIGHTENED YOU- SIT DOWN AND TELL ME ALL THE NEWS.&quot;&apos;]\n",
       "</div>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Capitalize each line in the RDD\n",
    "\n",
    "#instead of using for loop, the lambda function can be used to call the function parallely\n",
    "cap_lines = lines.map(lambda l: l.upper())\n",
    "cap_lines.take(15)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Use `flatMap()` to create an RDD of the words in War and Peace and count the number of words."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style scoped>\n",
       "  .ansiout {\n",
       "    display: block;\n",
       "    unicode-bidi: embed;\n",
       "    white-space: pre-wrap;\n",
       "    word-wrap: break-word;\n",
       "    word-break: break-all;\n",
       "    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n",
       "    font-size: 13px;\n",
       "    color: #555;\n",
       "    margin-left: 4px;\n",
       "    line-height: 19px;\n",
       "  }\n",
       "</style>\n",
       "<div class=\"ansiout\">573322\n",
       "<span class=\"ansired\">Out[</span><span class=\"ansired\">13</span><span class=\"ansired\">]: </span>[&apos;1869&apos;, &apos;WAR&apos;, &apos;AND&apos;, &apos;PEACE&apos;, &apos;by&apos;, &apos;Leo&apos;, &apos;Tolstoy&apos;, &apos;BK1&apos;, &apos;BOOK&apos;, &apos;ONE&apos;]\n",
       "</div>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Split each line into words using get_words()\n",
    "\n",
    "#the usage of get_words is make each word a line, instead of numerous words in a row or in a line\n",
    "words = lines.flatMap(lambda l: get_words(l))\n",
    "print(words.count())\n",
    "\n",
    "#the first 10 words\n",
    "words.take(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Finally, use `distinct()` to count the number of different words in the RDD."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style scoped>\n",
       "  .ansiout {\n",
       "    display: block;\n",
       "    unicode-bidi: embed;\n",
       "    white-space: pre-wrap;\n",
       "    word-wrap: break-word;\n",
       "    word-break: break-all;\n",
       "    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n",
       "    font-size: 13px;\n",
       "    color: #555;\n",
       "    margin-left: 4px;\n",
       "    line-height: 19px;\n",
       "  }\n",
       "</style>\n",
       "<div class=\"ansiout\"><span class=\"ansired\">Out[</span><span class=\"ansired\">14</span><span class=\"ansired\">]: </span>19206\n",
       "</div>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Count the number of distinct words\n",
    "distinct_words = words.distinct()\n",
    "distinct_words.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Learning activity: Set like transformations"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Use the functions `union()` and `intersection()` to create RDDs of lines with either war or peace mentioned, and both war and peace being mentioned. Count how many lines of each type there are and print some examples."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style scoped>\n",
       "  .ansiout {\n",
       "    display: block;\n",
       "    unicode-bidi: embed;\n",
       "    white-space: pre-wrap;\n",
       "    word-wrap: break-word;\n",
       "    word-break: break-all;\n",
       "    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n",
       "    font-size: 13px;\n",
       "    color: #555;\n",
       "    margin-left: 4px;\n",
       "    line-height: 19px;\n",
       "  }\n",
       "</style>\n",
       "<div class=\"ansiout\">RDDs of lines which have either war or peace mentioned:  369\n",
       "RDDs of lines which have both war and peace mentioned:  7\n",
       "</div>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "union_lines = war_lines.union(peace_lines)\n",
    "print('RDDs of lines which have either war or peace mentioned: ', union_lines.count())\n",
    "\n",
    "intersect_lines = war_lines.intersection(peace_lines)\n",
    "print('RDDs of lines which have both war and peace mentioned: ', intersect_lines.count())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Learning activity: `reduce()`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You have already seen three actions: `collect()` which returns all elements in the RDD, `take(n)`, which return the first `n` elements of the RDD, and `count()` which returns the number of elements in the RDD.\n",
    "\n",
    "The action `reduce()` takes as input a function which collapses two elements into one. Use it to find the longest word in War and Peace."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style scoped>\n",
       "  .ansiout {\n",
       "    display: block;\n",
       "    unicode-bidi: embed;\n",
       "    white-space: pre-wrap;\n",
       "    word-wrap: break-word;\n",
       "    word-break: break-all;\n",
       "    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n",
       "    font-size: 13px;\n",
       "    color: #555;\n",
       "    margin-left: 4px;\n",
       "    line-height: 19px;\n",
       "  }\n",
       "</style>\n",
       "<div class=\"ansiout\"><span class=\"ansired\">Out[</span><span class=\"ansired\">16</span><span class=\"ansired\">]: </span>&apos;characteristically&apos;\n",
       "</div>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "#reduce works by comparing one data with the neighbouring data\n",
    "def longer_words(x, y):\n",
    "    if len(x) > len(y):\n",
    "        return x\n",
    "    else:\n",
    "        return y\n",
    "\n",
    "words.reduce(lambda x, y: longer_words(x, y))\n",
    "\n",
    "#words.reduce(lambda x, y: return x if len(x) > len(y) else y)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Bonus activity: merging filters"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Find all the lines that mention both war and peace **without** using `intersection()`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style scoped>\n",
       "  .ansiout {\n",
       "    display: block;\n",
       "    unicode-bidi: embed;\n",
       "    white-space: pre-wrap;\n",
       "    word-wrap: break-word;\n",
       "    word-break: break-all;\n",
       "    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n",
       "    font-size: 13px;\n",
       "    color: #555;\n",
       "    margin-left: 4px;\n",
       "    line-height: 19px;\n",
       "  }\n",
       "</style>\n",
       "<div class=\"ansiout\"><span class=\"ansired\">Out[</span><span class=\"ansired\">17</span><span class=\"ansired\">]: </span>7\n",
       "</div>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "merging_lines = lines.filter(lambda l: 'war' in get_words(l)).filter(lambda l: 'peace' in get_words(l))\n",
    "#both works the same\n",
    "#merging_lines = lines.filter(lambda l: 'war' in get_words(l) and 'peace' in get_words(l))\n",
    "merging_lines.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Bonus learning activity: Finding proper nouns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The Python function `str.istitle()` returns `True` if the string `str` is titlecased: the first character is uppercase and others are lowercase. Use it to:\n",
    "* Find the set of distinct words in War and Peace which are titlecased\n",
    "* Find the set of distinct words in War and Peace which are not titlecased"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style scoped>\n",
       "  .ansiout {\n",
       "    display: block;\n",
       "    unicode-bidi: embed;\n",
       "    white-space: pre-wrap;\n",
       "    word-wrap: break-word;\n",
       "    word-break: break-all;\n",
       "    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n",
       "    font-size: 13px;\n",
       "    color: #555;\n",
       "    margin-left: 4px;\n",
       "    line-height: 19px;\n",
       "  }\n",
       "</style>\n",
       "<div class=\"ansiout\"><span class=\"ansired\">Out[</span><span class=\"ansired\">18</span><span class=\"ansired\">]: </span>3068\n",
       "</div>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "titlecased_words = words.filter(lambda l: l.istitle()).distinct()\n",
    "titlecased_words.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The Python function `str.lower` returns a string with all characters of `str` lowercase. Use it, along with your previously generated RDD to find the set of words in War and Peace which only appear titlecased."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style scoped>\n",
       "  .ansiout {\n",
       "    display: block;\n",
       "    unicode-bidi: embed;\n",
       "    white-space: pre-wrap;\n",
       "    word-wrap: break-word;\n",
       "    word-break: break-all;\n",
       "    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n",
       "    font-size: 13px;\n",
       "    color: #555;\n",
       "    margin-left: 4px;\n",
       "    line-height: 19px;\n",
       "  }\n",
       "</style>\n",
       "<div class=\"ansiout\">16138\n",
       "<span class=\"ansired\">Out[</span><span class=\"ansired\">19</span><span class=\"ansired\">]: </span>\n",
       "[&apos;Sivtsev&apos;,\n",
       " &apos;Wurttemberg&apos;,\n",
       " &apos;Magnitski&apos;,\n",
       " &apos;Bilibin&apos;,\n",
       " &apos;Karataev&apos;,\n",
       " &apos;Mamonov&apos;,\n",
       " &apos;Frenchies&apos;,\n",
       " &apos;Glinka&apos;,\n",
       " &apos;Hohenlohe&apos;,\n",
       " &apos;Prishprish&apos;,\n",
       " &apos;Elba&apos;,\n",
       " &apos;Fahrenheit&apos;,\n",
       " &apos;Lyadov&apos;,\n",
       " &apos;Bourbons&apos;,\n",
       " &apos;God&apos;]\n",
       "</div>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "#not_titlecased_words = words.filter(lambda l: not l.istitle()).distinct()\n",
    "\n",
    "not_titlecased_words = words.subtract(titlecased_words).distinct()\n",
    "print(not_titlecased_words.count())\n",
    "\n",
    "#to extract the titlecased words to only to proper nouns. words such as 'the' even though is titlecased, might be caused by\n",
    "#the placement that it is at the beginning of the sentence.\n",
    "proper_nouns = titlecased_words.subtract(not_titlecased_words.map(lambda l: l.title()))\n",
    "proper_nouns.take(15)\n",
    "#the proper nouns are only names"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Key/Value pairs in Spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "### Learning activity: WordCount in Spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Use the functions `flatMap()` and `reduceByKey()` to count the number of occurences of each word in War and Peace, and print the count of five words."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style scoped>\n",
       "  .ansiout {\n",
       "    display: block;\n",
       "    unicode-bidi: embed;\n",
       "    white-space: pre-wrap;\n",
       "    word-wrap: break-word;\n",
       "    word-break: break-all;\n",
       "    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n",
       "    font-size: 13px;\n",
       "    color: #555;\n",
       "    margin-left: 4px;\n",
       "    line-height: 19px;\n",
       "  }\n",
       "</style>\n",
       "<div class=\"ansiout\"><span class=\"ansired\">Out[</span><span class=\"ansired\">20</span><span class=\"ansired\">]: </span>\n",
       "[(&apos;combinations&apos;, 6),\n",
       " (&apos;reinstated&apos;, 2),\n",
       " (&apos;mates&apos;, 2),\n",
       " (&apos;reminiscence&apos;, 1),\n",
       " (&apos;repelling&apos;, 1)]\n",
       "</div>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "word_count = words.map(lambda l: (l, 1)).reduceByKey(lambda a, b: a+b)\n",
    "word_count.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Learning activity: using `groupByKey()`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Reimplement the above word count using `groupByKey()` instead of `reduceByKey()`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style scoped>\n",
       "  .ansiout {\n",
       "    display: block;\n",
       "    unicode-bidi: embed;\n",
       "    white-space: pre-wrap;\n",
       "    word-wrap: break-word;\n",
       "    word-break: break-all;\n",
       "    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n",
       "    font-size: 13px;\n",
       "    color: #555;\n",
       "    margin-left: 4px;\n",
       "    line-height: 19px;\n",
       "  }\n",
       "</style>\n",
       "<div class=\"ansiout\"><span class=\"ansired\">Out[</span><span class=\"ansired\">21</span><span class=\"ansired\">]: </span>\n",
       "[(&apos;combinations&apos;, 6),\n",
       " (&apos;reinstated&apos;, 2),\n",
       " (&apos;mates&apos;, 2),\n",
       " (&apos;reminiscence&apos;, 1),\n",
       " (&apos;repelling&apos;, 1)]\n",
       "</div>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "word_count = words.map(lambda l: (l, 1)).groupByKey().map(lambda l: (l[0], sum(l[1])))\n",
    "word_count.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Bonus learning activity: computing the average of each key"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The pair RDD defined below `word_line_pairs` has an element for each line in War and Peace with as key the first word, and as value the line itself. Use it to compute the average length of each line for each starting word."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {
    "collapsed": true
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style scoped>\n",
       "  .ansiout {\n",
       "    display: block;\n",
       "    unicode-bidi: embed;\n",
       "    white-space: pre-wrap;\n",
       "    word-wrap: break-word;\n",
       "    word-break: break-all;\n",
       "    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n",
       "    font-size: 13px;\n",
       "    color: #555;\n",
       "    margin-left: 4px;\n",
       "    line-height: 19px;\n",
       "  }\n",
       "</style>\n",
       "<div class=\"ansiout\"></div>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "word_line_pairs = lines.filter(lambda line: len(get_words(line)) > 0).map(lambda line: (get_words(line)[0], line))     "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style scoped>\n",
       "  .ansiout {\n",
       "    display: block;\n",
       "    unicode-bidi: embed;\n",
       "    white-space: pre-wrap;\n",
       "    word-wrap: break-word;\n",
       "    word-break: break-all;\n",
       "    font-family: \"Source Code Pro\", \"Menlo\", monospace;;\n",
       "    font-size: 13px;\n",
       "    color: #555;\n",
       "    margin-left: 4px;\n",
       "    line-height: 19px;\n",
       "  }\n",
       "</style>\n",
       "<div class=\"ansiout\"><span class=\"ansired\">Out[</span><span class=\"ansired\">23</span><span class=\"ansired\">]: </span>\n",
       "[(&apos;hastily&apos;, 66.71428571428571),\n",
       " (&apos;reinstated&apos;, 12.0),\n",
       " (&apos;packed&apos;, 39.333333333333336),\n",
       " (&apos;Separate&apos;, 64.0),\n",
       " (&apos;births&apos;, 66.0)]\n",
       "</div>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "#map the first word to how many occurence of the word\n",
    "word_line_pairs = word_line_pairs.map(lambda l: (l[0], (len(l[1]), 1)))\n",
    "#count the occurence\n",
    "word_line_pairs = word_line_pairs.reduceByKey(lambda x, y: (x[0] + y[0], x[1] + y[1]))\n",
    "#count the average of the how many words in the first word\n",
    "word_line_pairs = word_line_pairs.map(lambda l: (l[0], l[1][0] / float(l[1][1])))\n",
    "word_line_pairs.take(5)"
   ]
  }
 ],
 "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.8"
  },
  "name": "Spark",
  "notebookId": 4251968096026564
 },
 "nbformat": 4,
 "nbformat_minor": 1
}