<jupyter_start><jupyter_text>HW 2 (part 2) - Naive Bayes in Hadoop MR__`MIDS w261: Machine Learning at Scale | UC Berkeley School of Information | Spring 2018`__Now that you're up to speed in writing Hadoop MR jobs we'll use this framework to implement your first parallelized machine learning algorithm: Naive Bayes. As you develop your implementation you'll test it on a small dataset that matches the 'Chinese Example' in the _Manning, Raghavan and Shutze_ reading for Week 2. For the main task in part 2 you'll be working with a small subset of the Enron Spam/Ham Corpus. By the end of this portion of the assignment you should be able to:* __... describe__ the Naive Bayes algorithm including both training and inference.* __... perform__ EDA on a corpus using Hadoop MR.* __... implement__ parallelized Naive Bayes.* __... explain__ how smoothing affects the bias and variance of  a Multinomial Naive Bayes model.As always, your work will be graded both on the correctness of your output and on the clarity and design of your code. __Please refer to the `README` for homework submission instructions.__   Notebook SetupBefore starting part 2, run the following cells to confirm your setup.<jupyter_code># imports
import numpy as np
import matplotlib.pyplot as plt

%reload_ext autoreload
%autoreload 2
# global vars (paths) - ADJUST AS NEEDED
JAR_FILE = "/usr/lib/hadoop-mapreduce/hadoop-streaming-2.6.0-cdh5.7.0.jar"
HDFS_DIR = "/user/root/HW2"
HOME_DIR = "/media/notebooks/SP18-1-maynard242" # FILL IN HERE eg. /media/notebooks/w261-main/Assignments
# save path for use in Hadoop jobs (-cmdenv PATH={PATH})
from os import environ
PATH  = environ['PATH']
# data path
ENRON = HOME_DIR + "/HW02/data/enronemail_1h.txt"<jupyter_output><empty_output><jupyter_text>About the DataFor the main task in this portion of the homework you will train a classifier to determine whether an email represents spam or not. You will train your Naive Bayes model on a 100 record subset of the Enron Spam/Ham corpus available in the HW02 data directory (__`HW02/data/enronemail_1h.txt`__).__Source:__   The original data included about 93,000 emails which were made public after the company's collapse. There have been a number raw and preprocessed versions of this corpus (including those available [here](http://www.aueb.gr/users/ion/data/enron-spam/index.html) and [here](http://www.aueb.gr/users/ion/publications.html)). The subset we will use is limited to emails from 6 Enron employees and a number of spam sources. It is part of [this data set](http://www.aueb.gr/users/ion/data/enron-spam/) which was created by researchers working on personlized Bayesian spam filters. Their original publication is [available here](http://www.aueb.gr/users/ion/docs/ceas2006_paper.pdf). __`IMPORTANT!`__ _For this homework please limit your analysis to the 100 email subset which we provide. No need to download or run your analysis on any of the original datasets, those links are merely provided as context.___Preprocessing:__  For their work, Metsis et al. (the authors) appeared to have pre-processed the data, not only collapsing all text to lower-case, but additionally separating "words" by spaces, where "words" unfortunately include punctuation. As a concrete example, the sentence:  >  `Hey Jon, I hope you don't get lost out there this weekend!`  ... would have been reduced by Metsis et al. to the form:  > `hey jon , i hope you don ' t get lost out there this weekend !` ... so we have reverted the data back toward its original state, removing spaces so that our sample sentence would now look like:> `hey jon, i hope you don't get lost out there this weekend!`  Thus we have at least preserved contractions and other higher-order lexical forms. However, one must be aware that this reversion is not complete, and that some object (specifically web sites) will be ill-formatted, and that all text is still lower-cased.__Format:__   All messages are collated to a tab-delimited format:  >    `ID \t SPAM \t SUBJECT \t CONTENT \n`  where:  >    `ID = string; unique message identifier`      `SPAM = binary; with 1 indicating a spam message`      `SUBJECT = string; title of the message`      `CONTENT = string; content of the message`       Note that either of `SUBJECT` or `CONTENT` may be "NA", and that all tab (\t) and newline (\n) characters have been removed from both of the `SUBJECT` and `CONTENT` columns.<jupyter_code># take a look at t|he first 100 characters of the first 5 records (RUN THIS CELL AS IS)
!head -n 5 {ENRON} | cut -c-100
# NOTE: ADDED Required if running this notebook without first running part 1
!hdfs dfs -mkdir {HDFS_DIR}
# load the data into HDFS (RUN THIS CELL AS IS)
!hdfs dfs -copyFromLocal {ENRON} {HDFS_DIR}/enron.txt<jupyter_output>copyFromLocal: `/user/root/HW2/enron.txt': File exists<jupyter_text>Question 6:  Enron Ham/Spam EDA.Before building our classifier, lets get aquainted with our data. In particular, we're interested in which words occur more in spam emails than in real emails. In this question you'll implement two Hadoop MapReduce jobs to count and sort word occurrences by document class. You'll also learn about two new Hadoop streaming parameters that will allow you to control how the records output of your mappers are partitioned for reducing on separate nodes. __`IMPORTANT NOTE:`__ For this question and all subsequent items, you should include both the subject and the body of the email in your analysis (i.e. concatetate them to get the 'text' of the document). Q6 Tasks:* __a) code:__ Complete the missing components of the code in __`EnronEDA/mapper.py`__ and __`EnronEDA/reducer.py`__ to create a Hadoop MapReduce job that counts how many times each word in the corpus occurs in an email for each class. Pay close attention to the data format specified in the docstrings of these scripts _-- there are a number of ways to accomplish this task, we've chosen this format to help illustrate a technique in `part e`_. Run the provided unit tests to confirm that your code works as expected then run the provided Hadoop streaming command to apply your analysis to the Enron data.* __b) code + short response:__ How many times does the word "__assistance__" occur in each class? (`HINT:` Use a `grep` command to read from the results file you generated in '`a`' and then report the answer in the space provided.)* __c) short response:__ Would it have been possible to add some sorting parameters to the Hadoop streaming command that would cause our `part a` results to be sorted by count? Briefly explain why or why not.* __d) code + short response:__ Write a second Hadoop MapReduce job to sort the output of `part a` first by class and then by count. Run your job and save the results to a local file using the provide code. Then describe in words how you would go about printing the top 10 words in each class. (`HINT 1:` _remember that you can simply pass the `part a` output directory to the input field of this job; `HINT 2:` since this task is just reodering the records from `part a` we don't need to write a mapper or reducer, just use `/bin/cat` for both_)* __ e) code:__ A more efficient alternative to '`grep`-ing' for the top 10 words in each class would be to use the Hadoop framework to separate records from each class into its own partition so that we can just read the top lines in each. To do this, make sure to specify 2 reduce tasks and add the following parameters to your Hadoop streaming job from 'd':>__`-D mapreduce.partition.keypartitioner.options="-k2,2"`__ : tells Hadoop to partition based on the second field (which indicates spam/ham in our data).  >__`-partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner`__ : tells Hadoop that we want to partition on a field that is part of a composite key. (_This parameter should go at the end, not with the `-D` options at the top._)> __`NOTE:`__ these options should be used in conjunction with `-D stream.num.map.output.key.fields=` which you will have already added as part of your sort.  Q6 Student Answers:> __b)__ See cell below. ```assistance	1	8 (spam)assistance	0	2 (ham).```> __c)__ No, because Hadoop sorts by the key (word). We could do a secondary sort but this is premised on the word-key. If we wanted to sort by count, we would need to transpose our data such that count is the key. > __d)__ Pipe the results file, filtering by class, then sorting by count, before printing the top ten. [See cell]<jupyter_code># part a - do your work in the provided scripts then RUN THIS CELL AS IS
!chmod a+x EnronEDA/mapper.py
!chmod a+x EnronEDA/reducer.py
# part a - unit test EnronEDA/mapper.py (RUN THIS CELL AS IS)
!echo -e "d1	1	title	body\nd2	0	title	body" | EnronEDA/mapper.py
# part a - unit test EnronEDA/reducer.py (RUN THIS CELL AS IS)
!echo -e "one	1	1\none	0	1\none	0	1\ntwo	0	1" | EnronEDA/reducer.py
# part a - clear output directory in HDFS (RUN THIS CELL AS IS)
!hdfs dfs -rm -r {HDFS_DIR}/eda-output
# part a - Hadoop streaming job (RUN THIS CELL AS IS)
!hadoop jar {JAR_FILE} \
  -files EnronEDA/reducer.py,EnronEDA/mapper.py \
  -mapper mapper.py \
  -reducer reducer.py \
  -input {HDFS_DIR}/enron.txt \
  -output {HDFS_DIR}/eda-output \
  -numReduceTasks 2
