MapReduce takes a list of key-value pairs as input and produces a list of new key-value pairs as output. A key-value pair is a tuple where the first element is called the key and the second element is called the value. For example, ("name", "John") is a key-value pair where "name" is the key and "John" is the value. MapReduce takes a list of such key-value pairs as input and processes them to generate a list of new key-value pairs as output. As a side note, if you have many key-value pairs, you can think of them as a table with two columns, where the first column is the key and the second column is the value. A key can be used to look up the corresponding value in the table.

For example, suppose we have the following list of key-value pairs as input, where each key is a line number and each value is a line of text:

("1", "the quick brown fox")
("2", "jumps over the lazy dog")
("3", "the quick brown fox")
("4", "jumps over the lazy dog")
Or, in table form:

+-----+-------------------------+
| key | value                   |
+-----+-------------------------+
| 1   | the quick brown fox     |
| 2   | jumps over the lazy dog |
| 3   | the quick brown fox     |
| 4   | jumps over the lazy dog |
+-----+-------------------------+
Since MapReduce does not use this table form, we will not use it in the rest of this document.

Now, suppose we want to process the above list of key-value pairs to count the number of occurrences of each word. The output could look like the following, where each key is a word and each value is the number of occurrences of the word in the input:

("the", "4")
("quick", "2")
("brown", "2")
("fox", "2")
("jumps", "2")
("over", "2")
("lazy", "2")
("dog", "2")
(As a side note, using this output, we can use a word as a key to look up the corresponding value, i.e., the number of occurrences of the word.)

How do you perform such processing? There can be many algorithms, but one algorithm is to do it in three stages. In the first stage, we look at each input key-value pair, and for each word we encounter, we say there is one occurrence of the word. To keep track of this, we can use ("word", "1"), e.g., ("the", "1"), ("quick", "1"), etc., and save those in memory. This is just recording an encounter, not necessarily counting, i.e., if we encountered the same word again, e.g., "the", then we would put another ("the", "1") in memory.

Then in the second stage, we look at all those encounters and group them by each word. For example, if we encountered the word "the" twice, we would have ("the", "1") and ("the", "1") in memory. We then group them into a single entry, e.g., ("the", ["1", "1"]). This is still not counting but looking at what we have in memory and grouping the encounter recordings by each word.

Lastly, we perform the actual counting in the third stage. We look at all the groupings and add up. For example, if we encountered "the" three times and "quick" twice, the second stage would produce ("the", ["1", "1", "1"]) and ("quick", ["1", "1"]). We now look at each entry and add up the encounters to produce the actual count of each word, e.g., ("the", "3") and ("quick", "2").

Now, MapReduce allows us to perform such processing. It has three stages that generalize the stages described above. They are called the map stage, the group-by-key stage, and the reduce stage. The map stage runs a function called map for each and every input key-value pair. This map function typically generates new key-value pairs, and the second stage (the group-by-key stage) collects these new key-value pairs and group them by each unique key. The reduce stage runs a function called reduce for each and every key-value pair grouped by the group-by-key stage. Let's look at each stage in more detail.

The Map Stage

As mentioned, the map stage runs a function called map for each and every input key-value pair. The important thing is that MapReduce treats this map function as a black box. The actual implementation of map is customizable and can be written by anyone. The job of MapReduce is to run map correctly, i.e., repeatedly calling map for each and every input key-value pair. The output of map has to be a set of key-value pairs (no matter who writes it). For example, suppose you wrote a map function as follows (in pseudocode):

map(key, value):
  foreach word in value:
    write-output(word, "1")
This pseudocode implements the encounter counting algorithm described before. What it does is that, it takes a single (key, value) pair (as the input parameters (key, value) indicate), iterates over each word in the value (as foreach says), and produces a set of output key-value pairs in the form of (word, "1") (as write-output says). For example, suppose the input were ("1", "the quick brown fox"). The above map would produce ("the", "1"), ("quick", "1"), ("brown", "1"), and ("fox", "1") as the output. These are essentially recording the encounters of these words from the input.

Again, this map is customizable and MapReduce considers this a black box. The job of MapReduce is to repeatedly call this map function for each and every input key-value pair. For example, consider the following input.

("1", "the quick brown fox")
("2", "jumps over the lazy dog")
("3", "the quick brown fox")
Since there are three key-value pairs, MapReduce would call map three times, i.e., one call for each key-value pair. Those three map calls would produce the following output sets:

First set for the input ("1", "the quick brown fox"):

("the", "1")
("quick", "1")
("brown", "1")
("fox", "1")
Second set for ("2", "jumps over the lazy dog"):

("jumps", "1")
("over", "1")
("the", "1")
("lazy", "1")
("dog", "1")
Third set ("3", "the quick brown fox"):

("the", "1")
("quick", "1")
("brown", "1")
("fox", "1")
The outputs of map is called intermediate outputs as there are two more stages to go through in MapReduce.

