Due Dates

• Code: 11:59PM ET, April 29, 2022

---

---

---

Part 1: Introduction to MapReduce

30 points

In this part of the assignment you will solve two simple problems by making use of the PySpark library.

For each problem, you will turn in a python script (stencil provided) similar to wordcount.py that solves the problem using the supplied MapReduce framework, PySpark.

Problem 1: Inverted Index

Fill in the code for inverted_index.py, which creates an inverted index of a given set of documents. Given a set of documents, an inverted index is a matching from each word to a list of document IDs of documents in which that word appears.

Your task is to design a MapReduce pipeline that would generate inverted indices for words in the given documents. You will have to think about how your data will move between the various stages of the pipeline and implement the following accordingly:

1. def mapper1(record)
   
2. def reducer1(a, b)
   
3. def mapper2(record)
   
4. def reducer2(a, b)
   

As a note, you can feel free to use more mapper/reducer functions than those stated above, but you shouldn't need to - our solution manages to do it using only those 4 functions. In general, you have total control over the number of mapper and reducer functions that you use.

Your final task is to create such an inverted index matching with a MapReduce pipeline, using the mapper and reducer functions you just implemented. This query should return the inverted index of the given documents. You should use the variable inverted_index_result to store the result of your query.

For this problem, use the books.json and books_small.json datasets as the input to your pipeline. To run the file, execute the following command:

$ python3 inverted_index.py -d PATH/TO/data.json 

where PATH/TO/data.json is the path to the json file with the data (so either ending in books_small.json or books.json). By default, without the -d flag, the data file path is ../data/books_small.json.

Successfully running the script will create a file named output_inverted_index.json in a directory called output, which will contain the data that was collected by the pipeline in inverted_index_result. The format of the answer should look something like this:

[
    [
        "Answer",
        [
            "shakespeare-caesar.txt"
        ]
    ],
    ...
]

We provide you with a script to help you check the format of your json files:

$ ./check_format /PATH/TO/output_inverted_index.json

You can also verify the output of your pipeline on books_small.json using the provided check_outputs_equal script, passing it the path to your generated file and the ta solution's generated file, which can be found at ../data/ta_output/output_inverted_index.json (or at /course/cs1951a/pub/mapreduce/data/ta_output/output_inverted_index.json):

./check_outputs_equal output/output_inverted_index.json ../data/ta_output/output_inverted_index.json

Problem 2: Relational Join

Fill in the code for join.py, where your task is to implement a SQL join query using a MapReduce pipeline. You will work with the data provided in records.json which contains tuples belonging to both 'Release' and 'Disposal' tables.

Consider the following SQL query:

SELECT *
FROM Release, Disposal
WHERE Release.CompanyID = Disposal.CompanyID 

Your MapReduce query should produce the same output as the SQL query above, with each row formatted as a list. You can consider the two input tables, 'rele' and 'disp', as one big concatenated bag of records which gets fed into the map function record by record. For each line/record in records.json, record[0] is the name of the table (either "rele" or "disp") and record[2] is CompanyID. Feel free to open the json file to see examples.

You will have to implement the following funtions:

1. def mapper1()
   
2. def reducer()
   
3. def mapper2()
   
4. def filterer()
   

Hint: Consider what key might be necessary to organize all the given data. Once you have all the data for a key in one place, then you can figure out how to generate the output you want.

You should use the variable join_result to store the result of your pipeline. Like above, you have total control over the number and order of functions in your pipeline; the above is just the order that our solution uses.

For this problem, use records.json as the input to your pipeline. Similar to Part 1, to run the file, activate the virtual environment and then execute the following command:

$ python3 join.py -d PATH/TO/records.json 

This will create a file named output_join.json in your output directory, which will contain the data that was collected by the pipeline in join_result. It will look like:

[
    [
        "rele",
        "1995",
        "4836",
        ...,
        "disp",
        "2003",
        "4836",
        ...
    ],
    ...
]

You can also use this script to check the format.

$ ./check_format /PATH/TO/output_join.json

---

Part 2 - Movie Recommendation System

40 points

In Edwin Chen's blog article on movie similarities, he describes how he used the Scalding MapReduce framework to find similarities between movies. You will do the same by calculating the similarity of pairs of movies so that if someone watched Frozen (2013), you can recommend other movies they might like, such as Monsters University (2013).

Data

You are provided with a dataset of movie ratings:

