Due: 11:59PM, Mar. 2nd, 2021 ET

---

---

---

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:

$ python 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. 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 the json file, record[0] is the name of the table ({'rele', 'disp'}) and record[2] is CompanyID. You will have to implement the following funtions:

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

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:

$ python 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 571,073 ratings from 26,960 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 the 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.

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. 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."

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 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. def stageN()

   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. 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:

$ python 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.

---

Written Questions

Please write your answers to the following questions in written_questions.pdf (make sure that it is an actual pdf file and not other files renamed to be pdf. Us not being able to read your file result in a low grade for this component).

30 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 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 3,978,497 ~ 4 million records? Suggest a workaround for that issue.




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.

   Input: sentences

Mapper1 : sentence -> (word, count)
    for word in sentence:
        emit word, 1

Reducer1: list of (word, count) -> (letter, (word, count))
    total = 0
    for word, count in values:
        total += count
     emit word[0], (word, total)

Reducer2: list of (word, count) -> (letter, word)
    for word, count in values[:10]:
        emit letter, word
   
   

   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?

Beyond "Big Data"

MapReduce frameworks are used by data scientists trying to analyze large datasets. Large datasets contain many forms of bias that are often overlooked due to popular beliefs about the power of “Big Data.”

The goal of the following questions is to:
• Identify critical questions to ask and best practices to follow for data scientists working with large datasets
• Begin to demonstrate the relevance of social context and qualitative information to your final project



Your responses should be thoughtful, provide justification for your claims, and be concise but complete. See the response guide for more guidance.

Questions

3. (5 points) Read The Hidden Biases in Big Data. Based on the reading, propose two critical questions or best practices that data scientists should apply to all projects with large datasets. Your explanation of each critical question or best practice should include the point from the reading that inspired it. The goal of this exercise is to synthesize the reading in a way that’s actionable for data scientists and applicable to most projects.
4. (10 points) The reading emphasizes the importance of complimenting data sources with qualitative research. Start thinking about how you could compliment your final project with qualitative research, which you’ll do for the Socio-historical Context and Impact Report section of the Data Deliverable. For context, briefly describe your final project idea. Then, answer the following questions about your proposed final project. Each response should be about one paragraph. If you find information online, provide a link to cite your source.
   1. What socio-historical context, qualitative methods, or other information might help you better understand your data?
   2. How could the question or topic of your proposed project be studied without quantitative data? What are some advantages and limitations of the qualitative approaches you identify?
   
   

   ---

   
   

   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.

   
   

   ---

   
   

   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.

   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.