The Group-By-Key Stage

The second stage, the group-by-key stage, collects all intermediate outputs generated by the map stage. This is a generalization of the grouping algorithm described in the overview. The job of the group-by-key stage is to group all outputs by their keys. For example, if you examine the previous example, you can see that across different sets of intermediate outputs, there are overlapping keys. For example, the key "the" appears across three sets and the key "quick" appears across two sets. Then, the group-by-key stage looks at all the keys, identifies unique keys, and creates a list of values that are associated with each unique key. For example, for the key "the", there are three values associated with it---"1" from the first set's first key-value pair, "1" from the second set's third key-value pair, and "1" from the third set's first key-value pair. (In our example, all values are identical ("1"), but generally speaking, values can be different.) Thus, the group-by-key stage looks at the unique key "the" and groups the values by the key as follows:

("the", ["1", "1", "1"])
Likewise, for the unique key "quick", the group-by-key stage sees that there are two values associated with it and groups them as follows:

("quick", ["1", "1"])
The group-by-key stage does it for all unique keys and passes those to the next stage, the reduce stage.

The Reduce Stage

The reduce stage runs a function called reduce for each key-value pairs grouped by the group-by-key stage. Similar to the map stage, the reduce stage considers the reduce function a black box, and anyone can implement it. The job of the reduce stage is to run it correctly, i.e., calling reduce for each and every key-value pair grouped by the group-by-key stage.

For each key-value pair grouped by the group-by-key stage, reduce takes the value, which is a list, and produces a single value. This is why it's called reduce (reducing a list to a single value). For example, suppose you wrote a reduce function as follows:

reduce(key, value):
  total = 0
  foreach one-occurrence in value:
    total = total + 1
  return total
This reduce function implements the last part of the encounter algorithm described before---it takes a list of encounter recordings and produces an actual count. For example, consider the following key-value pairs grouped by the group-by-key stage:

("the", ["1", "1", "1"])
("quick", ["1", "1"])
The reduce stage would call reduce twice. One call would produce ("the", "3") as the output, and the other call would produce ("quick", "2") as the output. As a whole, the final output of the reduce stage would be the following.

("the", "3")
("quick", "2")
Complete Outputs

Using the following input, MapReduce with the above map and reduce functions would produce the following outputs.

Input:

("1", "the quick brown fox")
("2", "jumps over the lazy dog")
("3", "the quick brown fox")
("4", "jumps over the lazy dog")
The complete list of what the map stage produces:

("the", "1")
("quick", "1")
("brown", "1")
("fox", "1")
("jumps", "1")
("over", "1")
("the", "1")
("lazy", "1")
("dog", "1")
("the", "1")
("quick", "1")
("brown", "1")
("fox", "1")
("jumps", "1")
("over", "1")
("the", "1")
("lazy", "1")
("dog", "1")
The complete list of what the group-by-key stage produces:

("the", ["1", "1", "1", "1"])
("quick", ["1", "1"])
("brown", ["1", "1"])
("fox", ["1", "1"])
("jumps", ["1", "1"])
("over", ["1", "1"])
("lazy", ["1", "1"])
("dog", ["1", "1"])
The complete list of what the reduce stage produces, which is the final output of MapReduce:

("the", "4")
("quick", "2")
("brown", "2")
("fox", "2")
("jumps", "2")
("over", "2")
("lazy", "2")
("dog", "2")
The above final output is the list of key-value pairs that show the number of occurrences of each word in the input.

As it turns out, map and reduce are quite powerful and can be used to solve a wide variety of problems. For example, map and reduce can be used to find a line in a file that contains a word, to count the access frequency of a web page in a web server log, to get the list of all URLs that contain a link to a specific web page, and so on.

Understanding Concurrency in MapReduce

The original MapReduce is designed to process large data sets where the map and reduce functions are executed concurrently on a cluster of machines. In this task, you will not use a cluster of machines, but you will use threads to execute map and reduce concurrently on a single machine.

The concurrent execution of map and reduce works as follows.