• Source: MovieTweetings by Simon Dooms.
• Overview: Ratings are extracted from tweets and it contains up-to-date movie ratings (the earliest rating contained in this dataset is from Feb 28, 2013). It contains 906,727 ratings from 37,338 movies. And we will only be using a fraction of this.
• Run Time: Our implementation of similarities.py finishes executing in ~30 seconds on the small dataset and in ~45 seconds on the big one.

This dataset includes:

• ratings.dat
• Format: user_id::movie_id::rating
• Ratings are from 1 to 10 (whole-number ratings only)
• 34,123 ratings (small)
• 190,750 ratings (big)
• movies.dat
• Format: movie_id::movie_title
• Titles are identical to titles provided by IMDB (including year of release)

Algorithm

As mentioned in Edwin Chen's blog article, we will use the different metrics between movie pairs to determine the similarity between them:

• For every pair of movies A and B, find all the people who rated both A and B.
• Use these ratings to form a Movie A vector and a Movie B vector.
• Calculate the similarity metrics between these two vectors.
• Whenever someone watches a movie, then you can recommend the most similar movies.

If you aren't familiar with this definition of a vector, all we mean by it is a list of numbers that we can imagine as n coordinates of an n-dimensional space.

Like what Edwin did in his article, you will also experiment with four similarity metrics. The implementations for these metrics are provided in similarities.py. Click on the link to find out more about the usage of these similarity metrics in your program.

Implementation

In similarities.py, you will implement a series of mappers and reducers. You will pass two input files, ratings.dat and movies.dat, to similarities.py, which will then output a list of movie pairs along with their similarity metrics between them like below (pretty-print JSON):

[
    [
        [
            "movie_title1",
            "movie_title2"
        ],
        [
            correlation_value,
            regularized_correlation_value,
            cosine_similarity_value,
            jaccard_similarity_value,
            n,
            n1,
            n2
        ]
    ],
    ...
]

For every pair of movies A and B, find all the people who rated both A and B and compute the number of raters for every movie. Then you can calculate 4 similarity metrics for every movie pair.

Below are the mappers and reducers that you will implement. We have provided the first part of the pipeline for you in the stencil code. For the remaining part, your MapReduce pipeline can have as many mappers and reducers as long as your outputs match the the two checkpoints and the final requirement.

Checkpoint 1

1. def mapper0()