# part a - retrieve results from HDFS & copy them into a local file (RUN THIS CELL AS IS)
!hdfs dfs -cat {HDFS_DIR}/eda-output/part-0000* > EnronEDA/results.txt
# part b - write your grep command here
!grep -i 'assistance' EnronEDA/results.txt
# part d - clear the output directory in HDFS (RUN THIS CELL AS IS)
!hdfs dfs -rm -r {HDFS_DIR}/eda-sort-output
# part d - write your Hadoop streaming job here
!hadoop jar {JAR_FILE} \
  -D stream.num.map.output.key.fields=2 \
  -D mapred.output.key.comparator.class=org.apache.hadoop.mapred.lib.KeyFieldBasedComparator \
  -D mapred.text.key.comparator.options="-k2,2 -k3,3n" \
  -mapper /bin/cat \
  -reducer /bin/cat \
  -input {HDFS_DIR}/eda-output/part-0000* \
  -output {HDFS_DIR}/eda-sort-output \
  -numReduceTasks 2
# part d - print out the top ten words for each class

print "Top Ten Wrods for Ham:\n"
!hdfs dfs -cat {HDFS_DIR}/eda-sort-output/part-0000* | grep -P '^[a-z]+\t0' | sort -k3,3nr | head
print "\n"
print "Top Ten Words for Spam: \n"
!hdfs dfs -cat {HDFS_DIR}/eda-sort-output/part-0000* | grep -P '^[a-z]+\t1' | sort -k3,3nr | head


# part e - clear the output directory in HDFS (RUN THIS CELL AS IS)
!hdfs dfs -rm -r {HDFS_DIR}/eda-sort-output
# part e - write your Hadoop streaming job here
!hadoop jar {JAR_FILE} \
  -D stream.num.map.output.key.fields=3 \
  -D mapred.output.key.comparator.class=org.apache.hadoop.mapred.lib.KeyFieldBasedComparator \
  -D mapred.text.key.comparator.options="-k3,3nr" \
  -D mapreduce.partition.keypartitioner.options="-k2,2" \
  -partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner \
  -mapper /bin/cat \
  -reducer /bin/cat \
  -input {HDFS_DIR}/eda-output/part-0000* \
  -output {HDFS_DIR}/eda-sort-output \
  -numReduceTasks 2
    
# part e - view the top 10 records from each partition (RUN THIS CELL AS IS)
for idx in range(2):
    print "\n===== part-0000%s=====\n" % (idx)
    !hdfs dfs -cat {HDFS_DIR}/eda-sort-output/part-0000{idx} | head<jupyter_output>===== part-00000=====

the	0	549	
to	0	398	
ect	0	382	
and	0	278	
of	0	230	
hou	0	206	
a	0	196	
in	0	182	
for	0	170	
on	0	135	
cat: Unable to write to output stream.

===== part-00001=====

