notebooks/gutenberg/part3-lang-classification-EMPTY.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 subprocess\n",
    "\n",
    "from ipywidgets import interact, interactive, fixed\n",
    "import ipywidgets as widgets\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": [
    "# Analyzing the Gutenberg Books Corpus"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this notebook, we will use the cleaned, pre-processed data that we created in the [pre-processing part](gutenberg-preprocessing-SOLUTIONS.ipynb). As a reminder, we ended up with an RDD of `(gid, text)` tuples that has been cleaned and we stored it on HDFS at `/user/<YOUR_USERNAME>/gutenberg/cleaned_rdd`. \n",
    "\n",
    "In the [first analysis notebook](gutenberg-analysis-SOLUTIONS.ipynb) we build an N-gram viewer for the gutenberg books project. Now, we will use the corpus to train a simple language classification model using [Spark's machine learning library](http://spark.apache.org/docs/latest/mllib-guide.html)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Set up and launch the Spark runtime\n",
    "\n",
    "Remember from the previous notebook that we have a saved configuration in `./spark_config/` -- so all we need to do is set the `SPARK_CONF_DIR` environment variable and our default configuration will be used: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# specify the configuration directory\n",
    "os.environ['SPARK_CONF_DIR'] = os.path.realpath('./spark_config')\n",
    "\n",
    "# how many cores do we have for the driver\n",
    "ncores = int(os.environ.get('LSB_DJOB_NUMPROC', 1)) \n",
    "\n",
    "# here we set the memory we want spark to use for the driver JVM\n",
    "os.environ['SPARK_DRIVER_MEMORY'] = '%dG'%(ncores*2*0.7)\n",
    "\n",
    "# we have to tell spark which python executable we are using\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": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "conf = SparkConf()\n",
    "\n",
    "sc = SparkContext(master='yarn-client', conf=conf)\n",
    "\n",
    "HTML(yarn_web_app_string.format(yarn_web_app=yarn_web_app))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Load the data from HDFS"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# TODO: load cleaned_rdd from the HDFS\n",
    "cleaned_rdd = <FILL IN>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### A brief aside - correcting for data skew\n",
    "\n",
    "Before we begin the loop-heavy analysis, we can do one more bit of optimization. Not all of the books in the corpus are the same length. This means that when we extract the ngrams, the differences will become even more pronounced. Since our basic unit of work here is a full book, this will result in uneven task execution times. Furthermore, because an individual \"task\" will actually process many books, this means that shorter books will be stuck in the queue behind longer books. \n",
    "\n",
    "To visualize this, lets count the number of characters in each partition: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def count_chars_in_partition(iterator): \n",
    "    \"\"\"\n",
    "    Sum up the string lengths in each partition\n",
    "    \"\"\"\n",
    "    c = 0\n",
    "    for (gid, text) in iterator: \n",
    "        c+=len(text)\n",
    "    yield c    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "plt.hist(cleaned_rdd.mapPartitions(count_chars_in_partition).collect())\n",
    "plt.xlabel('Number of characters per partition')\n",
    "plt.ylabel('Number of partitions')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "It's easy to see that some of the partitions have much more data than others. You can also verify this by looking at the Spark web UI for your job and clicking on the last stage that just executed (it will be named \"collect\"). You should see something like this at the top of the page: \n",
    "\n",
    "![Task summary](../../slides/figs/summary_tasks.png)\n",
    "\n",
    "You can see that the longest tasks are taking several times more to execute than the mean even for this extremely simple operation. The data skew is clear if you scroll down a bit more and see the actual data input into each task and sort by duration: \n",
    "\n",
    "![Task duration](../../slides/figs/task_duration.png)\n",
    "\n",
    "The longest few tasks are processing considerably more data than even the next several. Such a data skew may not matter for certain applications, but in our case we can expect it to make a difference. \n",
    "\n",
    "To try and correct it, we must first realize that for our application we have very few requirements about text order. We don't care about the order of words, as long as the documents are tagged with the correct language. So splitting a book into two and processing them out of order doesn't really matter. \n",
    "\n",
    "With this in mind, we can use a function like `chunk_text` below that creates approximately equal-sized \"blocks\" of text. The blocking means we'll get some spurious character ngrams at the beginning and end of each document, but if we block the data in large enough chunks this shouldn't matter too much."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def chunk_text(text, block_len=10000) : \n",
    "    \"\"\"\n",
    "    Ensure that blocks of text are no larger than `block_len`.\n",
    "    \n",
    "    Arguments: \n",
    "        \n",
    "        text: the raw text string\n",
    "    \"\"\"\n",
    "    len_text = len(text)\n",
    "    \n",
    "    if len_text < block_len : \n",
    "        return text\n",
    "    else : \n",
    "        nblocks = len_text/block_len\n",
    "        return [text[i*block_len:(i+1)*block_len] for i in xrange(nblocks)]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we apply this function to the `cleaned_rdd` and visualize the partition size distribution again. Since we want to keep the keys untouched, we can use the `flatMapValues` RDD method. However, since this method keeps the same data within each partition, we also need to *repartition* the data afterwards to randomize the chunk distributions. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "plt.hist(cleaned_rdd.flatMapValues(chunk_text)\n",
    "                    .repartition(400)\n",
    "                    .mapPartitions(count_chars_in_partition)\n",
    "                    .collect())\n",
    "plt.xlabel('Number of characters per partition')\n",
    "plt.ylabel('Number of partitions')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The distribution is now still not perfect, but we have corrected it considerably. Lets make a restructured version of the full dataset and cache it: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "equalized_rdd = (cleaned_rdd.flatMapValues(chunk_text)\n",
    "                            .setName('equalized_rdd')\n",
    "                            .repartition(400)).cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Load in the metadata dictionary and broadcast it\n",
    "\n",
    "Just as in the previous notebook, we will load our pre-generated metadata dictionary and broadcast it to all the executors. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from cPickle import load\n",
    "\n",
    "with open('{home}/gutenberg_metadata.dump'.format(home=os.environ['HOME']), 'r') as f :\n",
    "    meta_dict = load(f)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# TODO: create meta_b by broadcasting meta_dict\n",
    "meta_b = <FILL IN>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now, our `cleaned_rdd` contains `gid`'s as keys and text as values and if we want some other piece of metadata, we can just access it via the lookup table, for example `meta_b.value[gid][meta_name]`. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We will use the same `extract_ngrams` and `vectorize_doc` functions as in the previous notebook: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from pyspark.mllib.linalg import SparseVector\n",
    "\n",
    "def extract_ngrams(tokens, ngram_range=[1,1], select_ngrams = None, ngram_type='word'):\n",
    "    \"\"\"\n",
    "    Turn tokens into a sequence of n-grams \n",
    "\n",
    "    Arguments:\n",
    "\n",
    "        tokens: a list of tokens\n",
    "\n",
    "    Optional Keywords:\n",
    "\n",
    "        ngram_range: a tuple with min, max ngram ngram_range\n",
    "    \n",
    "        select_ngrams: the vocabulary to use\n",
    "    \n",
    "        ngram_type: whether to produce word or character ngrams\n",
    "\n",
    "    Output:\n",
    "\n",
    "    Generator yielding a list of ngrams in the desired range\n",
    "    generated from the input list of tokens\n",
    "\n",
    "    \"\"\"\n",
    "    join_str = \"\" if ngram_type=='character' else \" \"\n",
    "    \n",
    "    # handle token n-grams\n",
    "    min_n, max_n = ngram_range\n",
    "    n_tokens = len(tokens)\n",
    "    \n",
    "    for n in xrange(min_n, min(max_n + 1, n_tokens + 1)):\n",
    "        for i in xrange(n_tokens - n + 1):\n",
    "            if n == 1: \n",
    "                res = tokens[i]\n",
    "            else : \n",
    "                res = join_str.join(tokens[i: i+n])\n",
    "           \n",
    "            # if we are using a lookup vocabulary, check for membership here\n",
    "            if select_ngrams is not None : \n",
    "                if res in select_ngrams: \n",
    "                    yield res\n",
    "            else : \n",
    "                yield res\n",
    "            \n",
    "def vectorize_doc(doc, vocab, ngram_range = [1,1], ngram_type='word') : \n",
    "    \"\"\"\n",
    "    Returns a vector representation of `doc` given the reference \n",
    "    vocabulary `vocab` after tokenizing it with `tokenizer`\n",
    "    \n",
    "    Arguments: \n",
    "        \n",
    "        doc: a sequence of tokens (words or characters)\n",
    "        \n",
    "        vocab: the vocabulary mapping\n",
    "        \n",
    "    Keywords:\n",
    "    \n",
    "        ngram_range: the range of ngrams to process\n",
    "        \n",
    "        ngram_type: whether to produce character or word ngrams; default is \n",
    "        \n",
    "    Returns:\n",
    "    \n",
    "        a sparse vector representation of the document given the vocabulary mapping\n",
    "    \"\"\"\n",
    "    from collections import defaultdict\n",
    "    from scipy.sparse import csr_matrix \n",
    "        \n",
    "    # count all the word occurences \n",
    "    data = np.zeros(len(vocab))\n",
    "    \n",
    "    for ngram in extract_ngrams(doc, ngram_range, vocab, ngram_type) : \n",
    "         data[vocab[ngram]] += 1\n",
    "            \n",
    "    return data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Language classification\n",
    "\n",
    "Here we will try to use some of the same techniques we developed before, but apply them to a classification problem: determining whether a text is English or German. \n",
    "\n",
    "We will use the rather straightforward method outlined in [Cavnar & Trenkle 1994](http://odur.let.rug.nl/~vannoord/TextCat/textcat.pdf):\n",
    "\n",
    "For each of the English/German training sets:\n",
    "\n",
    "1. tokenize the text (spaces are also tokens, so we replace them with \"_\")\n",
    "2. extract N-grams where 1 < N < 5\n",
    "3. determine the most common N-grams for each corpus\n",
    "4. encode both sets of documents using the combined top ngrams\n",
    "\n",
    "\n",
    "\n",
    "In the last notebook, we used words as \"tokens\" -- now we will use characters, even accounting for white space (which we will replace with \"_\"). We will use the two example sentences again:\n",
    "\n",
    "    document 1: \"a dog bit me\"\n",
    "    document 2: \"i bit the dog back\"\n",
    "    \n",
    "First, we use the `extract_ngrams` function: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "s1 = \"a dog bit me\"\n",
    "s2 = \"i bit the dog back\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "ngrams1 = list(extract_ngrams(s1.replace(' ','_'), ngram_range=[1,5], ngram_type='character'))\n",
    "ngrams2 = list(extract_ngrams(s2.replace(' ','_'), ngram_range=[1,5], ngram_type='character'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "print(list(ngrams1))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "We can create the vocabulary by getting the set of all ngrams and building a lookup table:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "vocab = set(ngrams1) | set(ngrams2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "vocab_dict = {word:ind for ind,word in enumerate(vocab)}\n",
    "print('number of ngrams: ',len(vocab_dict))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note that extracting ngrams can increase the size of the data quite a lot!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "vocab_dict"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And finally, we can use `vectorize_doc` with the vocabulary mapping to turn the documents into vectors: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "vectorize_doc(s1.replace(' ','_'), vocab_dict, ngram_range=[1,5], ngram_type='character')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Saving memory consumption with `mapPartitions`\n",
    "\n",
    "As you could see from the simple example of extracting ngrams from two short phrases above, by extracting the 1-5 grams from a simple 12-character string, we created a vector with 48 stored values. Our actual documents will therefore swell in size very rapidly -- we definitely don't want to be holding all of those huge lists in memory! \n",
    "\n",
    "Remember, our first goal is to get the top most-used N-grams. For this, we just need to create an RDD of N-grams and to avoid building lists we'll use the technique of generators discussed on the first day. \n",
    "\n",
    "Note that the `extract_ngrams` function above is already a generator: now we just want to make a small wrapper function that uses `extract_ngrams` to \"yield\" ngrams one by one into the RDD. \n",
    "\n",
    "A slight complication is that `mapPartitions` gives us a partition *iterator* over the data in the partition. This iterator will give us partition data one element at a time, which in this case are just the individual documents. We can pass each of these documents to `extract_ngrams`. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from collections import defaultdict\n",
    "\n",
    "def ngram_generator(iterator, ngram_range=[1,1], ngram_type='word') : \n",
    "    \"\"\"Take an iterator of documents and create a generator of ngrams\n",
    "    \n",
    "    Arguments:\n",
    "        \n",
    "        iterator: the document iterator\n",
    "        \n",
    "    Keywords:\n",
    "        \n",
    "        ngram_range: the range of ngrams to consider\n",
    "        \n",
    "        ngram_type: whether to extract word or character ngrams; default is word\n",
    "    \"\"\" \n",
    "    for text in iterator : \n",
    "        for ngram in extract_ngrams(text.replace(' ', '_'), ngram_range, ngram_type=ngram_type): \n",
    "            yield ngram"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We will subsample the english part of `equalized_rdd` in order to make this next set of cells complete in a reasonable amount of time -- once it's working, you can go back and do it for the full dataset, but it will take approximately 30 minutes. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# TODO: create english_rdd and german_rdd by filtering the equalized_rdd for 'en' and 'de' \n",
    "# make sure you only take a 10% sample of the english RDD by using the 'sample' method\n",
    "\n",
    "english_rdd = equalized_rdd.<FILL IN>.setName('english_rdd').cache()\n",
    "german_rdd  = equalized_rdd.<FILL IN>.setName('german_rdd').cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Making the sets of most common english and german ngrams\n",
    "\n",
    "To build the sets of ngrams, we use the now-familiar pattern: \n",
    "\n",
    "1. map the documents in the RDDs to their constituent ngrams (here we use the `mapPartitions` method with the `ngram_generator` defined above)\n",
    "2. do the distributed key count using the `map` --> `(key, 1)` --> `reduceByKey` pattern\n",
    "3. sort the result (in descending order) and take the top 1000 ngrams"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "ngram_range = [1,3] # should use 1-5 ngram range, but make it smaller to speed up the processing a bit\n",
    "\n",
    "# TODO: follow the three steps above -- remember that you need to pass the ngram_range to ngram_generator \n",
    "en_ngram_counts = (english_rdd.values()\n",
    "                               <FILL IN>.cache())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# TODO: follow the three steps above -- remember that you need to pass the ngram_range to ngram_generator\n",
    "de_ngram_counts = (german_rdd.values()\n",
    "                               <FILL IN>.cache())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%%time\n",
    "top_1000_en_ngrams = (en_ngram_counts.sortBy(lambda (ngram,count): count, False)\n",
    "                                     .map(lambda (ngram, count): ngram)\n",
    "                                     .take(1000))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%%time\n",
    "top_1000_de_ngrams = (de_ngram_counts.sortBy(lambda (ngram,count): count, False)\n",
    "                                     .map(lambda (ngram, count): ngram)\n",
    "                                     .take(1000))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# building the top_ngrams dictionaries\n",
    "\n",
    "# combine the german and english ngrams\n",
    "top_ngrams = set(top_1000_de_ngrams) | set(top_1000_en_ngrams)\n",
    "\n",
    "# build the ngrams dictionary lookup\n",
    "top_ngrams_dict = {ngram:i for (i,ngram) in enumerate(top_ngrams)}\n",
    "\n",
    "# broadcast the ngrams dictionary\n",
    "top_ngrams_dict_b = sc.broadcast(top_ngrams_dict)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Vectorizing the documents based on top 1000 ngrams from each language\n",
    "\n",
    "For our logistic regression application, we are going to want to balance out the number of documents from the two languages. We know that English is by far over-represented, so we'll create a list of english and german document ID's that will balance out the two samples. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "german_ids = cleaned_rdd.map(lambda (gid, _): gid).filter(lambda gid: meta_b.value[gid]['lang'] == 'de').collect()\n",
    "english_ids = cleaned_rdd.map(lambda (gid, _): gid).filter(lambda gid: meta_b.value[gid]['lang'] == 'en').collect()\n",
    "\n",
    "# here we ensure that the number of english ids is only as long as the list of german ids\n",
    "combined_ids = set(german_ids + english_ids[:len(german_ids)]) "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# create an RDD of german and english by only keeping the gids that are in combined_ids set\n",
    "en_de_rdd = equalized_rdd.<FILL IN>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can vectorize the documents just like we did in the previous notebook by using `vectorize_doc` with the broadcast vocabulary lookup we made above (`top_ngrams_dict_b` - remember to pass in its `value`!) and the pre-defined `ngram_range`. Be sure to change all spaces to '_' before passing in to `vectorize_doc` and to specify `ngram_type`. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# use vectorize_doc to map the en_de_rdd into (gid, vector) tuples \n",
    "# (remember to replace all spaces with '_' before passing text to vectorize_doc)\n",
    "vector_rdd = (en_de_rdd.map(<FILL IN>\n",
    "                                        .cache())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%time vector_rdd.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Since we split the documents to equalize the vectorization load, we must now add the vectors for a given document back together: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# TODO: make 'combined_rdd' by calling 'reduceByKey' on 'vector_rdd' \n",
    "# to add up all of the vectors for a given gid\n",
    "combined_rdd = vector_rdd.<FILL IN>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Train the language classification model\n",
    "\n",
    "Most Spark MLlib functions expect either plain vectors (for unsupervised learning) or `LabeledPoint` instances for supervised algorithms. Therefore, to train the model, we need to first map the `vector_rdd` elements into [`LabeledPoint`](http://spark.apache.org/docs/latest/api/python/pyspark.mllib.html?highlight=labeledpoint#pyspark.mllib.regression.LabeledPoint), which is just a Spark abstraction that encompases a *label* and a *vector*. We then split the data into a training and validation sets and produce the trained model. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.mllib.feature import LabeledPoint\n",
    "from pyspark.mllib.classification import LogisticRegressionWithSGD"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# making a simple LabeledPoint with label '0' and vector [1,2,3,4]\n",
    "LabeledPoint(0, [1,2,3,4])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "First, create a `vector_lp` RDD by mapping the contents of `vector_rdd` into a `LabeledPoint` using 0 if the language is English and 1 if it is German with the provided `gid_to_label` function:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# TODO: create an RDD of LabeledPoint with 0 for english and 1 for german\n",
    "def gid_to_label(gid):\n",
    "    \"\"\"Simple helper function that converts gid to 0 for english or 1 for german\"\"\"\n",
    "    lang = meta_b.value[gid]['lang']\n",
    "    if lang == 'en':\n",
    "        return 0\n",
    "    elif lang == 'de':\n",
    "        return 1\n",
    "    else:\n",
    "        raise RuntimeError(\"Can't handle language '{lang}'\".format(lang=lang))\n",
    "    \n",
    "vector_lp = combined_rdd.map(lambda (gid, vec): <FILL IN>)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The Spark machine learning library provides a simple method for creating training and validation sets, which we will use below. These will be our inputs for the logistic regression model fitting -- it is *always* a good idea to cache the inputs, since the training requires many iterations over the data. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "training, validation = vector_lp.randomSplit([0.7,0.3])\n",
    "training.cache().count()\n",
    "validation.cache().count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Pass the training set to the model\n",
    "\n",
    "Here we will use the basic logistic regression model with stochastic gradient descent -- feel free to experiment with different parameters and other models from [pyspark.mllib.classification](http://spark.apache.org/docs/latest/api/python/pyspark.mllib.html#module-pyspark.mllib.classification)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "model = LogisticRegressionWithSGD.train(training, regType='l1', regParam=0.2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "plt.hist(model.weights, bins=100, range=[-50,50]);"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To check the performance of the model, we define a function that takes a data RDD and a model as parameters and computes the error:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def check_model(data, model) : \n",
    "    \"\"\"Calculates the model error on the data\n",
    "    \n",
    "    Arguments: \n",
    "        \n",
    "        data: the data RDD\n",
    "        \n",
    "        model: the classification model\n",
    "        \n",
    "    Returns:\n",
    "    \n",
    "        the error, which is the fraction of incorrectly predicted elements\n",
    "    \"\"\"\n",
    "    \n",
    "    error = (data.map(lambda p: (p.label, model.predict(p.features)))\n",
    "                 .filter(lambda (v,p): v!=p).count())/float(data.count())\n",
    "    return error"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "train_error = check_model(training, model)\n",
    "print(\"Training Error = \" + str(train_error))\n",
    "validation_error = check_model(validation, model)\n",
    "print(\"Validation Error = \" + str(validation_error))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For fun, lets create a function that will score a new string of text: "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def predict_language(text, model) : \n",
    "    \"\"\"Predict the language given the pre-trained model\n",
    "    \n",
    "    Arguments: \n",
    "        text: a string\n",
    "        \n",
    "        model: the trained logistic regression model\n",
    "    \"\"\"\n",
    "    vec = vectorize_doc(text.replace(' ','_'), top_ngrams_dict_b.value, [1,2])\n",
    "    \n",
    "    res = 'german' if model.predict(vec) == 1 else 'english'\n",
    "    print text + ' is ' + res\n",
    "\n",
    "predict_lang_partial = lambda text: predict_language(text, model)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Try it out! Enter your own sentence:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "interact(predict_lang_partial, text='a dog bit me')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Which character sequences define each language? \n",
    "\n",
    "Here we'll use the weights of the trained model to identify the words with the lowest weights (i.e. those that pull a prediction toward zero) and the highest weights (those that pull a prediction toward 1). "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sorted_weights = np.argsort(model.weights)\n",
    "\n",
    "top_ngram_inds = {v:k for k,v in top_ngrams_dict.iteritems()}\n",
    "\n",
    "print('top 50 english and german ngrams:')\n",
    "for i in range(50): \n",
    "    eng_ind = sorted_weights[i]\n",
    "    ger_ind = sorted_weights[len(top_ngram_inds)-i-1]\n",
    "    print('%s\\t%d\\t%s\\t%d'%(top_ngram_inds[eng_ind], model.weights[eng_ind], \n",
    "                            top_ngram_inds[ger_ind], model.weights[ger_ind]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "Some other ideas for things you could do with this dataset: \n",
    "\n",
    "* try other [classifiers that are included in MLlib](http://spark.apache.org/docs/latest/mllib-classification-regression.html)\n",
    "* build a regression model to predict year of publication (may be better with word ngrams)\n",
    "* run a clustering algorithm on the full language data to identify all the languages present\n",
    "* do clustering on the english books and see if sub-groups of the language pop up\n",
    "* cluster by author -- do certain authors write in similar ways?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "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
}