2. def reducer()

   (Here you will be taking the parameters a and b and joining them. Don't overthink this! Refer to the lab if you get stuck.)

3. def mapper1()

   The output of your pipeline at this stage (after mapper1) should be of the following format:

   [[key, value], [key, value], ...]
     where -
       key:   movie_title
       value: [ [user1_ID, user1_rating], [user2_ID, user2_rating], [user3_ID, user3_rating], ...]

   The output at this stage should be stored in the variable stage1_result and will be written to the file netflix_stage1_output.json. This will serve as a checkpoint into your pipeline for the purposes of grading, so please make sure you implement this correctly. Its format will look like:

   [
       [
           "$5 a Day (2008)",
           [
               [
                   "22136",
                   7
               ],
               ...
           ]
       ],
       ...
   ]
   

   Note that the json file is very compact. If you want to pretty print it like above, you can use the following command. Don't worry about the order. It is because the collect() action is parallelized, and then the results are assembled. We will sort your results when grading.

   $ cat PATH/TO/netflix_stage1_output.json  | python -m json.tool
   

   You can use our script to check the format:

   $ ./check_format /PATH/TO/netflix_stage1_output.json
   

   You can also use our script to check the contents of your output on the small dataset:

   $ ./check_outputs_equal PATH/TO/YOUR/netflix_stage1_output.json PATH/TO/TA/netflix_stage1_output.json
   

   Checkpoint 2

   Next, you are free to design your own MapReduce pipeline. Just don't forget to satisfy the requirement of the second checkpoint before the final output.

4. def mapper2()

5. def mapper3()

6. ...

   You are provided with implementations of 4 similarity metrics. You should refer back to the Algorithm section above or the beginning of similarities.py to determine the input values for each of these metric functions. You will need to find the dot product between two vectors, the sum of each vector, the norm of each vector, and etc. In addition, you should ignore (do not include values for) movie pairs whose regularized correlation values are less than some threshold (i.e. 0.5) in order to keep only high value movie pairs.

   The output of your pipeline at the second checkpoint should be of the following format:

   [[key, value], [key, value], ...]
     where -
       key: movie_title1
       value: [[movie_title2, correlation_value, regularized_correlation_value, cosine_similarity_value, jaccard_similarity_value, n, n1, n2], [movie_title3, ...]]

   IMPORTANT: Only include movie_title2s for a movie_title1 when movie_title1 < movie_title2 (i.e. movie_title1 comes alphabetically before movie_title2). The output at this stage should be stored in the variable stage2_result and will be written to the file netflix_stage2_output.json. This will serve as the second checkpoint into your pipeline for the purposes of grading, so please make sure you implement this correctly. Its format will look like:

   [
       [
           "12 Years a Slave (2013)",
           [
               [
                   "Jagten (2012)",
                   0.6671378907298551,
                   0.5221079144842344,
                   0.9937391441268904,
                   0.04265402843601896,
                   36,
                   617,
                   263
               ],
               ...
           ]
       ],
       ...
   ]
   

   You can use our script to check the format:

   $ ./check_format /PATH/TO/netflix_stage2_output.json
   

   You can also use our script to check the contents of your output on the small dataset:

   $ ./check_outputs_equal PATH/TO/YOUR/netflix_stage2_output.json PATH/TO/TA/netflix_stage2_output.json
   

Final Output

7. ... # any mappers/reducers

   The output of the last stage should have the following format.

   [[key, value], [key, value], ...]
     where -
       key: [movie_title1, movie_title2]
       value: [correlation_value, regularized_correlation_value, cosine_similarity_value, jaccard_similarity_value, n, n1, n2]

   The output of the last stage should be stored in the variable final_result, which will be written to the file netflix_final_output.json. Then the format of the output file will look like:

   [
       [
           [
               "Captain America: The First Avenger (2011)",
               "Iron Man 3 (2013)"
           ],
           [
               0.7280290128482472,
               0.5824232102785978,
               0.9886495309825268,
               0.018682858477347034,
               40,
               82,
               2099
           ]
       ],
       [
           [
               "Captain America: The First Avenger (2011)",
               "The Avengers (2012)"
           ],
           [
               0.8188595535772019,
               0.603370197372675,
               0.990419871882812,
               0.0979020979020979,
               28,
               82,
               232
           ]
       ],
       ...
   ]
   

   You can use our script to check the format:

   $ ./check_format /PATH/TO/netflix_final_output.json
   

   You can also use our script to check the contents of your output on the small dataset:

   $ ./check_outputs_equal PATH/TO/YOUR/netflix_final_output.json PATH/TO/TA/netflix_final_output.json
   

   We have provided a skeleton MapReduce query based on the TA solution; however, you are free to choose the internal implementation of your query. Just please ensure that you adhere to the format of the data that you store in these three files: netflix_stage1_output.json, netflix_stage2_output.json and netflix_final_output.json.

How to Run

To test your program, first activate the virtual environment, and then enter:

$ python3 similarities.py -d PATH/TO/data

where PATH/TO/data is a path to the folder containing movies.dat and ratings.dat.

The default data path is ../data/recommendations/small/. It will generate three json files in the folder output in your working directory.

Grading Criteria

To get full credit for this assignment, your pipeline must in general be "MapReducy" - that is, you need to apply some of the common MapReduce techniques discussed in class. One very important technique is using yield when writing flatMap (this allows for data to be pushed down the pipeline without manually creating list and waiting for the function to finish).

Another (even more important) technique is using reduce functions to guarantee uniqueness instead of using sets whenever possible (since sets cannot be parallelized the same way that reducers can be). Otherwise, your computation could be skewed in favor of certain keys that appear more frequently than others, which could drastically slow down your pipeline (consider language, where common words such as "the" appear far more frequently than others).

In general, so long as your pipeline adheres to these rules, you will recieve full credit. Also, its important to note that the majority of your grade will come from simply having a pipeline that produces a correct output, even if it is inefficient.

---

Written Questions

Create a writeup.md and write your answers to the following questions in it.

15 points

1. (10 points) Consider that you are given a dataset containing the details of babies born in the US in 2018. Each record is of the form recordID :: year :: month :: state :: city and there are around 3,978,497 (4 million) records.
   In order to find the number of babies born during each month of the year, you come up with the following mapper and reducer (Refer to wordcount.py):
   

   Mapper:  record -> (record.month, 1) 
   
       For each record map the month to count 1.
   
   Reducer: k,[v]  -> k, sum(list[v])
   
       For each key sum all values associated.

   

   The MapReduce cluster provided to you consists of N mappers and but only 2 reducers as shown in the figure above. Reducer1 receives all (key, value) pairs where keys are between A and M inclusive and Reducer2 receives (key, value) pairs between N and Z inclusive.

   Given that mapper and reducer function produces the correct output, what possible issue(s) could you face while processing a job consisting of about 4 million records? Suggest a workaround for that issue (without changing the number of mappers or reducers).

   Hint: If you're stuck, think about how the load will be distributed across the pipeline!




2. (5 points) You are given the following MapReduce pipeline which finds the 10 most frequent words beginning with each letter, in a large English text corpus.

   def mapper1(sentence):
       for word in sentence.split(' '):
           yield (word.lower(), 1)
                       
   def mapper2(pair):
       word, count = pair[0], pair[1]
       return (word[0], [(word, count)])
                       
   def mapper3(letter_pair):
       letter, word_pairs = letter_pair[0], letter_pair[1]
       for word_pair in word_pairs[:10]:
           yield (letter, word_pair[0])
                       
   output = sentences.flatMap(mapper1).reduceByKey(add).map(mapper2).reduceByKey(add).flatMap(mapper3)

   
   

   After testing on a small text file, it was noted that the pipeline does not produce correct output. Explain why this pipeline does not produce the correct output.




---




Handing In




After finishing the assignment, run python3 zip_assignment.py in the command line from your assignment directory, and fix any issues brought up by the script.

After the script has been run successfully, you should find the file mapreduce-submission-1951A.zip in your assignment directory. Please submit this zip file on Gradescope under the respective assignment.




---




▶ Appendix A: Similarity Metrics (Click to expand) 😃

In this part, we provide more details about how the similarity metrics are being used in the assignment.

In the below equations, n is the number of users who rated both movie Xand movie Y, n1 is the number of users who rated movie X, and n2 is the number of users who rated movie Y.




1. Correlation
   

   $Correlation(X, Y) = \frac{n \sum xy - \sum x \sum y}{\sqrt{n \sum x^2 - (\sum x)^2} \sqrt{n \sum y^2 - (\sum y)^2}}$

2. Regularized Correlation
   

   $Weight(X, Y) = \frac{n}{n + VirtualCount}$

   $RegularizedCorrelation(X, Y) = Weight(X, Y) * Correlation(X, Y) + (1 - Weight(X, Y)) * PriorCorrelation$

   As Edwin states, "we can also also add a regularized correlation, by (say) adding N virtual movie pairs that have zero correlation. This helps avoid noise if some movie pairs have very few raters in common (for example, The Great Gatsby had an unlikely raw correlation of 1 with many other books, due simply to the fact that those book pairs had very few ratings)."

   The stencil code uses VIRTUAL_COUNT = 10 and PRIOR_CORRELATION = 0.0, and you are welcome to experiment with different values (but don't forget to change them back before you submit!)

3. Cosine Similarity

   $Cosine(X, Y) = \frac{\sum xy}{\sqrt{\sum x^2} \sqrt{\sum y^2}}$

4. Jaccard Similarity

   $Jaccard(X, Y) = \frac{n}{n_1 + n_2 - n}$

   As Edwin states, "recall that one of the lessons of the Netflix prize was that implicit data can be quite useful - the mere fact that you rate a James Bond movie, even if you rate it quite horribly, suggests that you'd probably be interested in similar action films. So we can also ignore the value itself of each rating and use a set-based similarity measure like Jaccard similarity."




---




Credits




Made by Shunjia Zhu, Solomon Zitter, and Nam Do in Spring 2020 with past contribtions from Neel Virdy, Colby Tresness, Haomo Ni, and Ashish Rawat. Updated in Spring 2021 by Suhye Park, Daniel Civita Ramirez, Matteo Lunghi, Mary Dong, and Nam Do, and again in Summer 2021 by Julia Windham, Evan Dong, and Nam Do.

Part 1 adapted from a previous assignment which was developed by Karthik, Harihar Reddy Battula, Ishan Bansal, Samuel Crisanto, Yufeng Zhou, Lezhi Qu, and John Ribbans with suggestions from Tim Kraska and Alex Galakatos. Movie recommendation problem is based on the Edwin Chen's blog article.