the	1	698	
to	1	566	
and	1	392	
your	1	357	
a	1	347	
you	1	345	
of	1	336	
in	1	236	
for	1	204	
com	1	153	
cat: Unable to write to output stream.<jupyter_text>__Expected output:__part-00000:part-00001:the	0	549	to	0	398	ect	0	382	and	0	278	of	0	230	hou	0	206	a	0	196	in	0	182	for	0	170	on	0	135the	1	698	to	1	566	and	1	392	your	1	357	a	1	347	you	1	345	of	1	336	in	1	236	for	1	204	com	1	153  Question 7: Document Classification Task Overview.The week 2 assigned reading from Chapter 13 of _Introduction to Information Retrieval_ by Manning, Raghavan and Schutze provides a thorough introduction to the document classification task and the math behind Naive Bayes. In this question we'll use the example from Table 13.1 (reproduced below) to 'train' an unsmoothed Multinomial Naive Bayes model and classify a test document by hand.DocIDClassSubjectBodyDoc11Chinese Beijing ChineseDoc21Chinese Chinese ShanghaiDoc31Chinese MacaoDoc40Tokyo Japan Chinese Q7 Tasks:* __a) short response:__ Equation 13.3 in Manning, Raghavan and Shutze shows how a Multinomial Naive Bayes model classifies a document. It predicts the class, $c$, for which the estimated conditional probability of the class given the document's contents,  $\hat{P}(c|d)$, is greatest. In this equation what two pieces of information are required to calculate  $\hat{P}(c|d)$? Your answer should include both mathematical notatation and verbal explanation.* __b) short response:__ The Enron data includes two classes of documents: `spam` and `ham` (they're actually labeled `1` and `0`). In plain English, explain what  $\hat{P}(c)$ and   $\hat{P}(t_{k} | c)$ mean in the context of this data. How will we would estimate these values from a training corpus? How many passes over the data would we need to make to retrieve this information for all classes and all words?* __c) hand calculations:__ Above we've reproduced the document classification example from the textbook (we added an empty subject field to mimic the Enron data format). Remember that the classes in this "Chinese Example" are `1` (about China) and `0` (not about China). Calculate the class priors and the conditional probabilities for an __unsmoothed__ Multinomial Naive Bayes model trained on this data. Show the calculations that lead to your result using markdown and $\LaTeX$ in the space provided or by embedding an image of your hand written work. [`NOTE:` _Your results should NOT match those in the text -- they are training a model with +1 smoothing you are training a model without smoothing_]* __d) hand calculations:__ Use the model you trained to classify the following test document: `Chinese Chinese Chinese Tokyo Japan`. Show the calculations that lead to your result using markdown and   $\LaTeX$ in the space provided or by embedding an image of your hand written work.* __e) short response:__ Compare the classification you get from this unsmoothed model in `d`/`e` to the results in the textbook's "Example 1" which reflects a model with Laplace plus 1 smoothing. How does smoothing affect our inference?  Q7 Student Answers:> __a)__ We need $\hat{P}(c)$ which the unconditional probability of class $c$, and $\hat{P}(t_{k} | c)$ which the probabibility of term $t_{k}$ ocurring in document of class $c$.> __b)__ $\hat{P}(c) = \frac{N_{c}}{N}$, that is the number of spam or ham over the total number of emails. For the conditional probability, $\hat{P}(t_{k} | c) = \frac{\text{word t in class c}}{\text{total words in class c}}$, we need to count the number of words given a class (e.g. if spam, how many times does 'the' appear and then divide it by total number of words in that class. We can estimate these numbers from counting words in the corpus, as we have done before. As before, ***one*** pass is all that is necessary to collect the needed data.> __c)__ $\hat{P}(c=1) = 3/4 =0.75$ and $ \hat{P}(c=0) = 1/4 =0.25.$  > $\hat{P}(\text{Chinese} \, | \, c=1 ) = \frac{5}{8} = 0.625.$ > $\hat{P}(\text{Beijing} \, | \, c=1 ) = \frac{1}{8} = 0.125.$ > $\hat{P}(\text{Shanghai} \, | \, c=1 ) = \frac{1}{8} = 0.125.$ > $\hat{P}(\text{Macao} \, | \, c=1 ) = \frac{1}{8} = 0.125.$ > $\hat{P}(\text{Tokyo} \, | \, c=0 ) = \frac{1}{3} = 0.333.$ > $\hat{P}(\text{Japan} \, | \, c=0 ) = \frac{1}{3} = 0.333.$ > $\hat{P}(\text{Chinese} \, | \, c=0 ) = \frac{1}{3} = 0.333.$ > __d)__ $\hat{P}(d|c=1) \propto 0.75 \times 0.625^3 \times 0 \times 0 = 0 $ and > $\hat{P}(d|c=0) \propto 0.25 \times (1/3)^3 \times 1/3 \times 1/3 = (1/3)^5 $ > Without smoothing, the most likely class = 0, non China, and that is because in our training data we never saw class=1 document with world Tokyo and Japan.> __e)__ With Laplace plus one smoothing, the best class is now 1 (China). This seem intuitively correct, and we get this result because smoothing assigns some small probability to seeing words Tokyo and Japan in class 1.<jupyter_code># part d/e - if you didn't write out your calcuations above, embed a picture of them here:
from IPython.display import Image
Image(filename="path-to-hand-calulations-image.png")<jupyter_output><empty_output><jupyter_text>Question 8: Naive Bayes Inference.In the next two questions you'll write code to parallelize the Naive Bayes calculations that you performed above. We'll do this in two phases: one MapReduce job to perform training and a second MapReduce to perform inference. While in practice we'd need to train a model before we can use it to classify documents, for learning purposes we're going to develop our code in the opposite order. By first focusing on the pieces of information/format we need to perform the classification (inference) task you should find it easier to develop a solid implementation for training phase when you get to question 9 below. In both of these questions we'll continue to use the Chinese example corpus from the textbook to help us test our MapReduce code as we develop it. Below we've reproduced the corpus, test set and model in text format that matches the Enron data. Q8 Tasks:* __a) short response:__ run the provided cells to create the example files and load them in to HDFS. Then take a closer look at __`chinese/model.txt`__. This text file represents a Naive Bayes model trained (with Laplace +1 smoothing) on the example corpus. What are the 'keys' and 'values' in this file? Which record means something slightly different than the rest? The value field of each record includes two numbers which will be helpful for debugging but which we don't actually need to perform inference -- what are they? [`HINT`: _This file represents the model from Example 13.1 in the textbook, if you're having trouble getting oriented try comparing our file to the numbers in that example._]* __b) short response:__ When performing Naive Bayes in practice instead of multiplying the probabilities (as in equation 13.3) we add their logs (as in equation 13.4). Why do we choose to work with log probabilities? If we had an unsmoothed model, what potential error could arise from this transformation?* __c) short response:__ Documents 6 and 8 in the test set include a word that did not appear in the training corpus (and as a result does not appear in the model). What should we do at inference time when we need a class conditional probability for this word?* __d) short response:__ The goal of our MapReduce job is to stream over the test set and classify each document by peforming the calculation from equation 13.4. To do this we'll load the model file (which contains the probabilities for equation 13.4) into memory on the nodes where we do our mapping. This is called an in-memory join, we'll learn more about those in Week 5. For now, explain how this is a slight departure from one of the functional programming principles. From a scalability perspective when might this departure be justified? when would it be unwise?* __e) code:__ Complete the code in __`NaiveBayes/classify_mapper.py`__. Read the docstring carefully to understand how this script should work and the format it should return. Run the provided unit tests to confirm that your script works as expected then write a Hadoop streaming job to classify the Chinese example test set. [`HINT 1:` _you shouldn't need a reducer for this one._ `HINT 2:` _Don't forget to add the model file to the_ `-files` _parameter in your Hadoop streaming job so that it gets shipped to the mapper nodes where it will be accessed by your script._]* __f) short response:__ In our test example and in the Enron data set we have fairly short documents. Since these fit fine in memory on a mapper node we didn't need a reducer and could just do all of our calculations in the mapper. However with much longer documents (eg. books) we might want a higher level of parallelization -- for example we might want to process parts of a document on different nodes. In this hypothetical scenario how would our algorithm design change? What could the mappers still do? What key-value structure would they emit? What would the reducers have to do as a last step?  Q8 Student Answers:> __a)__ The key here are the words, followed by values for, count in class 0, count in class 1, conditional probability in class 0, conditional probability in class 1. The line for ClassPriors constains the conditional class count (c=0, then c=1), and correspondng probabilities. Strictly speaking we don't need counts for ham and spam (or Chinese, non-Chinese).> __b)__ We have two reasons: 1) avoiding underflow errors (from multiply very small probabilities together), and 2) in an unsmoothed model, zero probabibility when only one estimate is zero (x*0 = 0). See example in Q7.> __c)__ The word trade is missing from our model. Hence we ignore it, or set the  $\hat{P} (\,Trade\, |\, c ) = 0$. > __d)__ The model is "hardcoded" and not derived from the data. In memory join might be justified if we have enough memory to run the jobs and if Hadoops "suffle and sort" becomes a bottle neck i.e. there is a lot of key-value pairs to process. It becomes unwise if our mapper nodes does not have enough memory. > __e)__ Complete the coding portion of this question before answering 'f'.> __f)__ If we had huge data set then we should go back to our word counting 'model' for mappers. Mappers would not load a model into memory, but rather emit the model parameters. We would then use reducers to calculate propabilities, and make the final prediction. Run these cells to create the example corpus and model.<jupyter_code>%%writefile NaiveBayes/chineseTrain.txt
D1	1		Chinese Beijing Chinese
D2	1		Chinese Chinese Shanghai
D3	1		Chinese Macao
D4	0		Tokyo Japan Chinese
%%writefile NaiveBayes/chineseTest.txt
D5	1		Chinese Chinese Chinese Tokyo Japan
D6	1		Beijing Shanghai Trade
D7	0		Japan Macao Tokyo
D8	0		Tokyo Japan Trade
%%writefile NBmodel.txt
beijing	0.0,1.0,0.111111111111,0.142857142857
chinese	1.0,5.0,0.222222222222,0.428571428571
tokyo	1.0,0.0,0.222222222222,0.0714285714286
shanghai	0.0,1.0,0.111111111111,0.142857142857
ClassPriors	1.0,3.0,0.25,0.75
japan	1.0,0.0,0.222222222222,0.0714285714286
macao	0.0,1.0,0.111111111111,0.142857142857
# load the data files into HDFS
!hdfs dfs -copyFromLocal NaiveBayes/chineseTrain.txt {HDFS_DIR}
!hdfs dfs -copyFromLocal NaiveBayes/chineseTest.txt {HDFS_DIR}<jupyter_output>copyFromLocal: `/user/root/HW2/chineseTrain.txt': File exists
copyFromLocal: `/user/root/HW2/chineseTest.txt': File exists<jupyter_text>Your work for `part e` starts here:<jupyter_code># part e - do your work in NaiveBayes/classify_mapper.py first, then run this cell.
!chmod a+x NaiveBayes/classify_mapper.py
# part e - unit test NaiveBayes/classify_mapper.py (RUN THIS CELL AS IS)
!cat NaiveBayes/chineseTest.txt | NaiveBayes/classify_mapper.py
# part e - clear the output directory in HDFS (RUN THIS CELL AS IS)
!hdfs dfs -rm -r {HDFS_DIR}/chinese-output
# part e - write your Hadooop streaming job here
!hadoop jar {JAR_FILE} \
  -files NaiveBayes/classify_mapper.py,NBmodel.txt \
  -mapper classify_mapper.py \
  -reducer /bin/cat \
  -cmdenv PATH=/opt/anaconda/bin:$PATH \
  -input {HDFS_DIR}/chineseTest.txt \
  -output {HDFS_DIR}/chinese-output 
