notebooks/gutenberg/part1-preprocessing-EMPTY.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from __future__ import print_function\n",
    "%matplotlib inline\n",
    "import matplotlib.pylab as plt\n",
    "import sys, os, glob\n",
    "import numpy as np\n",
    "import subprocess\n",
    "\n",
    "from IPython.display import HTML\n",
    "import xml.etree.ElementTree as ET\n",
    "try:\n",
    "    tree = ET.parse(os.environ['HADOOP_CONF_DIR'] + '/yarn-site.xml')\n",
    "except IOError:\n",
    "    raise IOError(\"Can't find the yarn configuration -- is HADOOP_CONF_DIR set?\")\n",
    "root = tree.getroot()\n",
    "yarn_web_app = root.findall(\"./property[name='yarn.resourcemanager.webapp.address']\")[0].find('value').text\n",
    "yarn_web_app_string = \"If this works successfully, you can check the <a target='_blank' href='http://{yarn_web_app}'>YARN application scheduler</a> and you should see your app listed there. Clicking on the 'Application Master' link will bring up the familiar Spark Web UI. \"\n",
    "\n",
    "plt.rcParams['figure.figsize'] = (10,6)\n",
    "plt.rcParams['font.size'] = 18\n",
    "plt.style.use('fivethirtyeight')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Gutenberg N-Grams\n",
    "\n",
    "In this series of notebooks, we will quantitatively explore the text of the [Gutenberg E-Books Project](https://www.gutenberg.org/), a free repository of e-books that are in the public domain. All of the books in html format have been downloaded for this tutorial."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Preprocessing procedure\n",
    "\n",
    "#### In this notebook, we will pre-process the text corpus using Spark\n",
    "\n",
    "The basic steps will be something like:\n",
    "\n",
    "2. create a metadata lookup table that can be distributed to the workers\n",
    "3. clean the dataset for missing values and repeated entries\n",
    "4. save the final, cleaned dataset to HDFS\n",
    "\n",
    "But first, we need to set up the Spark runtime."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Setting up Python and Spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Setting up the spark environment can be a bit of a trial and error procedure. Often you'll need to configure settings (in particular dealing with memory) to fit your cluster and your particular application. Below, we will specify a few of the most important ones -- but you can see the full list in the [Spark Configuration guide](http://spark.apache.org/docs/latest/configuration.html) and if you are using YARN there are critical options also listed under the [YARN deployment guide](http://spark.apache.org/docs/latest/running-on-yarn.html).\n",
    "\n",
    "This is also a great time to have a look at the [Transformations](http://spark.apache.org/docs/latest/programming-guide.html#transformations) and [Actions](http://spark.apache.org/docs/latest/programming-guide.html#actions) in the Spark programming guide or the list of [RDD methods in the Spark API](http://spark.apache.org/docs/latest/api/python/pyspark.html#pyspark.RDD)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Python environment setup\n",
    "\n",
    "First, we need to make sure that the `pyspark` libraries are accessible in this notebook. To do this, we can add them to the library search path with a convenient `findspark` library: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Runtime settings\n",
    "\n",
    "Below we will specify a range of settings for the Spark runtime system. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<div class=\"alert alert-info\">\n",
    "<p><strong>Note</strong></p> \n",
    "\n",
    "<p>The environment variables have to be declared before any other spark initialization takes place (including creating a <code>SparkConf</code> object.</p>\n",
    "</div>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "ncores = int(os.environ.get('LSB_DJOB_NUMPROC', 1)) # how many cores do we have for the driver\n",
    "\n",
    "os.environ['SPARK_DRIVER_MEMORY'] = '%dG'%(ncores*2*0.7)\n",
    "os.environ['PYSPARK_PYTHON'] = subprocess.check_output('which python', shell=True).rstrip()\n",
    "\n",
    "import pyspark\n",
    "from pyspark import SparkConf, SparkContext"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Initializing `SparkConf` \n",
    "\n",
    "When starting the Spark runtime through the notebook or inside a script (i.e. when not calling one of the spark scripts like `spark_submit`), you can create a `SparkConf` object that allows you to set up the runtime. This is quite convenient and much cleaner and more readable than specifying the options on the commandline. \n",
    "\n",
    "<div class=\"alert alert-info\" role=\"alert\" style=\"margin-top: 10px\">\n",
    "<p><strong>Pro tip</strong></p>\n",
    "\n",
    "<p>If you use the same base configuration often, you can store it in a file and only override those that are needed through `SparkConf` (we'll do this at the end of this section).</p>\n",
    "</div>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# initializing the SparkConf\n",
    "conf = SparkConf()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Executor options\n",
    "\n",
    "The full list of options is very long, but the basic ones you'll *always* want to at least think about are ones pertaining to the basic configuration of the executors: number of executors, memory per executor, and number of cores per executor. \n",
    "\n",
    "A few notes about the memory configuration: the `spark.executor.memory` should not be set to the total memory of the node. Some memory is needed for the OS (including HDFS and other services), and still more is required for the Spark overhead. So in our case here, we have 16 Gb of memory per node but can only use around 12 Gb of this for the executors. Since we need to leave room for 10% YARN overhead, we specify 9 Gb here to be safe. If your executors start dying off for strange reasons, try reducing the memory here. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "conf.set('spark.executor.memory', '9g')\n",
    "\n",
    "# put the number of executors and cores into variables so we can refer to it later\n",
    "num_execs = 20\n",
    "exec_cores = 4\n",
    "\n",
    "conf.set('spark.executor.instances', str(num_execs))\n",
    "conf.set('spark.executor.cores', str(exec_cores))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Memory configuration\n",
    "\n",
    "Two other key memory options exist, specifying the amount of executor memory reserved for *cached* data and for *shuffle* data. Depending on what your application is doing, you may need more of one or the other. For example, if you are running a lot of iterative operations on a large dataset, you probably want a good amount of memory for RDD caching. On the other hand, if you are doing lots of expensive shuffles that occur when sorting of grouping by key, you may want more shuffle memory. Note that if either one starts to run low, your application (hopefully) won't crash, but instead simply spill to local disk. \n",
    "\n",
    "You can check on the cache memory and shuffle memory in two ways while your application is running. In the Spark UI, you can see the cached RDDs under the `Storage` tab - if they start spilling to disk, this is where you will see it. Similarly, if you are running a large shuffle job, you can click on the stage details in the Spark UI and see the shuffle memory and disk statistics. We will check on both of these later on in this application. \n",
    "\n",
    "Here we will set these two options explicitly (30% of the heap for caching, 50% for shuffles). Note that the default here is 60% for caching and 20% for shuffles, but we expect to have some fairly memory-hungry shuffles in our application, so we adjust the allocations a bit. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "conf.set('spark.storage.memoryFraction', 0.3)\n",
    "conf.set('spark.shuffle.memoryFraction', 0.5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Driver memory\n",
    "\n",
    "The amount of memory allocated to the driver program could be crucial if the driver has to deal with a lot of late-stage aggegation products or if you want to collect a significant chunk of data out of the RDD. You can see how much memory has been allocated to the driver either in the Spark Web UI or in the messages printed to the console at initialization. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "conf.set('spark.yarn.am.memory', '8g')\n",
    "conf.set('spark.yarn.am.cores', 2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Reusing a saved configuration\n",
    "\n",
    "Often, you will probably have a few different jobs that all need the same configuration. Instead of always specifying it inside the notebook, it might be a little cleaner to save it in an external file. \n",
    "\n",
    "In this case, we have the configuration saved in `~/spark_workshop/notebooks/gutenberg/spark_config`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "!cat ./spark_config/spark-defaults.conf"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Next time we start the `SparkContext` we can simply set the `SPARK_CONF_DIR` environment variable to point to the directory where our configuration lives, and it will be taken care of automatically. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<div class=\"alert alert-info\">\n",
    "<p><strong>Pro tip</strong></p> \n",
    "\n",
    "<p>You can edit the configuration straight from Jupyter by navigating to it directly in the Jupyter file browser.\n",
    "</div>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Starting the `SparkContext`\n",
    "\n",
    "This is our entry point to the Spark runtime - it is used to push data into spark or load RDDs from disk etc. Note that here we specify `master = 'yarn-client'`, which will automatically request the resources we have configured in `conf` above. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sc = SparkContext(master = 'yarn-client', conf = conf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "HTML(yarn_web_app_string.format(yarn_web_app=yarn_web_app))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Loading the dataset from HDFS\n",
    "\n",
    "I have already pre-loaded the data for you to avoid some tedious boilerplate code. If you want to see the process of loading the data into Spark from local disk anyway, you can have a look at the [data input notebook](part0-data-input.ipynb).\n",
    "\n",
    "Here, we just load the data from my HDFS:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "text_rdd = sc.pickleFile('hdfs:///user/roskarr/gutenberg/raw_text_rdd')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "%time text_rdd.cache().count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This operation loaded the data from the Hadoop distributed filesystem and created an RDD. If you look at the details for this stage in the Spark UI (click on the \"Stages\" tab at the top, then click on the \"count\" stage at the top of the list) you can understand why this is: in the column named \"locality level\", you see that for many tasks it says `NODE LOCAL` while for others it might say `RACK LOCAL`. These mean that either the data chunk was physically present on the disk of the node that was reading it in (`NODE LOCAL`) or it was on one of the nodes on the same switch (`RACK LOCAL`). Of course the additional advantage is not having to deal with the filesystem overhead of 40k+ small files. \n",
    "\n",
    "You can also check how much memory this raw data occupies by clicking on the \"Storage\" tab in the Spark UI. Note that if the value reported there says the data is serialized, it means it's also likely compressed so the actual size is a bit larger. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## A first look at the data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Lets have a look at what this RDD looks like -- this is often most easily done using the `first` method of the RDD: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# TODO: use the 'first()' method of the text_rdd to extract the the meta and text out of the first element \n",
    "meta, text = text_rdd.first()\n",
    "print('metadata: ', meta)\n",
    "print('\\ntext: ', text[20000:21000])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For each data element we have a tuple containing a dictionary of metadata (`'lang'`, `'first_name'` etc.), and the text. Lets have another look at what's inside the metadata dictionary: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "for key, value in meta.iteritems(): \n",
    "    print(key + ': ', value)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Clearly, some of these are not very useful for this particular book, but note a really important one: `gid` -- this is the Gutenberg book ID and is a unique identifier for each book. We'll use this a lot later on. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### First bit of analysis -- how long are the documents? \n",
    "\n",
    "Lets have a quick look at the number of words in these books. We'll do this as follows: \n",
    "\n",
    "1. take the values (text) of the `text_rdd` (there is a `values()` method that discards the keys of the RDD)\n",
    "2. use the `split` method of the text string to break it up into a list of words\n",
    "3. use `len` to get the length of the list of words. \n",
    "4. finally, we will use the `collect` method to extract the numbers to the driver. \n",
    "\n",
    "**Hint**: whenever you are not sure how exactly to proceed, make a new cell and experiment! Use the `first` and `take` methods of the RDD you are creating to see what is happening with the data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# TODO: map text_rdd to contain only the number of words in each document\n",
    "text_lengths = (text_rdd.values()\n",
    "                        .map(<FILL IN>)\n",
    "                        .collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "assert(text_lengths[:10] == [66976, 17180, 8455, 53847, 65001, 50678, 72798, 16398, 30705, 68609])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can quickly plot a histogram:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "plt.hist(text_lengths, bins = 100, range=[0,3e5]);\n",
    "plt.xlabel('$N_{words}$'); plt.ylabel('count');"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Broadcast variables: a good way to distribute metadata"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note that our dataset at the moment is composed of `(dictionary, text)` tuples. This is not ideal for several reasons: first, complex structures like dictionaries are more expensive to serialize, so replacing them by simple integer IDs can improve performance. And second, dictionaries are not hashable so cannot serve as keys. \n",
    "\n",
    "The solution is to replace the dictionary with just a single ID integer (in this case we already have it, the `gid` or the gutenberg ID of the book). However, as we manipulate the dataset, we will still want ready access to the metadata, but at the moment we have no efficient way of working with it. What we want is to generate a lookup table of metadata that can easily be retrieved when needed. \n",
    "\n",
    "For these kinds of situations when passing around some data to all the workers is required, you should use Spark's [broadcast variable](http://spark.apache.org/docs/latest/programming-guide.html#broadcast-variables) mechanism. By placing data inside a broadcast variable, you make it available to all the workers and it only needs to be sent across the network once. The next time a value from the broadcast variable is needed, it's simply taken from the local executor's memory, incurring no extra network cost. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# set up the list of fields\n",
    "metadata_fields = text_rdd.first()[0].keys() # this takes the metadata of the first element and extracts the keys\n",
    "meta_dict = dict()\n",
    "\n",
    "for meta in text_rdd.keys().collect() :\n",
    "    meta_dict[meta['gid']] = {key: meta[key] for key in metadata_fields}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This is now a look-up table that allows us to quickly access all the metadata indexed by `gid`. For example to get the metadata of book with `gid=101`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "meta_dict[101]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To get, for example, the author birth year for book with `gid = 101`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "meta_dict[101]['birth_year']"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we want to send this lookup-table to all the executors, so turn it into a broadcast variable: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# call it meta_b for 'broadcast'\n",
    "meta_b = sc.broadcast(meta_dict)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The underlying data object stored in `meta_b` can be accessed simply by\n",
    "\n",
    "    > meta_b.value\n",
    "    \n",
    "We'll make use of this soon. If you check the console/terminal output, you will see an INFO message that the broadcast has been created, i.e. \n",
    "\n",
    "```\n",
    "15/06/24 17:18:44 INFO storage.MemoryStore: Block broadcast_6_piece0 stored as bytes in memory (estimated size 910.7 KB, free 4.1 GB)\n",
    "15/06/24 17:18:44 INFO storage.BlockManagerInfo: Added broadcast_6_piece0 in memory on 10.201.20.22:47821 (size: 910.7 KB, free: 4.1 GB)\n",
    "15/06/24 17:18:44 INFO spark.SparkContext: Created broadcast 6 from broadcast at PythonRDD.scala:403\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Save the metadata dictionary for later use\n",
    "We will need the metadata dictionary at a later point, so we save it to disk now to avoid having to regenerate it later. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from cPickle import dump"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "dump(meta_dict, open('{home}/gutenberg_metadata.dump'.format(home=os.environ['HOME']), 'w'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This saved it to your home directory into a file called `gutenberg_metadata.dump`. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Cleaning the data with filtering"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we're ready to do some quality checks on the data. Let's check out the first couple of metadata entries: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "text_rdd.keys().take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If you look at just the first few entries it becomes clear that we're going to have to do some quality control here. For example, we probably don't want books with \"None\" as either of the author names, and likewise we have to have the birth date in order to be able to create a time series out of the data in the end. \n",
    "\n",
    "Construct an RDD, as above, except that you filter out all the elements that have `None` for `title`, `first_name`, `last_name`, or `birth_year`. In addition, filter out the data with \"BC\" in either birth or death year. \n",
    "\n",
    "As a reminder, here is a cartoon illustration of the difference between `map` and `filter` RDD methods. `map` simply applies the function to each element, returning another element. \n",
    "\n",
    "![map](../figs/map_example.svg)\n",
    "\n",
    "In this example, with `filter` we are filtering out all the even elements of the RDD. The function that is passed to `filter` just has to evaluate to either `True` (1) or `False` (0) given the input data. The function `lambda (k,v): v%2` evaluates to 0 if `v` is even and 1 of `v` is odd. Hence, only the odd values pass the filter. \n",
    "\n",
    "![filter](../figs/filter_example.svg)\n",
    "\n",
    "The `filter_func` has already been defined for you below, but you need to apply it to `text_rdd`. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def filter_func(meta) : \n",
    "    no_none = all([meta[name] is not None for name in ['title', 'first_name', 'last_name', 'birth_year']])\n",
    "    if not no_none : \n",
    "        return False\n",
    "    else : \n",
    "        no_birth_bc = 'BC' not in meta['birth_year']\n",
    "        no_death_bc = True if meta['death_year'] is None else 'BC' not in meta['death_year']\n",
    "        return no_birth_bc & no_death_bc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# TODO: filter the meta data of the RDD using the filter_func \n",
    "# remember that the data of text_rdd is a (metadata, text) tuple -- filter_func only needs the metadata\n",
    "filtered_rdd = text_rdd.filter(lambda (meta, text): filter_func(meta))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "filtered_rdd.keys().take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "How many do we have left? "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "nfiltered = filtered_rdd.count()\n",
    "print('number of books after filtering: ', nfiltered)\n",
    "assert(nfiltered == 34123)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### A final bit of cleanup: \n",
    "\n",
    "some of the books end up split across multiple entries. Since it's the same book, each of the entries should have the same `gid`. \n",
    "\n",
    "To check for this we will use one of the most basic and common MapReduce patterns -- the key count: \n",
    "\n",
    "* map the data into `key`,`value` pairs where `key` is the quantity we want to count and `value` is just 1. In this case, the `key` will be `gid`\n",
    "* invoke a reduction *by key*, where the reduction operator is a simple addition\n",
    "\n",
    "Finally, we will sort the result in descending order and print out the first few elements to check whether we have to worry about documents spanning multiple files or not. \n",
    "\n",
    "The RDD operations that are needed are [`reduceByKey`](http://spark.apache.org/docs/latest/api/python/pyspark.html#pyspark.RDD.reduceByKey) and [sortBy](http://spark.apache.org/docs/latest/api/python/pyspark.html#pyspark.RDD.sortBy).\n",
    "\n",
    "`reduceByKey` works by grouping all data of a key together and applying the reduction function just to that data. Here's a simple illustration, in this case using a simple addition of two elements as a reduction:\n",
    "\n",
    "![reducebykey](../../slides/figs/reduceByKey_example.svg)\n",
    "\n",
    "\n",
    "\n",
    "For the `keyFunc` of the call to `sortBy`, use a `lambda` function that extracts the counts obtained from the `reduceByKey`. \n",
    "\n",
    "So, the procedure should be : \n",
    "\n",
    "1. `map` the `filtered_rdd` using a lambda function to contain (`gid`, 1) tuples\n",
    "2. `reduceByKey`\n",
    "3. `sortBy` (specify decreasing order, see the API) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from operator import add"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# map the filtered_rdd to contain just the tuple (gid, 1)\n",
    "map_filtered = filtered_rdd.map(lambda (meta, text): (meta['gid'],1))\n",
    "\n",
    "# reduce the map_filtered rdd by key using the add operator to get the total counts per gid\n",
    "reduced_gid_rdd = map_filtered.reduceByKey(<FILL IN>)\n",
    "\n",
    "# sort by count and print out the top 5\n",
    "reduced_gid_rdd.sortBy(<FILL IN>).take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "assert(_ == [(30310, 51), (6478, 43), (3772, 40), (8700, 35), (3332, 33)])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note that there are several transformations here that lead to the final result, `sorted_reduced`. A common syntax is to group them all together, by enclosing them in `( )` and chaining them: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# TODO: repeat the above steps but chain them together\n",
    "(filtered_rdd.<FILL IN>\n",
    "                         )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Looks like we have a few that are made up of multiple sections. To combine them, we will use `reduceByKey` which will result in having an RDD of `gid`'s as keys and the combined text of each `gid`. The reduction function in `reduceByKey` can be a simple in-line lambda function that just adds two elements together -- in this case we'll use the `add` function."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# first, map filtered_rdd to contain tuples (gid, text)\n",
    "# second, use reduceByKey to combine together text belonging to the same gid\n",
    "cleaned_rdd = (filtered_rdd.map(lambda (meta, text): (meta['gid'], text))\n",
    "                           .reduceByKey(add))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As a simple sanity check, lets look at `gid`=6478, which according to the cell above has 43 sections in the original dataset: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "len(filtered_rdd.map(lambda (meta, text): (meta['gid'],1))\n",
    "                .lookup(6478))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "len(cleaned_rdd.lookup(6478))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To avoid having to do all these pre-processing steps again at a later point, lets also save the `cleaned_rdd`. \n",
    "\n",
    "**In the cells below, replace \"YOUR USERNAME\" with your own username.**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "cleaned_rdd.saveAsPickleFile('/user/YOUR USERNAME/gutenberg/cleaned_rdd')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This is now saved in the directory we specified, one file per partition:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "!hadoop fs -ls /user/YOUR USERNAME/gutenberg/cleaned_rdd | head"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "!hadoop fs -du -h /user/YOUR USERNAME/gutenberg/"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note that here we used the `hadoop` command in the local bash shell (the `!` at the beginning of the line means we are executing the command in the shell). This allows us to access the hadoop filesystem (HDFS), which is separate from the local file system we are used to. You'll notice, for example, that this directory doesn't exist in the *local* filesystem:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "!ls /user/YOUR USERNAME/gutenberg/"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You can also browse the filesystem via the [HDFS web UI](http://hadoop.ethz.ch:50070). The `hadoop fs` command has many of the same options as regular Linux/Unix shell commands you might be used to for manipulating files and directories. Try running\n",
    "\n",
    "```bash\n",
    "cluster $> module load hadoop\n",
    "cluster $> hadoop fs -help\n",
    "```\n",
    "\n",
    "in a new shell to see all the options. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Recap of steps up until this point\n",
    "\n",
    "We've done quite a lot already with our dataset in Spark, although it's only the beginning!\n",
    "\n",
    "1. created an RDD of filenames (`filename_rdd`)\n",
    "2. transformed the `filename_rdd` into an RDD of `(metadata, text)` (`text_rdd`); we also saved this to HDFS\n",
    "3. filtered out data with bad metadata, e.g. missing author names etc.\n",
    "3. cleaned up the entries a bit more by merging ones with identical IDs; we called this `cleaned_rdd`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Shutting down the `SparkContext`\n",
    "\n",
    "Now that the pre-processing is done, we will shut down the `SparkContext` before continuing to the data analysis notebook. We have all of our results saved in HDFS, so to continue from where we left off will just require loading data from there. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "sc.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now that the pre-processing steps are complete, we can continue to the [analysis notebook](part2-ngram-viewer-EMPTY.ipynb)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}