notebooks/gutenberg/part0-data-input.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "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 os, subprocess\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 English and German books have been downloaded for this tutorial and a small python package has been made available that allows you to easily parse the text and the associated metadata. \n",
    "\n",
    "In this part \"zero\" notebook, we just ingest the data, process the text, and save the raw text RDD (with punctuation and html tags removed). "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Spark configuration\n",
    "\n",
    "Below we specify that this notebook should use the configuration stored in <code>./spark_config</code> -- the options will be discussed in detail in the next notebook."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<div class=\"alert alert-info\">\n",
    "Note that the environment variables have to be declared before any other spark initialization takes place (including creating a <code>SparkConf</code> object.\n",
    "</div>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "ncores = int(os.environ.get('LSB_DJOB_NUMPROC', 1))\n",
    "\n",
    "os.environ['SPARK_CONF_DIR'] = os.path.abspath('./spark_config')\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 findspark\n",
    "findspark.init()\n",
    "\n",
    "import pyspark\n",
    "from pyspark import SparkConf, SparkContext"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Starting the `SparkContext`\n",
    "This is our entry point to the Spark runtime - it is used to push data into spark or load RDDs from disk etc. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sc = SparkContext(master = 'yarn-client')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If this works successfully, you can check the [YARN application scheduler](http://hadoop.ethz.ch:8088/cluster) and you should see your app listed there. Clicking on the \"Application Master\" link will bring up the familiar Spark Web UI. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Make a key-value RDD of book metadata and text"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Getting data into spark from a collection of local files is a very common task. A useful pattern to keep in mind is the following: \n",
    "\n",
    "1. make a list of filenames and distribute it among the workers\n",
    "3. \"map\" each filename to the data you want to get out\n",
    "4. now you are left with the RDD of raw data distributed among the workers!"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In our case of the Gutenberg Project e-book data, we have a directory of `html` files which hold the actual book text, and another directory of associated metadata files (the RDF files). To make your life easier for the purpose of this tutorial, we have made a small python module called `gutenberg_cleanup` that has some handy functions for pulling out the relevant text and metadata out of the raw dataset. \n",
    "\n",
    "The [`gutenberg_cleanup`](gutenberg_cleanup.py) module contains three functions that can help with this: `get_gid`, `get_metadata` and `get_text`.\n",
    "\n",
    "They pretty much do the obvious: \n",
    "\n",
    "`get_gid` takes an html path and pulls out the book ID (`gid`)\n",
    "\n",
    "`get_metadata` takes a `gid` and returns a metadata object with various useful fields that will be used to create a unique key for each book\n",
    "\n",
    "`get_text` takes a path to an html file and returns the raw text extracted from HTML, cleaned of tags and punctuation and converted to lower case. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Initializing the raw dataset using `sc.parallelize`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "number of books:  48177\n"
     ]
    }
   ],
   "source": [
    "import glob\n",
    "\n",
    "# get a list of all html files in the data directory\n",
    "flist = glob.glob('/cluster/work/sdid/roskarr/gutenberg/html/*html')\n",
    "print('number of books: ', len(flist))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "When you use `sc.parallelize` to distribute a dataset across the cluster, you can choose the number of partitions across which to distribute the dataset. The higher the number of partitions, the higher the \"parallelism\". When Spark subsequently executes maps and reduces on this dataset, it does so by dispatching tasks to different executors, which then request the cores under their control to do the actual work. By increasing the number of partitions, you increase the number of tasks - more tasks gives the Spark scheduler more flexibility in distributing the work across the cluster and therefore maximally leveraging the compute resources at its disposal. In some cases, where a single partition might require a lot of memory it can cause `Out of memory` errors - in such cases, simply reducing the amount of data per task by increasing the parallelism can help. \n",
    "\n",
    "Note that as long as the tasks take a few hundred milliseconds the scheduler should have no trouble dispatching them. On the other hand, there is a bit of overhead associated with partitioning the data so you don't want an unreasonably high number of partitions. You can see the [Spark guide](http://spark.apache.org/docs/latest/tuning.html#level-of-parallelism) for a bit more detail. \n",
    "\n",
    "Below, we will choose to use 5 times as many partitions as we have cores in the job. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "job_conf = dict(sc._conf.getAll())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "num_execs = int(job_conf['spark.executor.instances'])\n",
    "exec_cores = int(job_conf['spark.executor.cores'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "files_rdd = sc.parallelize(flist, num_execs*exec_cores*5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['/cluster/work/sdid/roskarr/gutenberg/html/1000.html',\n",
       " '/cluster/work/sdid/roskarr/gutenberg/html/10000.html',\n",
       " '/cluster/work/sdid/roskarr/gutenberg/html/10001.html',\n",
       " '/cluster/work/sdid/roskarr/gutenberg/html/10002.html',\n",
       " '/cluster/work/sdid/roskarr/gutenberg/html/10003.html']"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "files_rdd.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Transforming the list of filenames into a `key,value` pair RDD of metadata and text\n",
    "\n",
    "The raw Gutenberg Project dataset consists of HTML files and files that hold metadata in JSON format. For example: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<?xml version='1.0' encoding='utf-8'?>\r",
      "\r\n",
      "<!DOCTYPE html PUBLIC '-//W3C//DTD XHTML 1.1//EN' 'http://www.w3.org/TR/xhtml11/DTD/xhtml11.dtd'>\r",
      "\r\n",
      "<html xmlns=\"http://www.w3.org/1999/xhtml\">\r",
      "\r\n",
      "  <head><title> </title><meta http-equiv=\"Content-Style-Type\" content=\"text/css\"/><meta http-equiv=\"Content-Type\" content=\"application/xhtml+xml; charset=utf-8\"/><link rel=\"schema.DCTERMS\" href=\"http://purl.org/dc/terms/\"/>\r",
      "\r\n",
      "<link rel=\"schema.MARCREL\" href=\"http://id.loc.gov/vocabulary/relators/\"/>\r",
      "\r\n",
      "<meta content=\"The Magna Carta\" name=\"DCTERMS.title\"/>\r",
      "\r\n",
      "<meta content=\"http://www.gutenberg.orgfiles/10000/10000.txt\" name=\"DCTERMS.source\"/>\r",
      "\r\n",
      "<meta content=\"en\" scheme=\"DCTERMS.RFC4646\" name=\"DCTERMS.language\"/>\r",
      "\r\n",
      "<meta content=\"2015-04-04T04:40:30.599547+00:00\" scheme=\"DCTERMS.W3CDTF\" name=\"DCTERMS.modified\"/>\r",
      "\r\n",
      "<meta content=\"Public domain in the USA.\" name=\"DCTERMS.rights\"/>\r",
      "\r\n"
     ]
    }
   ],
   "source": [
    "!head /cluster/work/sdid/roskarr/gutenberg/html/10000.html"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<?xml version=\"1.0\" encoding=\"utf-8\"?>\r\n",
      "<rdf:RDF xml:base=\"http://www.gutenberg.org/\"\r\n",
      "  xmlns:cc=\"http://web.resource.org/cc/\"\r\n",
      "  xmlns:dcterms=\"http://purl.org/dc/terms/\"\r\n",
      "  xmlns:rdfs=\"http://www.w3.org/2000/01/rdf-schema#\"\r\n",
      "  xmlns:pgterms=\"http://www.gutenberg.org/2009/pgterms/\"\r\n",
      "  xmlns:rdf=\"http://www.w3.org/1999/02/22-rdf-syntax-ns#\"\r\n",
      "  xmlns:dcam=\"http://purl.org/dc/dcam/\"\r\n",
      ">\r\n",
      "  <pgterms:ebook rdf:about=\"ebooks/10000\">\r\n"
     ]
    }
   ],
   "source": [
    "!head /cluster/work/sdid/roskarr/gutenberg/rdf-files/10000/pg10000.rdf"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Data Ingestion procedure\n",
    "\n",
    "Our first task is to ingest this dataset by doing the following: \n",
    "\n",
    "1. convert the html into raw text\n",
    "2. deal with special characters, HTML tags, etc.\n",
    "3. match each metadata entry with its corresponding raw text and compose tuples of the type (metadata, text)\n",
    "\n",
    "\n",
    "These three steps are often the first step of any analysis, and can be quite time consuming to get right. For the purposes of this exercise, we have already built the functions needed to perform these operations. They are found in [`gutenberg_cleanup.py`](gutenberg_cleanup.py) if you want to have a look. \n",
    "\n",
    "The important functions are:\n",
    "\n",
    "* `get_gid` -- returns the Gutenberg ID given an html file\n",
    "* `get_text` -- get cleaned, raw text out of an html file\n",
    "* `get_metadata` -- return the metadata given an ID \n",
    "\n",
    "These will be used to construct a `key,value` pair RDD. The `key` will be the dictionary returned by `get_metadata`, while the `value` we will use the raw text returned by `get_text`. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import gutenberg_cleanup\n",
    "from gutenberg_cleanup import get_metadata, get_text, get_gid"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To pass the `gutenberg_cleanup` source file to the executors, we will use the `addPyFile` method of the `SparkContext`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "sc.addPyFile('{cwd}/gutenberg_cleanup.py'.format(cwd=os.getcwd()))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Use the `map` method of `files_rdd` to map the filenames to `(metadata, text)` tuples using `get_gid` and `get_text` functions:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# TODO\n",
    "id_text_rdd = files_rdd.map(lambda filename: (get_gid(filename), get_text(filename)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we have (ID, text), and we need to make another `map` to get (`metadata, text`) tuples:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# TODO\n",
    "text_rdd = (id_text_rdd.map(lambda (ID, text): (get_metadata(ID), text)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "So that we don't have to constantly re-load the data off disk, lets cache this RDD: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 86 ms, sys: 19 ms, total: 105 ms\n",
      "Wall time: 2min 26s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "text_rdd.cache()\n",
    "text_rdd.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "variables": {
     "UI_url": {}
    }
   },
   "source": [
    "Since we called `count()`, it means that the entire RDD was generated/calculated. This combination of `cache` and `count` is a common way to check how much memory your dataset needs - once `count` completes you can check the memory taken up by the RDD by going to the \"Storage\" tag of the Spark UI. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Because the data is cached, next time you try to use `text_rdd` it will be much much quicker. For example, "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 47 ms, sys: 10 ms, total: 57 ms\n",
      "Wall time: 3.09 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "assert(text_rdd.count() == 48177)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As an aside, we could call the native python `map` in exactly the same way (and run it on the local machine only), though this would take much longer to complete, i.e. \n",
    "\n",
    "    text = map(lambda f: (get_metadata(get_gid(f)), get_text(f)), flist)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Save the raw dataset to HDFS (or local storage)\n",
    "\n",
    "As a final bit of preparation before continuing with analysis, we save the raw data in a way that makes it faster to access later. We don't want to have to read the data off local disk every time we need to repeat some part of the analysis. Instead, it's much more advantageous to use the Hadoop Distributed File System (HDFS) to store the data once we've read it in and put it in a `key,value` format. \n",
    "\n",
    "By storing the data in HDFS, we make sure that the system can take advantage of data-locality at a later stage in our analysis. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Picked up _JAVA_OPTIONS: -XX:ParallelGCThreads=1\n",
      "15/11/20 15:00:58 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "15/11/20 15:00:59 INFO fs.TrashPolicyDefault: Namenode trash configuration: Deletion interval = 0 minutes, Emptier interval = 0 minutes.\n",
      "Deleted /user/roskarr/gutenberg/raw_text_rdd\n"
     ]
    }
   ],
   "source": [
    "!hadoop fs -rm -r -f /user/roskarr/gutenberg/raw_text_rdd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "text_rdd.saveAsPickleFile('hdfs:///user/roskarr/gutenberg/raw_text_rdd')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, whenever we need it, we can read the data off the HDFS instead: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "text_rdd = sc.pickleFile('hdfs:///user/roskarr/gutenberg/raw_text_rdd')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 48 ms, sys: 9 ms, total: 57 ms\n",
      "Wall time: 6.17 s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "48177"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%time text_rdd.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "sc.stop()"
   ]
  }
 ],
 "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
}