# part e - retrieve test set results from HDFS
!hdfs dfs -cat {HDFS_DIR}/chinese-output/part-000* > NaiveBayes/chineseResults.txt
# part e - take a look
!cat NaiveBayes/chineseResults.txt<jupyter_output>d5	1	0.0001354807	0.0003012138	1
d6	1	0.0030864198	0.0153061224	1
d7	0	0.0013717421	0.0005466472	0
d8	0	0.0123456790	0.0038265306	0<jupyter_text>Expected output for the test set:d5	1	0.00013548	0.00030121	1d6	1	0.00308641	0.01530612	1d7	0	0.00137174	0.00054664	0d8	0	0.01234567	0.00382653	0  Question 9: Naive Bayes Training.In Question 8 we used a model that we had trained by hand. Next we'll develop the code to do that same training in parallel, making it suitable for use with larger corpora (like the Enron emails). The end result of the MapReduce job you write in this question should be a model text file that looks just like the example (`NBmodel.txt`) that we created by hand above.To refresh your memory about the training process take a look at  `7a` and `7b` where you described the pieces of information you'll need to collect in order to encode a Multinomial Naive Bayes model. We now want to retrieve those pieces of information while streaming over a corpus. The bulk of the task will be very similar to the word counting excercises you've already done but you may want to consider a slightly different key-value record structure to efficiently tally counts for each class. The most challenging (interesting?) design question will be how to retrieve the totals ( of documents and  of words in documents for each class). Of course, counting these numbers is easy. The hard part is the timing: you'll need to make sure you have the counts totalled up _before_ you start estimating the class conditional probabilities for each word. It would be best (i.e. most scalable) if we could find a way to do this tallying without storing the whole vocabulary in memory... _can you see why this is gong to be a challenge?_ [`HINT:` _There is a problem with trying to tally in the mapper but also a problem with waiting to do it in the reducer. _]. __For part `b` of this question you will receive full credit for any design that results in the correct conditional probabilities _without hard coding the class totals_ (i.e. you _can_ store records in memory if you so choose).__ However, for fun, we challenge you to try and figure out the stateless solution. It involves a technique that you will learn formally next week but we think you may be able to figure it out for yourself. [`HINT:` _Think about Hadoop's default sorting behavior, what determines the order in which records arrive at the reducer?  If we compute partial totals in our mappers can you think of a way to make sure that those numbers arrive at our reducer before any of the other 'regular' key-value pairs?_`]__`IMPORTANT NOTE:`__ Please use a single reducer for all of the jobs in this question. Q9 Tasks:* __a) make a plan:__  Fill in the docstrings for __`NaiveBayes/train_mapper.py`__ and __`NaiveBayes/train_reducer.py`__ to appropriately reflect the format that each script will input/output. [`HINT:` _the input files_ (`enronemail_1h.txt` & `chineseTrain.txt`) _have a prespecified format and your output file should match_ `NBmodel.txt` _so you really only have to decide on an internal format for Hadoop_].* __b) implement it:__ Complete the code in __`NaiveBayes/train_mapper.py`__ and __`NaiveBayes/train_reducer.py`__ so that together they train a Multinomial Naive Bayes model __with no smoothing__. Make sure your end result is formatted correctly (see note above). Test your scripts independently and together (using `chineseTrain.txt` or test input of your own devising). When you are satisfied with your Python code design and run a Hadoop streaming command to run your job in parallel on the __chineseTrain.txt__. Confirm that your trained model matches your hand calculations from Question 7.* __c) short response:__ We saw in Question 7 that adding Laplace +1 smoothing makes our classifications less sensitve to rare words. However implementing this technique requires access to one additional piece of information that we had not previously used in our Naive Bayes training. What is that extra piece of information? [`HINT:` see equation 13.7 in Manning, Raghavan and Schutze].* __d) short response:__ There are three approaches that we could take to handle the extra piece of information you identified in `c`: 1) we could hard code it into our reducer (_where would we get it in the first place?_). Or 2) we could compute it inside the reducer which would require storing some information in memory (_what information?_). Or 3) we could compute it in the reducer without storing any bulky information in memory but then we'd need some postprocessing or a second MapReduce job to complete the calculation (_why?_). Breifly explain what is non-ideal about each of these options. BONUS: which of these options is incompatible with using multiple reducers?* __e) code + short response:__ In future weeks we'll learn better ways to handle this kind of situation. For now, chose one of the 3 options above. State your choice & reasoning in the space below then use that strategy to complete the code in __`NaiveBayes/train_reducer_smooth.py`__. Test this alternate reducer then write and run a Hadoop streaming job to train an MNB model with smoothing on the Chinese example. Your results should match the model that we provided for you in Question 9 (and the calculations in the textbook example). [`HINT:` _don't start from scratch with this one -- you can just copy over your reducer code from part `b` and make the needed modifications_].  Q9 Student Answers:> __ c)__ Size of the vocabulary i.e. number of distinct words.> __ d)__ Option 1 is not ideal because it makes our reducer less generalizable since the numbers are hard-coded. Option 2 may not work if required memory on the reducer is large (at least, it won't be so scalable). The memory problem is exacerbated since we have to keep everything in memory. I also cannot see how this method will allow multiple reducers since we will need to count total vocab and this has to be done globally. Option 3 requires us to run another mapping job, or to post-process the data.  > __ e)__ I choose to implement it via variant of Option 1 where the mapper counts the vocab and emits it (the string of distinct words) and is read by the reducer. This method does not rely on full memory join on the reducer (memory is confined to mapper nodes), does not require post-processing, and can still be run on two reducers (max).<jupyter_code># part a - do your work in train_mapper.py and train_reducer.py then RUN THIS CELL AS IS
!chmod a+x NaiveBayes/train_mapper.py
!chmod a+x NaiveBayes/train_reducer.py
!echo "=========== MAPPER DOCSTRING ============"
!head -n 8 NaiveBayes/train_mapper.py | tail -n 6
!echo "=========== REDUCER DOCSTRING ============"
!head -n 8 NaiveBayes/train_reducer.py | tail -n 6<jupyter_output>=========== MAPPER DOCSTRING ============
Mapper reads in text documents and emits word counts by class.
INPUT:
    ID \t SPAM \t SUBJECT \t CONTENT \n
OUTPUT:
    word \t class \t count 
    
=========== REDUCER DOCSTRING ============
Reducer aggregates word counts by class and emits frequencies.

INPUT:
    word \t class \t partialCount 
OUTPUT:
    word \t ham_count \t spam_count \t (P(word|ham)) \t P(word|spam)<jupyter_text>__`Q9 part c starts here`:__ MNB _without_ Smoothing (training on Chinese Example Corpus).<jupyter_code># part b - unit test for mapper
!cat NaiveBayes/chineseTrain.txt | NaiveBayes/train_mapper.py
# part b - unit test for reducer
!echo -e "0ClassPriors\t1\t3\n0ClassPriors\t0\t1\n1PartialCounts\t1\t8\n1PartialCounts\t0\t3\none	1	1\none	0	1\none	0	1\ntwo	0	1" | NaiveBayes/train_reducer.py
# part b - system test
!cat NaiveBayes/chineseTrain.txt | NaiveBayes/train_mapper.py | sort | NaiveBayes/train_reducer.py
# part b - clear (and name) an output directory in HDFS for your unsmoothed chinese NB model
!hdfs dfs -rm -r {HDFS_DIR}/chinese-nbmodel
# part b - write your hadoop streaming job
!hadoop jar {JAR_FILE} \
  -files NaiveBayes/train_mapper.py,NaiveBayes/train_reducer.py \
  -mapper train_mapper.py \
  -reducer train_reducer.py \
  -input {HDFS_DIR}/chineseTrain.txt \
  -output {HDFS_DIR}/chinese-nbmodel \
  -numReduceTasks 1

# part b - extract your results (i.e. model) to a local file
!hdfs dfs -cat {HDFS_DIR}/chinese-nbmodel/part-000* > NaiveBayes/chineseNBmodel.txt
# part b - print your model so that we can confirm that it matches expected results
!cat NaiveBayes/chineseNBmodel.txt<jupyter_output>beijing	0.0,1.0,0.000000000000,0.125000000000
chinese	1.0,5.0,0.333333333333,0.625000000000
japan	1.0,0.0,0.333333333333,0.000000000000
macao	0.0,1.0,0.000000000000,0.125000000000
shanghai	0.0,1.0,0.000000000000,0.125000000000
tokyo	1.0,0.0,0.333333333333,0.000000000000
ClassPriors	1.0,3.0,0.250000000000,0.750000000000<jupyter_text>__`Q9 part e starts here`:__ MNB _with_ Smoothing (training on Chinese Example Corpus).<jupyter_code>!cat NaiveBayes/chineseTrain.txt | NaiveBayes/train_mapper.py
# part e - write a unit test for your NEW reducer here
!echo -e "0ClassPriors\t1\t3\n0ClassPriors\t0\t1\n1PartialCounts\t1\t8\n1PartialCounts\t0\t3\none	1	1\none	0	1\none	0	1\ntwo	0	1" | NaiveBayes/train_reducer_smooth.py
# part e - write a systems test for your mapper + reducer together here
!cat NaiveBayes/chineseTrain.txt | NaiveBayes/train_mapper.py | sort | NaiveBayes/train_reducer_smooth.py
# part e - clear (and name) an output directory in HDFS for your SMOOTHED chinese NB model
!hdfs dfs -rm -r {HDFS_DIR}/chinese-smooth-nbmodel
# part e - write your hadoop streaming job
!hadoop jar {JAR_FILE} \
  -files NaiveBayes/train_mapper.py,NaiveBayes/train_reducer_smooth.py \
  -mapper train_mapper.py \
  -reducer train_reducer_smooth.py \
  -input {HDFS_DIR}/chineseTrain.txt \
  -output {HDFS_DIR}/chinese-smooth-nbmodel \
  -numReduceTasks 1
# part e - extract your results (i.e. model) to a local file
!hdfs dfs -cat {HDFS_DIR}/chinese-smooth-nbmodel/part-000* > NaiveBayes/chineseNBmodel2.txt
!cat NaiveBayes/chineseNBmodel2.txt
print '=================vs============\n'
!cat NBmodel.txt<jupyter_output>beijing	0.0,1.0,0.111111111111,0.142857142857
chinese	1.0,5.0,0.222222222222,0.428571428571
japan	1.0,0.0,0.222222222222,0.071428571429
macao	0.0,1.0,0.111111111111,0.142857142857
shanghai	0.0,1.0,0.111111111111,0.142857142857
tokyo	1.0,0.0,0.222222222222,0.071428571429
ClassPriors	1.0,3.0,0.250000000000,0.750000000000
=================vs============

beijing	0.0,1.0,0.111111111111,0.142857142857
chinese	1.0,5.0,0.222222222222,0.428571428571
tokyo	1.0,0.0,0.222222222222,0.0714285714286
shanghai	0.0,1.0,0.111111111111,0.142857142857
ClassPriors	1.0,3.0,0.25,0.75
japan	1.0,0.0,0.222222222222,0.0714285714286
macao	0.0,1.0,0.111111111111,0.142857142857<jupyter_text>Question 10: Enron Ham/Spam NB Classifier & Results.Fantastic work. We're finally ready to perform Spam Classification on the Enron Corpus. In this final homework question you'll run the analysis you've developed, report its performance, and draw some conclusions. Q10 Tasks:* __a) train/test split:__ Run the provided code to split our Enron file into a training set and testing set then load them into HDFS. [`NOTE:` _If you hard coded the vocab size in question 9d make sure you re calculate the vocab size for just the training set!_]* __b) train 2 models:__ Write Hadoop Streaming jobs to train MNB Models on the training set with and without smoothing. Save your models to local files at __`NaiveBayes/Unsmoothed/NBmodel.txt`__ and __`NaiveBayes/Smoothed/NBmodel.txt`__. [`NOTE:` _This naming is important because we wrote our classification task so that it expects a file of that name... if this inelegance frustrates you there is an alternative that would involve a few adjustments to your code [read more about it here](http://www.tnoda.com/blog/2013-11-23). If you can let it slide for now, MRJob (the next framework we'll learn) will have an easier way to handle this._] Finally run the checks that we provide to confirm that your results are correct.* __c) code:__ Recall that we designed our classification job with just a mapper. An efficient way to report the performance of our models would be to simply add a reducer phase to this job and compute precision and recall right there. Complete the code in __`NaiveBayes/evaluation_reducer.py`__ and then write Hadoop jobs to evaluate your two models on the test set. Report their performance side by side. [`NOTE:` if you need a refresher on precision, recall and F1-score [Wikipedia](https://en.wikipedia.org/wiki/F1_score) is a good resource.]* __d) short response:__ Compare the performance of your two models. What do you notice about the unsmoothed model's predictions? Can you guess why this is happening? Which evaluation measure do you think is most relevant in our use case? [`NOTE:` _Feel free to answer using your common sense but if you want more information on evaluating the classification task checkout_ [this blogpost](https://tryolabs.com/blog/2013/03/25/why-accuracy-alone-bad-measure-classification-tasks-and-what-we-can-do-about-it/) or [this paper](http://www.flinders.edu.au/science_engineering/fms/School-CSEM/publications/tech_reps-research_artfcts/TRRA_2007.pdf)]* __e) code + short response:__ Let's look at the top ten words with the highest conditional probability in `Spam` and in `Ham`. We'll do this by writing a Hadoop job that sorts the model file (`NaiveBayes/Smoothed/NBmodel.py`). Normally we'd have to run two jobs -- one that sorts on $P(word|ham)$ and another that sorts on $P(word|spam)$. However if we slighly modify the data format in the model file then we can get the top words in each class with just one job. We've written a mapper that will do just this for you. Read through __`NaiveBayes/model_sort_mapper.py`__ and then briefly explain how this mapper will allow us to partition and sort our model file. Write a Hadoop job that uses our mapper and `/bin/cat` for a reducer to partition and sort. Print out the top 10 words in each class (where 'top' == highest conditional probability).[`HINT:` _this should remind you a lot of what we did in Question 6._]* __f) short response:__ What do you notice about the 'top words' we printed in `e`? How would increasing the smoothing parameter 'k' affect the probabilities for the top words that you identified for 'e'. How would they affect the probabilities of words that occur much more in one class than another? In summary, how does the smoothing parameter 'k' affect the bias and the variance of our model. [`NOTE:` _you do not need to code anything for this task, but if you are struggling with it you could try changing 'k' and see what happens to the test set. We don't recommend doing this exploration with the Enron data because it will be harder to see the impact with such a big vocabulary_]  Q10 Student Answers:> __d)__ On all metrics, the smoothed model does better. This is probably because smoothing allows us to assign some probability to words we saw for spam or ham but are not present respective class in the training set.> __e)__ The mapper splits the sample into spam and ham, the we use the partition option as in the last part of Q6 to sort by maximum conditional probability over the population of spam and ham. > __f)__ The higher the k, the higher the bias (and lower the variance). Many of the words in both classes seem to be common in English (note: one thing to do might be to exclude these words), and if we increase smoothing one would except their conditional probabilities to drop as greater probability mass is transfered to less common words.  __Test/Train split__<jupyter_code># part a - test/train split (RUN THIS CELL AS IS)
!head -n 80 data/enronemail_1h.txt > data/enron_train.txt
!tail -n 20 data/enronemail_1h.txt > data/enron_test.txt
!hdfs dfs -copyFromLocal data/enron_train.txt {HDFS_DIR}
!hdfs dfs -copyFromLocal data/enron_test.txt {HDFS_DIR}<jupyter_output>copyFromLocal: `/user/root/HW2/enron_train.txt': File exists
copyFromLocal: `/user/root/HW2/enron_test.txt': File exists<jupyter_text>__Training__ (Enron MNB Model _without smoothing_ )<jupyter_code># part b -  Unsmoothed model (FILL IN THE MISSING CODE BELOW)

# clear the output directory
!hdfs dfs -rm -r {HDFS_DIR}/enron-model

# hadoop command
!hadoop jar {JAR_FILE} \
  -files NaiveBayes/train_mapper.py,NaiveBayes/train_reducer.py \
  -mapper train_mapper.py \
  -reducer train_reducer.py \
  -input {HDFS_DIR}/enron_train.txt \
  -output {HDFS_DIR}/enron-model \
  -numReduceTasks 1

# save the model locally
!mkdir NaiveBayes/Unsmoothed
!hdfs dfs -cat {HDFS_DIR}/enron-model/part-000* > NaiveBayes/Unsmoothed/NBmodel.txt
# part b - check your UNSMOOTHED model results (RUN THIS CELL AS IS)
!grep assistance NaiveBayes/Unsmoothed/NBmodel.txt
# EXPECTED OUTPUT: assistance	2,4,0.000172547666293,0.000296823983378
# part b - check your UNSMOOTHED model results (RUN THIS CELL AS IS)
!grep money NaiveBayes/Unsmoothed/NBmodel.txt
# EXPECTED OUTPUT: money	1,22,8.62738331464e-05,0.00163253190858<jupyter_output>money	1.0,22.0,0.000086273833,0.001632531909<jupyter_text>__Training__ (Enron MNB Model _with Laplace +1 smoothing_ )<jupyter_code># part b -  Smoothed model (FILL IN THE MISSING CODE BELOW)

# clear the output directory
!hdfs dfs -rm -r {HDFS_DIR}/smooth-model

# hadoop command
!hadoop jar {JAR_FILE} \
  -files NaiveBayes/train_mapper.py,NaiveBayes/train_reducer_smooth.py \
  -mapper train_mapper.py \
  -reducer train_reducer_smooth.py \
  -input {HDFS_DIR}/enron_train.txt \
  -output {HDFS_DIR}/smooth-model \
  -numReduceTasks 1

# save the model locally
!mkdir NaiveBayes/Smoothed
!hdfs dfs -cat {HDFS_DIR}/smooth-model/part-000* > NaiveBayes/Smoothed/NBmodel.txt
# part b - check your SMOOTHED model results (RUN THIS CELL AS IS)
!grep assistance NaiveBayes/Smoothed/NBmodel.txt
# EXPECTED OUTPUT: assistance	2,4,0.000185804533631,0.000277300205202
# part b - check your SMOOTHED model results (RUN THIS CELL AS IS)
!grep money NaiveBayes/Smoothed/NBmodel.txt
# EXPECTED OUTPUT: money	1,22,0.000123869689087,0.00127558094393<jupyter_output>money	1.0,22.0,0.000123869689,0.001275580944<jupyter_text>__Evaluation__<jupyter_code># part c - write your code in NaiveBayes/evaluation_reducer.py then RUN THIS CELL
!chmod a+x NaiveBayes/evaluation_reducer.py
# part c - unit test your evaluation job on the chinese model (RUN THIS CELL AS IS)
!cat NaiveBayes/chineseTest.txt | NaiveBayes/classify_mapper.py 
!cat NaiveBayes/chineseTest.txt | NaiveBayes/classify_mapper.py | NaiveBayes/evaluation_reducer.py
# part c - Evaluate the UNSMOOTHED Model Here (FILL IN THE MISSING CODE)

# clear output directory
!hdfs dfs -rm -r {HDFS_DIR}/evaluation-unsmooth

# hadoop job
!hadoop jar {JAR_FILE} \
  -files NaiveBayes/classify_mapper.py,NaiveBayes/evaluation_reducer.py,NaiveBayes/Unsmoothed/NBmodel.txt \
  -mapper classify_mapper.py \
  -reducer evaluation_reducer.py \
  -cmdenv PATH=/opt/anaconda/bin:$PATH \
  -input {HDFS_DIR}/enron_test.txt \
  -output {HDFS_DIR}/evaluation-unsmooth \
  -numReduceTasks 1

# save the model locally
!hdfs dfs -cat {HDFS_DIR}/evaluation-unsmooth/part-000* > NaiveBayes/Unsmoothed/results.txt

# retrieve results locally
#!cat NaiveBayes/Unsmoothed/results.txt

# part c - Evaluate the SMOOTHED Model Here (FILL IN THE MISSING CODE)

# clear output directory
!hdfs dfs -rm -r {HDFS_DIR}/evaluation-smooth

# hadoop job
!hadoop jar {JAR_FILE} \
  -files NaiveBayes/classify_mapper.py,NaiveBayes/evaluation_reducer.py,NaiveBayes/Smoothed/NBmodel.txt \
  -mapper classify_mapper.py \
  -reducer evaluation_reducer.py \
  -cmdenv PATH=/opt/anaconda/bin:$PATH \
  -input {HDFS_DIR}/enron_test.txt \
  -output {HDFS_DIR}/evaluation-smooth \
  -numReduceTasks 1

# save the model locally
!hdfs dfs -cat {HDFS_DIR}/evaluation-smooth/part-000* > NaiveBayes/Smoothed/results.txt

# retrieve results locally
#!cat NaiveBayes/Smoothed/results.txt

# part c - display results (RUN THIS CELL AS IS*)
# NOTE: *feel free to modify the tail commands to match the format of your results file
print '=========== UNSMOOTHED MODEL ============'
!tail -n 9 NaiveBayes/Unsmoothed/results.txt
print '=========== SMOOTHED MODEL ============'
!tail -n 9 NaiveBayes/Smoothed/results.txt<jupyter_output>=========== UNSMOOTHED MODEL ============
0018.2001-07-13.sa_and_hp	1	0.0000000000	0.0000000000	False
0018.2003-12-18.gp	1	0.0000000000	0.0000000000	False
# Documents: 	20
True Positives: 	0
True Negatives: 	9
False Positives: 	0
False Negatives: 	11
Accuracy: 	0.45
Recall: 	0.00
=========== SMOOTHED MODEL ============
# Documents: 	20
True Positives: 	6
True Negatives: 	7
False Positives: 	2
False Negatives: 	5
Accuracy: 	0.6500
Precision: 	0.7500
Recall: 	0.5455
F1-Score: 	0.6316<jupyter_text>__`EXPECTED RESULTS:`__ Unsmoothed ModelSmoothed Model Documents:	20True Positives:	0True Negatives:	9False Positives:	0False Negatives:	11Accuracy	0.45Recall	0.0 Documents:	20True Positives:	6True Negatives:	7False Positives:	2False Negatives:	5Accuracy	0.65Precision	0.75Recall	0.5454F-Score	0.6315__`NOTE:`__ _Don't be too disappointed if these seem low to you. We've trained and tested on a very very small corpus... bigger datasets coming soon!___`Q10 part d starts here:`__<jupyter_code># part d - write your Hadoop job here (sort smoothed model on P(word|class))
 
# copy model file    
!hdfs dfs -copyFromLocal NaiveBayes/Smoothed/NBmodel.txt {HDFS_DIR}

# clear output directory
!hdfs dfs -rm -r {HDFS_DIR}/top-words

# hadoop job
!hadoop jar {JAR_FILE} \
  -files NaiveBayes/model_sort_mapper.py \
  -D stream.num.map.output.key.fields=4 \
  -D mapreduce.job.output.key.comparator.class=org.apache.hadoop.mapred.lib.KeyFieldBasedComparator \
  -D mapreduce.partition.keycomparator.options="-k4,4nr" \
  -D mapreduce.partition.keypartitioner.options="-k3,3" \
  -partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner \
  -mapper model_sort_mapper.py \
  -reducer /bin/cat \
  -input {HDFS_DIR}/NBmodel.txt \
  -output {HDFS_DIR}/top-words \
  -numReduceTasks 2


# part d - print top words in each class
for idx in range(2):
    print "\n===== part-0000%s=====\n" % (idx)
    !hdfs dfs -cat {HDFS_DIR}/top-words/part-0000{idx} | grep -v '^ClassPriors' | head<jupyter_output>===== part-00000=====

ect	378.0,0.0,0.023473306082,0.000055460041	ham	0.023473306082	
and	258.0,277.0,0.016041124737,0.015417891409	ham	0.016041124737	
hou	203.0,0.0,0.012634708287,0.000055460041	ham	0.012634708287	
in	160.0,157.0,0.009971509972,0.008762686484	ham	0.009971509972	
for	148.0,153.0,0.009228291837,0.008540846320	ham	0.009228291837	
on	122.0,95.0,0.007617985879,0.005324163940	ham	0.007617985879	
enron	116.0,0.0,0.007246376812,0.000055460041	ham	0.007246376812	
i	113.0,106.0,0.007060572278,0.005934224391	ham	0.007060572278	
will	113.0,69.0,0.007060572278,0.003882202873	ham	0.007060572278	
this	99.0,90.0,0.006193484454,0.005046863735	ham	0.006193484454	
grep: write error
cat: Unable to write to output stream.

===== part-00001=====

the	453.0,535.0,0.028118419423,0.029726581998	spam	0.029726581998	
to	350.0,420.0,0.021739130435,0.023348677278	spam	0.023348677278	
a	168.0,274.0,0.010466988728,0.015251511286	spam	0.015251511286	
your	35.0,271.0,0.002229654404,0.015085131163	s[...]<jupyter_text>Bonus Question: Sorting with multiple Reducers (Optional)Congratulations you've completed all the required questions! If you have energy for more, let's take a moment to preview one of the important design issues that we'll tackle next week: total order sort.In the jobs in this assignment we've mostly used a single reducer and mostly used the default Hadoop sorting (alphabetical by key). However when we took the time to specify an alternate key to sort on and tried using multiple reducers (as we did in Questions 5d, 6e and 10d)... we saw that sorting and partitioning the data on different fields can cause our final output to no longer be sorted from top to bottom. Instead it will usually be sorted just within each partition. This is called a partial sort and in the examples we saw that was mostly what we wanted.So far, we've also always partitioned on a categorical field. But what if we wanted to sort on values (eg. counts in a word count file) but have too much data to fit on a single reducer and there isn't a logical categorical variable to use as a partition key? How could we ensure the the top-n words end up in `part-00000`, the next n end up in `part-00001`, etc? The solution is similar in idea to what we did in Question 10b (where we created extra fields to partition on)... but a bit more complicated because the way that Hadoop orders its partitions and writes them to file depends on an internal hash function (eg. we might expect the partition with key `ham` to end up in `part-00000` and `spam` to end up in `part-00001` but Hadoop won't necessarily order the words the way a human reader would).__Challenge Exercise:__ Write a Hadoop job that will use 5 reducers and will sort your `NBmodel.txt` file (you can use the smoothed or unsmoothed one) in descending order according to the words' conditional probability in `spam`. A few tips:* start by reading the [Total Sort Notebook](https://github.com/UCB-w261/main/blob/master/Resources/TotalSortGuide/_total-sort-guide-spark2.01-JAN27-2017.ipynb) that is provided in the course repo under the HelpfulResources folder.* you will need to write a mapper for this job, you can do this easily on the fly using the jupyter magic command `%%writefile`.* when you run your job, you should see that the records in `part-00000` have higher conditional probabilities in Spam, than the records in `part-00001` which in turn are higher than in `part-00002` etc.  Bonus Question: Sorting with multiple Reducers (Optional)Congratulations you've completed all the required questions! If you have energy for more, let's take a moment to preview one of the important design issues that we'll tackle next week: total order sort.In the jobs in this assignment we've mostly used a single reducer and mostly used the default Hadoop sorting (alphabetical by key). However when we took the time to specify an alternate key to sort on and tried using multiple reducers (as we did in Questions 5d, 6e and 10d)... we saw that sorting and partitioning the data on different fields can cause our final output to no longer be sorted from top to bottom. Instead it will usually be sorted just within each partition. This is called a partial sort and in the examples we saw that was mostly what we wanted.So far, we've also always partitioned on a categorical field. But what if we wanted to sort on values (eg. counts in a word count file) but have too much data to fit on a single reducer and there isn't a logical categorical variable to use as a partition key? How could we ensure the the top-n words end up in `part-00000`, the next n end up in `part-00001`, etc? The solution is similar in idea to what we did in Question 10b (where we created extra fields to partition on)... but a bit more complicated because the way that Hadoop orders its partitions and writes them to file depends on an internal hash function (eg. we might expect the partition with key `ham` to end up in `part-00000` and `spam` to end up in `part-00001` but Hadoop won't necessarily order the words the way a human reader would).__Challenge Exercise:__ Write a Hadoop job that will use 5 reducers and will sort your `NBmodel.txt` file (you can use the smoothed or unsmoothed one) in descending order according to the words' conditional probability in `spam`. A few tips:* start by reading the [Total Sort Notebook](https://github.com/UCB-w261/main/blob/master/Resources/TotalSortGuide/_total-sort-guide-spark2.01-JAN27-2017.ipynb) that is provided in the course repo under the HelpfulResources folder.* you will need to write a mapper for this job, you can do this easily on the fly using the jupyter magic command `%%writefile`.* when you run your job, you should see that the records in `part-00000` have higher conditional probabilities in Spam, than the records in `part-00001` which in turn are higher than in `part-00002` etc.<jupyter_code>%%writefile demo/bonus_mapper.py
#!/usr/bin/env python
"""
Mapper reads model and emits words their conditonal probability (for spam) and assigns them a class/partition depending on value

INPUT:
    word \t hamCount,spamCount,pHam,pSpam

OUTPUT:
    pSpam \t word \t partition
"""

max = 0.0029726581998 # Max probability; got this by iterating through once

import sys
for line in sys.stdin:
    word, payload = line.split()
    spam_cProb = payload.strip().split(',')[3]
    
    # Ideally we want to split the partition by distribution but we won't know that at first read so have to manually adjust
    #if max <= float(spam_cProb):
    #    max = float(spam_cProb)
    
    if float(spam_cProb) >= 0.1*max:
        print "%s\t%s\t%s" % (spam_cProb, word , 'A')
    elif float(spam_cProb) >= 0.08*max and float(spam_cProb) < 0.1*max:
        print "%s\t%s\t%s" % (spam_cProb, word , 'B')
    elif float(spam_cProb) >= 0.05*max and float(spam_cProb) < 0.08*max:            
        print "%s\t%s\t%s" % (spam_cProb, word , 'C')
    elif float(spam_cProb) >= 0.02*max and float(spam_cProb) < 0.05*max:  
        print "%s\t%s\t%s" % (spam_cProb, word , 'D')
    else:             
        print "%s\t%s\t%s" % (spam_cProb, word , 'E')

# Used once        
#print max
                                                                                                                               
#!hdfs dfs -cat {HDFS_DIR}/NBmodel.txt | grep -v '^ClassPriors' | demo/bonus_mapper.py
# BONUS - write your hadoop job here!

# clear output directory
!hdfs dfs -rm -r {HDFS_DIR}/bonus-sort

# hadoop job
!hadoop jar {JAR_FILE} \
  -files demo/bonus_mapper.py \
  -D stream.num.map.output.key.fields=3 \
  -D mapreduce.job.output.key.comparator.class=org.apache.hadoop.mapred.lib.KeyFieldBasedComparator \
  -D mapreduce.partition.keypartitioner.options="-k3,3" \
  -partitioner org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner \
  -mapper bonus_mapper.py \
  -reducer /bin/cat \
  -input {HDFS_DIR}/NBmodel.txt \
  -output {HDFS_DIR}/bonus-sort \
  -numReduceTasks 5
!hdfs dfs -ls {HDFS_DIR}/bonus-sort<jupyter_output>Found 6 items
-rw-r--r--   1 root supergroup          0 2018-01-25 12:28 /user/root/HW2/bonus-sort/_SUCCESS
-rw-r--r--   1 root supergroup      11330 2018-01-25 12:28 /user/root/HW2/bonus-sort/part-00000
-rw-r--r--   1 root supergroup       3230 2018-01-25 12:28 /user/root/HW2/bonus-sort/part-00001
-rw-r--r--   1 root supergroup      19523 2018-01-25 12:28 /user/root/HW2/bonus-sort/part-00002
-rw-r--r--   1 root supergroup      44127 2018-01-25 12:28 /user/root/HW2/bonus-sort/part-00003
-rw-r--r--   1 root supergroup      38564 2018-01-25 12:28 /user/root/HW2/bonus-sort/part-00004<jupyter_text>NoteNot the best distribution but that is because words are distributed in a power law manner (Zipf's Law).<jupyter_code># BONUS - show us it worked here!
for idx in range(5):
    print "\n===== part-0000%s=====\n" % (idx)
    !hdfs dfs -cat {HDFS_DIR}/bonus-sort/part-0000{idx} | head<jupyter_output>===== part-00000=====

0.000332760246	above	A	
0.000332760246	ads	A	
0.000332760246	allow	A	
0.000332760246	answer	A	
0.000332760246	answers	A	
0.000332760246	avoid	A	
0.000332760246	away	A	
0.000332760246	bd	A	
0.000332760246	body	A	
0.000332760246	bonus	A	

===== part-00001=====

0.000277300205	across	B	
0.000277300205	activate	B	
0.000277300205	advertisement	B	
0.000277300205	again	B	
0.000277300205	age	B	
0.000277300205	akkabay	B	
0.000277300205	announcement	B	
0.000277300205	application	B	
0.000277300205	approved	B	
0.000277300205	assistance	B	

===== part-00002=====

0.000166380123	accepting	C	
0.000166380123	added	C	
0.000166380123	adding	C	
0.000166380123	addresses	C	
0.000166380123	adult	C	
0.000166380123	advertise	C	
0.000166380123	agreement	C	
0.000166380123	alhaji	C	
0.000166380123	almost	C	
0.000166380123	along	C	

===== part-00003=====

0.000110920082	abn	D	
0.000110920082	absenteeism	D	
0.000110920082	absolute	D	
0.000110920082	absolutely	D	
0.000110920082	absorb	D	
0.0[...]
{'fraction_non_alphanumeric': 0.08381999928343664, 'fraction_numerical': 0.0607824871914299, 'mean_word_length': 4.255908106581301, 'pattern_counts': {'":': 0, '<': 55, '<?xml version=': 0, '>': 102, 'https://': 4, 'lorem ipsum': 0, 'www.': 6, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 38, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '16', 'dir': 'jupyter-structured-clean-dedup', 'id': '658847', 'license': 'MIT', 'n_tokens_mistral': 19995, 'n_tokens_neox': 17376, 'n_words': 7491, 'path': 'HW02/hw2_part2_Workbook.ipynb', 'repo_name': 'maynard242/Machine-Learning-At-Scale', 'size': '55822'}