The MapReduce framework creates a set of threads to execute the map function and another set of threads to execute the reduce function.
The framework partitions the input key-value pairs into multiple chunks. The number of chunks is equal to the number of map threads. Let's consider the above example again to understand this better. If we have two map threads, the framework partitions the input key-value pairs into two chunks. The following will be the partitioning of the input key-value pairs:
Chunk 0: ("1", "the quick brown fox"), ("2", "jumps over the lazy dog")
Chunk 1: ("3", "the quick brown fox"), ("4", "jumps over the lazy dog")
The framework then assigns each chunk to a map thread.
Each map thread processes its assigned chunk by calling the map function repeatedly, once for each input key-value pair in the assigned chunk. Note that all the map threads run at the same time, processing their assigned chunks concurrently.
The framework then collects all the intermediate key-value pairs produced by all map threads.
The framework then again partitions the intermediate key-value pairs into multiple chunks and assigns each chunk to a reduce thread. Again using the above example, if we have four reduce threads, the framework partitions the intermediate key-value pairs into four chunks. The following will be the partitioning of the intermediate key-value pairs:
Chunk 0: ("the", ["1", "1", "1", "1"]), ("quick", ["1", "1"])
Chunk 1: ("brown", ["1", "1"]), ("fox", ["1", "1"])
Chunk 2: ("jumps", ["1", "1"]), ("over", ["1", "1"])
Chunk 3: ("lazy", ["1", "1"]), ("dog", ["1", "1"])
The framework then assigns each chunk to a reduce thread.
Each reduce thread processes its assigned chunk by calling the reduce function repeatedly, once for each intermediate key and its associated list of intermediate values in its assigned chunk. Note that all the reduce threads run at the same time, processing their assigned chunks concurrently.
The framework then collects all the final key-value pairs produced by all reduce threads and outputs them as the final output.
Requirements

There are several requirements for this task.

Your MapReduce framework is a library that implements necessary functions that a developer uses to process their data.
You should not change the provided code as we will replace it for our grading. This can easily result in a 0 for the task. Be very careful about this. You can add your own code/files as needed.
interface.h defines the functions that you need to implement.
mr_exec() executes the MapReduce framework. A developer calls this function to run map() and reduce(), and get the final output. You need to write this function to run the framework. A developer can call this function as many times as they want to process different input key-value pairs. The function takes the following arguments.
The input key-value pairs.
The map() and reduce() functions that the framework should execute. Note that a developer writes map() and reduce() functions. map() takes one key-value pair as input and generates zero or more new intermediate key-value pairs as output. reduce() takes one intermediate key and the list of intermediate values associated with the key as input and generates zero or more new final key-value pairs as output.
The number of map threads and the number of reduce threads. You need to create the specified number of threads to execute the map and reduce functions.
A buffer for storing the final output. The framework should store the final output in this buffer.
mr_emit_i() and mr_emit_f() handle intermediate and final key-value pairs. A developer calls (within map) mr_emit_i() to pass intermediate key-value pairs to the framework, and calls (within reduce) mr_emit_f() to pass final key-value pairs to the framework. You need to write these two functions to collect intermediate and final key-value pairs. A developer can call these functions as many times as they want within map() and reduce(), respectively.
The order of the input key-value pairs should be preserved for map(). For example, if the input key-value pairs have the following order: ("1", "the quick brown fox"), ("2", "jumps over the lazy dog"), ("3", "the quick brown fox"), and ("4", "jumps over the lazy dog"), the framework should call map() with the input key-value pairs in the same order. When dividing the input into chunks, the framework should also preserve the order of the input key-value pairs. For example, if we use the above input key-value pairs and two map threads, the framework should call map() with the following two chunks: ("1", "the quick brown fox"), ("2", "jumps over the lazy dog") and ("3", "the quick brown fox"), ("4", "jumps over the lazy dog"), one for each thread.
The order of the intermediate key-value pairs should be sorted by the key, according to the result of the strcmp() function. For example, if the intermediate key-value pairs have the following key-value pairs: ("brown", "1"), ("fox", "1"), ("dog", "1"), ("brown", "1"), and ("fox", "1"), the framework should sort the intermediate key-value pairs in the following order: ("brown", "1"), ("brown", "1"), ("dog", "1"), ("fox", "1"), and ("fox", "1"). The framework should then group them and divide them into chunks in the same way as the input key-value pairs.
Similarly, the order of the output key-value pairs should be sorted by the key, according to the result of the strcmp() function. For example, if the final output has the following key-value pairs: ("brown", "2"), ("fox", "2"), and ("dog", "2"), the framework should store the final output in the following order: ("brown", "2"), ("dog", "2"), and ("fox", "2").
When partitioning input or intermediate key-value pairs into chunks, the size of each chunk should be the size of (all key-value pairs) / (number of threads), except for the last chunk, which should contain the remaining key-value pairs if there's any. For example, if we have 7 key-value pairs and 3 threads, the first two chunks should contain 3 key-value pairs each, and the last chunk should contain 1 key-value pair. If we have 6 key-value pairs instead, each chunk should contain 2 key-value pairs.
You need to use pthread to create threads. You should also use appropriate synchronization primitives to ensure that the threads work correctly.
You need to write CMakeLists.txt that produces a single executable named mapreduce that runs the test cases in main.c.
You may find uthash useful.
Similar to previous tasks, you should not hard-code or customize your implementation tailored to our test cases. Generally, you should consider the provided test cases as examples or specific instances of general cases. For example, we can change the constants, the input key-value pairs, the map and reduce functions, the number of map and reduce threads, and so on. Your implementation should still work correctly.

Implement it in C