Parallelism Models, MapReduce, and Iterative Computation

3.1 Top-Down Design

3.1.1 The Divide-and-Conquer Idea

Top-down design is the practice of taking a larger problem, dividing it into smaller problems, solving those sub-problems, and then combining the results of the sub-problems to arrive at the solution of the larger problem. The technique should feel familiar: the divide-and-conquer strategy taught in a data structures course works the same way — divide the problem, solve the sub-problems, combine the results. What changes as we move toward distributed programming is not the idea but the constraints under which the division happens: how many processors exist, how the work is shared among them, and how the final combination is done.

Think of cooking a dinner with friends. The whole job — a three-course meal — is too much for one cook. So you split the menu into dishes, one person takes the soup, another the main course, a third the dessert, and in the end everyone brings their dish to the same table. The dinner (the larger problem) is solved by combining the dishes (the sub-problem solutions). The mapping to parallel programming is direct: the meal is the data or the task, the cooks are the processors, and "the table" is the combine step. The analogy breaks in one important place: in the kitchen the dishes are usually independent of each other, while in parallel programming the sub-problems often need one another's results — and that dependency is exactly what decides whether a division is even possible (we come back to this in the quick sort discussion in Section 3.5).

In a sequential context — a single processor — the division is easy to execute because the sub-problems run one by one. How you divide the problem is left entirely to the programmer. The classic picture comes from C programming, which most of us have touched at some point: the execution starts in a main function, and the program calls a separate function for each functionality whenever it needs that functionality. The call sequence is strictly linear — say F1 is called first, then F3, then F4; inside F1 there may be a call to F2, and F2 may call F5. At any point in time exactly one function is executing, and at the end the solution of the problem simply emerges from that single chain of execution.

3.1.2 Top-Down Design in a Parallel Context

In a parallel context the division is no longer left to the programmer's taste; it is dictated by the number of processors. The objectives change too. In a sequential setting with one processor, a typical goal is modularity — writing one function per functionality so the code stays organized — plus keeping the sub-problems manageable: you do not want so many sub-problems that you lose track of the computation progress of each one. In the parallel context the main objective is different: processor utilization — how busy the processors are — and a faster overall runtime. If there are \(n\) processors, we can potentially have \(n\) sub-problems, and that gives efficient use of the resources.

The two failure modes sit on either side of that sweet spot:

• Fewer sub-problems than processors: with \(n\) processors and fewer than \(n\) sub-problems, every processor that gets a sub-problem takes one, but some processors stay idle. The resources are under-utilized.
• More sub-problems than processors: equal to \(n\) sub-problems can run at once, one per processor, and the rest wait for processor time. That is acceptable — the processors are fully used — but it costs scheduling and waiting time.

There is a second set of questions the programmer must answer when dividing a problem for parallel execution. First, how many processors do we have, and can we assign each sub-problem to its own processor to get an efficient runtime? Second, is the problem itself divisible — can the solutions of the sub-problems actually be combined at the end? Combining is the real constraint: each sub-problem now runs on a different processor, so we must decide how to combine. Is the combination trivial, just collecting the outputs from each processor and appending them into the final result? Or does it need extra logic? And if a small combination logic is needed, can that logic itself run in parallel? The combination should stay straightforward — not very, very complex — because a heavy combine step eats the time the parallelism saved.

The same reasoning applies level by level in a decomposition tree: a problem \(P\) divides into \(P_1\) and \(P_2\); at the next level \(P_1\) divides further, and so does \(P_2\). Whenever the sub-problems at a level can run in parallel and there is no interdependency among them, the parallelism is clean and the execution time stays low. Interdependency is the enemy: if a sub-problem's result is needed before a sibling can start, the sibling just waits.

3.1.3 Student Questions and Answers

3.2 Running Time Complexity and the Parallel Keyword Search

3.2.1 Asymptotic Complexity Recap

The running time complexity of an algorithm is a function we associate with the algorithm with respect to its input size — how many numbers are in the list. The standard notation is big-O, the asymptotic notation. Two classic search algorithms from the data structures course set the pattern:

• Linear search compares the target with the first element, then the second, then the third, and so on until the element is found, or until the list ends. In the worst case you make \(n\) comparisons, so the running time complexity is \(O(n)\) — it varies linearly with the input size. Double the list, double the worst-case work.
• Binary search requires a sorted list. You compare the target with the middle element directly. If it matches, you are done. If the target is less than the middle element, you search in the left half; if it is greater, you search in the right half. Then you compute the middle of that half and repeat. Every step halves the search space: initially \(n\), then \(n/2\), then \(n/4\), then \(n/8\), and so on. The complexity function is \(\log_2 n\), written \(O(\log_2 n)\). The one hard requirement: the list must be sorted.

So in the sequential context, searching a sorted list \(LS\) of size \(N\) for a key \(K\) costs \(O(N)\) with linear search and \(O(\log_2 N)\) with binary search.

3.2.2 Worked Example: Searching a Nine-Element List

3.2.3 Parallel Binary Search over \(P\) Processors

Now suppose the sorted list \(LS\) of size \(N\) is distributed over a cluster of \(P\) processors. Divide the data equally: each processor gets \(N/P\) elements. If the list has a thousand numbers and there are ten processors, each processor gets \(N/P = 100\) numbers.

Each processor now runs a binary search on its own chunk in parallel. The search space per processor is no longer \(N\) but \(N/P\), so the running time complexity becomes:

\[ O\!\left(\log \frac{N}{P}\right) \]

Since all the processors search at the same time, the wall-clock time is that of a single processor doing \(\log(N/P)\) work — the parallelism adds no extra time. The ideal speed-up over the sequential version is \(p\) times:

\[ \text{speed-up} = \frac{\log_2 N}{\log_2 (N/P)} \approx P \]

This is theoretical — there is some time spent collecting and combining the results from the different processors, which we are not counting here — but as a first model, the speed-up is \(P\). (A note on the speed-up rule: the speed-up is always \(p\) times for \(p\) processors; when the example happens to use the three-processor dependency diagram from Section 3.1, the same rule gives three times. The "three" is the number of processors in that picture, not a different formula.)

With four processors the chunk size is \(N/4\), and the search completes in \(\log(N/4)\) steps — each processor halves its own quarter, not the whole list.

3.2.4 Worked Example: A Thousand Numbers on Ten Processors

3.2.5 Student Questions and Answers

3.3 More Parallel Search Examples

3.3.1 Fingerprint Matching

Fingerprint matching is a search problem with a different flavor. We have a data set \(D\) of fingerprints — say \(n\) fingerprints total — and we want to search for a particular fingerprint \(f\) in that set. To parallelize, we partition \(D\) based on the number of processors, and the partitions are evenly stored in a distributed database: the first processor stores one chunk, the second processor stores another, and so on.

The match itself is a linear search — comparing fingerprint images is not a sorted-list operation, so there is no binary search here. With \(n\) fingerprints and \(p\) processors, each processor gets \(n/p\) fingerprints, and the complexity drops from \(O(n)\) to:

\[ O\!\left(\frac{n}{p}\right) \]

All processors run their linear search in parallel, so the result is again a speed-up of \(p\) equal to the number of processes in hand.

3.3.2 Worked Example: When Do We Repartition?

3.3.3 Document Search in a Distributed Collection

The third example is search again, presented differently. We have a distributed document collection \(D\), a set of documents \(d_1, d_2, \ldots, d_n\), and a set of keywords to search in each document. We want to find which documents contain those keywords.

The parallel recipe is the same: divide the documents among the number of processors you have, then run the search in parallel instead of scanning the documents one by one. What each search returns — the location where the keyword is present, or simply a flag that the document matches — is an application requirement: the search logic returns whatever the application needs. The results are collected, and the answer is: these are the documents that contain these keywords.

Real-world: This is the kind of workload MongoDB was built for. MongoDB is a document-oriented database — we saw it in the previous discussion. If you host MongoDB on a cluster of nodes and divide your documents across the cluster, then a query — say a search for keywords inside documents — can execute in parallel on the different nodes of the cluster, and it runs much faster than the same query on a single node. The same pattern appears in biometric identity systems (fingerprint matching over a partitioned, distributed fingerprint database — Section 3.3.1) and in full-text search across a document warehouse. In every case the economics are identical: build the partition once, search it many times, and let the parallelism divide the search — not the search logic divide the data set per query.

3.4 The Data Parallel Execution Model

3.4.1 Same Task on Different Data (SIMD)

Everything we did in the three search examples was an instance of the data parallel execution model. The defining move: the data is partitioned across multiple nodes — not the tasks. The task stays identical on every node, and each node works on its own slice. In the keyword search, every processor ran the same search routine on its hundred numbers. In fingerprint matching, every node performed the same match task, but the data set within was different: processor one had \(d_1\), processor two had \(d_2\), processor three had \(d_3\), and so on up to \(d_p\). One task \(T\) runs against data \(d_1, d_2, \ldots, d_p\).

This maps directly onto one of the four architecture classes from Flynn's taxonomy: SIMD — single instruction, multiple data. There is a single instruction — searching, for instance — that runs across the processors, but the data is multiple: each processor has a different set of data.

3.4.2 Homogeneous and Heterogeneous Nodes

The goal of the partition logic is to make the partitions equal or balanced, because balanced partitions give the fastest results. Whether balance is easy depends on the hardware:

• Homogeneous nodes — nodes with the same level of hardware configuration. Here we try to balance the partitions, and each node performs an equal amount of work. With balanced partitions on homogeneous nodes, the processors stay 100 percent busy and the execution time is minimal.
• Heterogeneous nodes — some nodes have high-end processors while others are low-end. The partition logic becomes complex, because we want to balance the load: the nodes with low-end processors should get less work, and the nodes with high-end processors should get more.

Two failure modes follow. If the nodes are homogeneous but the partitioning logic is unable to balance the partitions — say the data sizes end up unequal — the execution time rises. If the partitions are balanced but the nodes are heterogeneous, the slowest node — the one with the low-end configuration, weaker memory, and so on — becomes the bottleneck: we cannot present the result until all the nodes have finished, because the final result is collected from every processor. So balance is needed on both sides: equal work if the nodes are equal, proportionate work if they are not.

3.4.3 Balancing Partitions and the Slowest-Node Bottleneck

3.4.4 When Data Parallelism Is Not Possible

There are two scenarios where data parallelism cannot be used effectively:

1. You cannot divide the work equally. If the data cannot be split into roughly equal chunks, the data parallel model has nothing to balance.
2. You cannot divide the work independently. If the sub-problems are not independent — if one sub-problem needs the result of another before it can proceed — they cannot run in parallel, and data parallelism is out of the question.

Quick sort is the classic example that trips both conditions, which is why it gets its own treatment.

3.5 Quick Sort: When Data Parallelism Fails, Tree Parallelism Steps In

3.5.1 How Quick Sort Partitions

Quick sort is a comparison-based sorting technique from the data structures course. You are given an unsorted list; quick sort returns a sorted list. At each step it chooses one pivot element — say the last element of the current segment — and it tries to fix the position of that pivot within the list. When the pivot's position is fixed, it is fixed in a specific way: all elements to the left of the pivot are less than the pivot, and all elements to the right are greater than the pivot.

Notice what that buys and what it does not. The pivot's own position is now final — it will never move again. But the left part is not necessarily sorted yet, and neither is the right part; we have only fixed one element. The algorithm then recurses: choose a pivot in each part, fix it, and continue until each part is too small to divide.

3.5.2 Why the Pivot Breaks Balanced Partitioning

The problem for parallelism is that the partition size depends on the pivot. Suppose the pivot you choose is the second smallest element of the list: its fixed position is near the front, so on the left of it you have just one element, and on the right you have \(n-2\) elements. The list has not been cut into anything close to equal halves — it has been cut into \(1\) and \(n-2\).

That violates the requirement of the data parallel execution model, which wants a balanced partition at each step: you should be able to divide the data in a balanced way. If the pivot divides the list so that one side has many more elements than the other, the partition is unequal, and you cannot achieve the efficiency data parallelism promises. A 45/55 split is still a good partition — 55 percent of the elements on one side is fine. A 10/90 split is bad, and a 1/99 split is worse. In the pathological worst case, if every pivot lands so that one element sits on one side and \(n-1\) on the other, the running time is as bad as the sequential version — the same one-and-\(n-1\) pattern repeats at every level. The probability of that happening is very low, but it can happen in the worst case, so at maximum the time taken may be as bad as sequential with bad partitioning.

There is a second, subtler reason quick sort is not a one-shot data parallel problem: the division is stepwise, not a single level. In the search examples we took a thousand elements, divided into ten chunks of a hundred once, and that division was done. In quick sort, we first find the position of the pivot in the full list and split it into two parts; then each part is handed to a processor, which picks its own pivot and splits again; and so on. The partition at every level depends on the partition at the parent level — it is not a single-level problem, and the choice of pivot position \(m\) cannot guarantee equal partitions. One set of processors can end up with a large partition while another set gets a small one.

3.5.3 Tree Parallelism: Parallel Partitioning at Each Level

So quick sort is not suited to data level parallelism — but it is suited to a third kind of parallelism, the tree parallel execution model. The insight: even though the partition sizes are unpredictable, the partitioning operations at the same level are independent of each other and can run in parallel.

The picture is a tree. Level zero: the full list is partitioned into two parts. Level one: the two parts are partitioned in parallel — each of the two processors picks a pivot and splits its own part — producing four parts. Level two: those four are partitioned in parallel, and so on. The parallelism we can achieve is "dividing the list into two at each step; this division can be done in parallel; and finally you combine the result from each of the nodes to get your final result." Each level of partitioning happens in parallel, and the result is a sorted list when you append the outputs of all the processes.

The stopping rule for the recursion comes back to the number of processors: at each step you ask whether to keep dividing. If the problem is divided into \(k\) sub-problems at a level and the processors can execute them in parallel, that level takes roughly one step, and that is the speed-up you achieve. Three cases:

• Sub-problems equal to processors: every processor gets exactly one sub-problem — every processor fully busy, best outcome.
• Fewer sub-problems than processors: some processors sit idle.
• More sub-problems than processors: all processors are busy, and the rest wait in line — fine, but they add wait time.

So the number of processors is the key factor in deciding whether to keep dividing: stop when the sub-problems match the processors you have.

3.5.4 Student Questions and Answers

3.6 Task Level Parallelism

3.6.1 Different Tasks on the Same Data (MISD)

Task level parallelism is the third kind. In data level parallelism the data gets divided, and every processor runs the same task. In tree level parallelism the data gets divided too, but stepwise and level by level. Task level parallelism flips the relationship: the data does not get divided. Instead, the tasks get divided into sub-tasks, and the sub-tasks may work on the same data instance — possibly on the same data copy, or on different copies that must be kept in sync.

The tasks are different: one processor computes one thing, another computes something else, and they run in parallel on the same data. This maps onto another class of Flynn's taxonomy: MISD — multiple instruction, single data stream. There are multiple instructions, and a single data stream is shared.

3.6.2 Examples: A Spell Checker and Mean-Median-Mode

Two examples make it concrete.

The word processor on a multi-core machine. You are typing something in a word processor on a multi-core machine. While you type, your spell checker runs in the background on a different core, highlighting the incorrect words wherever you missed a spelling. The typing/editing task and the spell-checking task are different tasks running on the same document — the same data — at the same time, on different cores.

Mean, median, and mode. You are given a list of integers and you want to compute three independent statistics: the mean, the median, and the mode. Here the tasks differ for each statistic — the instructions are not the same — but the data is the same: the same list. Give each processor the same data and a different task, and run them in parallel: one processor finds the mean, another finds the median, a third finds the mode. Multiple instructions, one data stream — a textbook MISD pattern — and the three results come back at the same time, giving high performance.

3.6.3 Characteristics of Task Parallel Subtasks

Three important points to note about task level parallelism:

1. Subtasks are identified by functionality, with no common function. In tree and data parallelism the tasks are identical functions; in task parallelism you identify the sub-tasks by what they compute, and the sub-tasks share nothing.
2. Independent sub-tasks can be executed in parallel. The parallelism only exists if the sub-tasks do not depend on one another — which is why mean, median, and mode work so cleanly.
3. Sub-tasks are often limited and known statically in advance. They do not change frequently. In the word processor example we know the sub-tasks up front: editing the document, running the spell checker, statistical analysis. In the statistics example we know we want mean, median, and mode. These are known before the run starts.

3.6.4 Scalability: Data Parallelism vs Task Parallelism

Scalability — what happens when you add more resources — separates the three models sharply. Take data parallelism with a thousand numbers divided into ten sets of a hundred, run on ten processors. Now suppose the data grows to 5,000 numbers. If we increase the resources to 50 processors, each node again gets a hundred numbers to search, and the results come back in the same time as before. We added 40 nodes, and the per-node workload did not change: the system is scalable by adding more resources — economies of scale. This works because data level parallelism divides the data; the workload splits evenly no matter how big it gets.

Task level parallelism does not scale this way. If you simply add more nodes, you cannot split the tasks further — one task may differ from another, and one task may take much longer than the others. The heavy-task node becomes the bottleneck: the other nodes finish quickly, but the result cannot be presented to the end user until the heavy-task node completes its work. So task parallelism is not easily scalable by adding more resources. When it comes to data level parallelism — whether the tree form or one-step division — it makes sense to add resources and get scalability; for task parallelism, adding resources does not buy the same benefit.

Real-world: The word-processor example is how a modern laptop works: the main thread renders and edits, background threads run spell-check, auto-save, and telemetry — all over the same document buffer on different cores. The statistics example is the shape of embarrassingly parallel analytics dashboards where several independent aggregates (counts, sums, percentiles) are computed over one data frame at the same time. In the distributed world, task parallelism appears wherever a small, fixed set of heterogeneous jobs runs over a shared data set — and the lesson of this section is that such systems scale by making the data bigger or the task set larger, not by adding machines to the same fixed task set.

3.7 Request Level Parallelism

3.7.1 The Client-Server Use Case

Request level parallelism is the typical pattern in a client-server architecture. Multiple clients are requesting something from a server, and the server responds by executing tasks in parallel — some tasks happen for one request, some for another, possibly several tasks at the server end at the same time — and after parallelizing all of it, the server returns the results. The concurrent requests are typically executed as different threads.

The difference from the previous models is where the parallelism lives. Data, tree, and task parallelism are all one program, one problem, many workers. Request level parallelism is many problems, many users, one serving program: the parallelism comes from the requests themselves, not from splitting any single request. Each user's request is already a separate unit of work; the server's job is to run as many of them concurrently as it can.

Real-world: The typical examples are an email server, which serves many clients at once, any sort of API interface where multiple requests come in and must be handled at the same time, a reservation system, or an online banking application. All of these handle request level parallelism on the server side.

3.7.2 The Throughput Metric

The scalability metric for request level parallelism is how many requests per unit of time the server can serve — that is, the throughput the system can generate. That is the number you have to achieve when you configure a server to handle many concurrent clients.

3.7.3 Student Questions and Answers

3.8 Setting the Stage: The Distributed Context for MapReduce

3.8.1 Loosely Coupled Systems and Message Passing

Distributed systems are typically loosely coupled: you do not have a shared memory. When you divide the data, the memory and storage of each node are separate — memory lives on separate nodes, and the data is segregated. Any exchange of data between nodes happens through a message passing mechanism.

This is different from a tightly coupled system with shared memory. With shared memory, combining results is simpler: each process can write into a memory location, and other processors can read the value from that memory location. In a distributed system, we need to collect the data from the different nodes explicitly.

3.8.2 Locality of Reference: Move the Task to the Data

The important design rule when dividing a problem in a distributed system: divide the problem in a way that the computation task can run on the local data. This is the concept of locality of reference (LOR) from earlier: we try to move the task to the data, because moving the data around is a quite costly thing. Each node's memory is its own, exchanges go through message passing, so the computation should be scheduled where its data already lives.

Think of a library. To answer a question, you do not haul every book into a central reading room and then search them — you walk to the shelf where the relevant book lives and read it there. Moving the book is expensive and risks damage; walking (moving the reader) is cheap. In a cluster, the "books" are data chunks on disks, and the "readers" are computation tasks: schedule the task on the node that already holds the chunk, and the network stays quiet. This is why Hadoop later assigns a map task to the very node that stores its input split — the data never moves; only the code moves.

3.8.3 The Conquer Step: Combining Results

The conquer step states that we have to combine the results from the different processes or nodes. In some cases the combining may be as simple as just collecting the results and giving them to the end user — as in the search examples, where merging means collecting the results at low cost. In other cases it may involve logic. Keeping the combination simple makes the processing more and more efficient.

In the quick sort example, if you ultimately append the output of all your processes, you get your sorted list — the combine step is nearly free. But sometimes the merge is sequential, and that hurts. K-means clustering is the example the discussion sets up for this: the reduce step in k-means is not as simple as in quick sort or the parallel linear search — it is not straightforward, and if the combine is going to be sequential, it hampers the performance of the distributed system.

3.8.4 K-Means: The Clustering Problem

K-means clustering is a data mining problem — clustering is a data mining problem. The setting: you have data about your customers, and you want to group them into different clusters — say three or four groups — based on the demographics of their purchasing behavior. You have no historical data and no label data; you only have demographic data, and you want to partition the customers into four groups. That is the clustering problem, and k-means is the algorithm used to cluster such data.

The steps that appear in the material: guess clusters in parallel to improve the clusters, but checking whether we have found the right clusters is sequential. What this ultimately says is that the reduce step of k-means is not as simple as the combine steps of the search examples or quick sort: the reduce step is going to be a sequential one, and a sequential combine at the end hampers the performance of the distributed system.

Real-world: Customer segmentation is the canonical deployment — an online retailer groups customers by purchasing demographics and serves each group a different campaign. The same algorithm clusters sensor readings, image pixels, and search result sets. The domain lesson for this lecture: whenever a clustering step must run on distributed data, its sequential combine makes it the weakest link — which is exactly the problem the iterative MapReduce frameworks of Section 3.11 were built to fix.

3.9 The MapReduce Programming Model

3.9.1 Map as Data Parallelism, Reduce as Inverse Tree Parallelism

We touched the word count problem before; now we define MapReduce properly, in terms of data parallelism and tree parallelism.

In word count, we take a sentence, break it into multiple partitions, and give each partition to one processor. Each processor returns key-value pairs — we will study exactly what those are. Then a reduce task counts the number of values corresponding to each key, giving the occurrence of each word in the text.

Seen through the parallelism lens:

• The map step is a kind of data parallelism: you divide your data into multiple sets and run the same logic on each set. You are dividing the problem into sub-problems based on the data.
• The reduce step is a kind of inverse tree parallelism. Look at the combine picture: the sub-problems keep being combined — eight results combine into four, four into two, two into one — until we arrive at the final result. That is the conquer step of the merge sort diagram: dividing until a base case, then combining upward, with both the dividing and the combining happening in parallel.

So map is parallel, reduce is parallel, and depending on the problem the reduce step may be as simple as collecting the outputs, or may need logic to combine the results of different reduce tasks.

3.9.2 Word Count: Map, Shuffle and Sort, Reduce

The word count example in detail. We want to count the occurrences of each word in a sentence. The map job picks up each word in the sentence and emits a key-value pair: the word followed by the value 1. It is not counting how many times the word appears — each occurrence emits its own \((word, 1)\). Tokenizing gives a stream of \((word, 1)\) pairs.

Then comes the step that was asked about earlier: where does the sorting and shuffling happen? Between the map job and the reduce job, the MapReduce runtime performs a shuffle and sort: the pairs are sorted by key, and all the values belonging to the same unique key are grouped together. In the example, the word that starts with "b" comes first, then the "c" words, then "da" — the pairs are sorted. The values for a unique key are united: if the word "blue" got two separate 1s from two map emissions, those two values are grouped into one list. This is a hidden step performed by the MapReduce runtime — we do not write code for it. The shuffling also moves the data around among the different nodes, and the framework takes care of that too.

Finally, the reduce job sums the values for each key and emits another key-value pair: the key (the word) together with its frequency of occurrence. That is the final count.

3.9.3 Worked Example: One Sentence, Word by Word

3.9.4 The Hadoop Word Count Diagram and Reducer Assignment

The Hadoop-flavored version of the same problem works over a bigger input. The input — "welcome to hadoop class hadoop is good hadoop is and bad" — is split into four pieces, and this is where the "welcome to Hadoop" style of the classic example comes from.

• Input splitting: the input is split into four parts, each handled by its own map.
• Mapping: every map function does the same job — tokenize and return key-value pairs: (welcome,1), (to,1), (hadoop,1), (class,1), (hadoop,1), (is,1), (good,1), (hadoop,1), (is,1), (and,1), (bad,1). Every map does the same job on its own slice — this is the data parallelism connection.
• Sorting: the hidden step — the pairs get sorted by key.
• Shuffling: the crisscross arrows in the diagram show the data moving around among the different processors, distributing the sorted pairs to the different reducers.
• Reducing: each reducer takes the values for its keys and produces the final counts — "hadoop" comes out with a count of 3.
• Combining: the results of the reducers are combined into one and you get the final output file.

3.9.5 A Third View: Key-Value Pairs and GROUP BY

There is a third way to look at the MapReduce steps, adapted from Jeff Ullman's course slides — a good reference to look at. The input to map is a set of key-value pairs: say a document ID as the key and the document content as the value. Four documents: doc 1 = "welcome to hadoop", doc 2 = "class hadoop", doc 3 = "is good", doc 4 = "hadoop is".

• The map step generates intermediate key-value pairs: one pair per word occurrence.
• These intermediate pairs go through shuffle and sort and are then handed to the reduce step.
• The reduce step is grouped by unique key — the same GROUP BY you know from SQL, where you group data based on a particular attribute. Here we group by key: for a key with four or five entries, we pick up those values and make a list.
• The reduce step is the aggregation function: sum the values, produce the reducer output, combine the outputs, and present the final result to the end user.

The three views answer the recurring question — where shuffle and sort happen: the shuffle and sort step sits between map and reduce, and it is taken care of by the MapReduce framework itself.

3.9.6 Formal Definition: From Input to Output

Formally, MapReduce works like this:

1. Partitions. You partition your data into a bunch of partitions.
2. Record readers. Each partition is read by a record reader — every programming language has something to read data: scanner classes in Java and similar facilities elsewhere.
3. Map function. The map function generates key-value pairs, termed intermediate key-value pairs.
4. Shuffle and sort. Some sorting happens — the shuffle and sort acts on these key-value pairs.
5. Reduce function. The sorted, shuffled data is fed to the reducer, which generates another key-value pair for each key.
6. Combine. You combine the output of the different reducers — that is your final output.

3.9.7 When to Use MapReduce Instead of SQL

When do we choose MapReduce over a traditional relational approach? Suppose you have a huge set of documents that do not fit into memory: you need file-based processing in stages. If you used SQL — the GROUP BY and COUNT functions — the RDBMS would compute the count by unique keys fine, but only for a certain range of data. An RDBMS is where you store data in relational tables, and that fits only a limited range.

MapReduce is suited to the problems of huge data sets — billions of records that cannot fit into the main memory — because it needs a lot of data partitioning and high data level parallelism, and then it simply merges the partitions to get the final result. So the decision rule is practical: if you have a small set of records that fits in an RDBMS, use GROUP BY and be done. When you have billions of records that do not fit into main memory, partition them, store them across multiple nodes, retrieve them, run a map-reduce model, and retrieve the final result.

3.9.8 Worked Example: Transactions by Country

3.9.9 Student Questions and Answers

3.10 Running MapReduce on a Cluster

3.10.1 Master, Slaves, and Workers

MapReduce runs on a cluster with a master node and slave nodes kind of configuration. A concept introduced here is the worker: workers are assigned to perform the map and reduce jobs. Given a bunch of documents where we want to count the words, the workers take these documents; which worker executes on which processor is taken care of by the system itself. The framework creates some workers to do the map jobs and some to do the reduce jobs.

Think of a restaurant kitchen. The master is the head chef who plans the service; the workers are the line cooks; the slave nodes are the stations where the ingredients sit. The head chef does not tell each cook which ingredient to buy — the ingredients are already at their stations — he only decides who cooks what. In the same way the master schedules map and reduce work onto workers, and the data does not travel to the code; the code travels to the data (Section 3.8.2).

3.10.2 Intermediate Files and the Shuffle and Sort Step

The execution flow is two-stage. Each map worker performs a map job on its share of the input and writes its output to an intermediate file — you can see this in the diagrams as files drawn in double lines. The output is saved to a file on disk. Then another worker on the reducer side reads this input; in between, the data is shuffled and sorted — that is why there are two steps: first the output is written, then it is shuffled and sorted, and then it is given to the reduce workers, who come up with the final output.

If there are multiple reducers, they write to multiple files. You can combine the results at the end and present a unified output.

This is a data-centric design: move computation closer to the data. The intermediate results are stored on the disk — saved into files — because the volume is too large for memory.

3.10.3 What the Framework Handles

The MapReduce library does all the work of allocating resources, starting workers, managing them, moving data, and handling failures — if one worker fails, shifting its task to the other workers is also taken care of by the system itself. What is important for the programmer: you only need to write the code for a map task and a reduce task; the rest is taken care of by the system. If you know how to divide the problem effectively — the divide-and-conquer habit — and how to write the combine logic, that is good enough; the system handles distribution, job tracking, and fault tolerance.

The programming model is a restricted programming interface to the system: you map your data this way, you reduce your data this way, and the efficient execution is taken care of by the runtime itself. You do not bother about resource allocation or scheduling.

3.10.4 History: GFS, Hadoop, and Functional Programming

A bit of history puts the model in context. Google created a distributed file system known as the Google File System (GFS), and then started running analytics on top of that file system. The open-source version of that stack was created as Apache Hadoop. Hadoop performs map and reduce on data using many machines; the system takes care of distributing the data and managing fault tolerance — if one node fails, how to divide its task is handled by the system.

The intellectual lineage matters: this is an older idea from functional programming, transferred onto large-scale distributed computing. In functional programming you write each function yourself; here we have found that mapping something and reducing something are the two pieces of logic a programmer must write — the "which part goes to which processor", "how it sorts", "how resources are allocated" parts are handled by the framework. Divide-and-conquer thinking plus a map and a reduce function is all the programmer contributes.

Real-world: GFS and Hadoop are the founding pair of the modern data stack — GFS showed how to store petabytes on commodity disks with replication, and Hadoop's MapReduce showed how to analyze that data with plain map/reduce code. The same two-stage philosophy lives on in its successors: Hive translates SQL to MapReduce-style jobs, Pig translates scripts, and Spark (Section 3.11) keeps the map/reduce model but moves the intermediate data from disk to memory. Whenever a company says "we ran a Hadoop job," what ran is exactly this: master, workers, intermediate files, shuffle, and reduce — with the framework silently doing everything except the two functions the programmer wrote.

3.11 Iterative MapReduce

3.11.1 One-Pass Computation vs Iterative Computation

All the MapReduce examples so far are one-pass computations: the static data is fed to the map jobs, they run, the output is saved somewhere; the runtime shuffles and sorts it and gives it to the reducer; the reducer does some logic and places its output into a file. One pass — you load your data once.

But many applications — especially in machine learning and data mining — need to process the data iteratively, because training models is an iterative process. Those applications need iterative execution of map-reduce jobs: once the result is produced by the reduce, it should be fed back into the system again. The diagram shows what happens: we feed the variable data in, call map and reduce, and when this output is achieved, we call them again, time and again.

One way to do this is to write a driver program — a main program — that runs a loop: take the output, feed it back to the map as the new variable data, call map and reduce again. But the iteration has a special structure: there is variable data, which changes each iteration, and static data, which remains the same. Every time a map job runs, it needs the static data — it must load that static data again. So a naive loop over map and reduce carries real costs:

1. Loading the static data every iteration — it has to be loaded from files again and again.
2. Combining the results of the different reducers every iteration, then feeding the combination back to the map.

You run these iterations until you converge, and the convergence criteria is defined based on the application. Two classic examples: k-means clustering and PageRank.

3.11.2 Worked Example: K-Means Clustering

K-means clustering, worked through by hand. We are given a bunch of data points — plotted on a two-dimensional plane as gray points. The objective: find the groups among these data points based on the distance matrix — group the points that are close together.

Setup. First we decide how many groups we need; in this example, three groups. Three centroid points are chosen randomly — these are the cluster centers, marked as circles.

Iteration 1 — assign. Compute the distance of each gray data point from each of the three centroids. Whichever centroid a point is closest to, that point joins that centroid's group. For example: take a point and measure its distance to centroid one, centroid two, and centroid three; it is closer to centroid one, so it forms a group with centroid one. Another point is closer to the green centroid, so it goes to the green cluster. Doing this for all the points gives the clusters after the first iteration: one cluster ends up with just two points; another cluster ends up with six points plus its centroid — seven total, counting the centroid itself.

Iteration 2 — recompute centroids. We do not stop here — the clusters are not done. We compute a new centroid for each cluster: take the data points of the cluster and compute the mean of each value — the mean of the x-coordinates and the mean of the y-coordinates. The new centroid of each cluster moves: the red centroid shifts a little; the green centroid moves downward because the density of points there is higher; the blue centroid stays more or less in place, shifting slightly, because it is computed as the mean of all its points.

Iteration 3 — reassign and check for change. Compute distances again and see if anything changed. In the example, with the new centroids, two points are now closer to the red centroid than to the green one, so they move to the red cluster. The green cluster is left with its remaining points.

Repeat until no change. Another iteration: take the mean of the points in each cluster, recompute the three centroids, compute distances again, reshuffle the points into clusters. Keep doing this until there is no change — until no points shift from one cluster to another. At that point, stop: the k-means clustering is done.

3.11.3 K-Means on MapReduce: Static and Variable Data

Now the same algorithm on a MapReduce cluster. Suppose we have 100 data points and three centroids, and ten processors. The data points are divided into sets of ten; each processor computes the distance of its ten points to the three centroids; the reduce step gives the result — which points belong to which cluster. That is one iteration.

But after the reduce, we want to compute the new centroids and feed that variable data back in. The static data — the ten points each map job runs on — does not change; those points are stable. What changes are the circle points, the centroids, because we compute new centroids every time.

A simple driver loop that calls map and reduce again and again with the new variable data is not performance-optimized: the static data has to be loaded each and every time — loaded from files, stored back to files, read from files again. Each iteration goes: load the file, run the map job, shuffle and sort, run the reduce job, run the user program, and start again — with the static data re-loaded from files on every pass. Reading from and writing to a file is a costly thing, so the whole loop is dominated by I/O rather than computation.

3.11.4 PageRank: The Problem

PageRank is the algorithm behind ranking in the Google search engine, and it is another iterative problem — though the algorithm itself is a bit difficult, and this represents a simplified version. When you search, certain pages are retrieved, and the question is how to show them in a particular sequence — whether this website should come first or that website should come first. Each web page has a rank associated with it, and based on that rank the relevance of the page is decided and the pages are shown in order in the search results.

The major difficulty is calculating the rank of each web page. The rank of a page depends on a number of inbound and outbound links — the links coming into the page and going out of it — and inbound links are more important than outbound links.

The setup for one page: there is a page \(x\), and web pages \(t_1, t_2, \ldots, t_n\) around it. From each \(t_i\) there is a hyperlink pointing to \(x\) — all \(n\) pages link to \(x\) — and there are some other outbound links from the \(t_i\) pages as well. How does a user reach page \(x\)? Two ways: directly — you have a link and go straight to the page — or by browsing, coming from \(t_1\) or \(t_2\) or any other page that points to \(x\).

3.11.5 The PageRank Equation

Let \(p\) be the probability of a random jump, and \(N\) the total number of nodes in the graph. To calculate the rank of \(x\): there is a probability that the user jumps directly to a particular node, and a probability that the user visits it from other nodes. How much rank does a page \(t_i\) transfer to \(x\)? Page \(x\) has an inbound link from \(t_i\), so it takes some rank from \(t_i\). The rank it takes is the page rank of \(t_i\) divided by the number of outbound links of \(t_i\) — the rank is split evenly among everything \(t_i\) links to.

Look at \(t_1\): its outbound links are 1, 2, and 3, so it transfers \(\frac{1}{3}\) of its rank to \(x\), \(\frac{1}{3}\) to the second page, \(\frac{1}{3}\) to the third. So \(\frac{PR(t_1)}{3}\) contributes to \(x\). Look at \(t_2\): it has four outbound links, 1, 2, 3, and 4, so \(\frac{PR(t_2)}{4}\) contributes to \(x\). Similarly for all the other pages. In the example numbers, 0.5 is the probability that you arrive through the sequence of pages, and 0.5 is the probability that you directly access page \(x\), and that is how the page rank equation is written in general:

\[ PR(x) = (1-p) \cdot \frac{1}{N} + p \cdot \sum_{t_i \,:\, t_i \text{ links to } x} \frac{PR(t_i)}{C(t_i)} \]

where \(p\) is the probability of a random jump, \(N\) is the total number of nodes, and \(C(t_i)\) is the out-degree of \(t_i\) — the number of outbound links from \(t_i\) — since a page splits its rank equally among its outbound links.

3.11.6 Why PageRank Is Iterative — and the Convergence Rule

The page rank keeps on updating — the process is iterative. You start with an initial page rank \(PR_i\) for each node — assumed from some probability distribution — and then you compute the page ranks for every page in every iteration. Because the ranks feed each other, each iteration changes the values: if the rank of one page changes, the rank of the pages that receive its contribution changes too, since that page contributes to their rank. You keep iterating until you converge.

The convergence criteria is application-defined. Each iteration you compute the difference between the ranks of the previous iteration and the new iteration; if the difference is minute, you may stop; if it is a significant gap, you keep going — whether to stop depends on the threshold value of the difference that has been kept. If there is no change in the rank at all, that is also possible, and you stop there. (For a deeper understanding of how the values change iteration by iteration, working through one of the worked examples on Wikipedia is recommended.)

Structurally, PageRank is exactly like k-means from MapReduce's point of view: the variable data — the current rank vector — has to be fed in time and again, and a general map-reduce implementation would read the data from files again and again, which is not performance-optimized.

3.11.7 The Challenges — and the Frameworks That Solve Them

Running these problems with a general MapReduce implementation brings large overheads: the reinitialization of tasks, the reloading of static data, and communication and data transfers that happen multiple times. After the reduce, the output is saved onto multiple files; then this data has to be fed back to the map; the static data is loaded in every iteration. Each iteration carries the cost of reading from and writing to files.

There are ways to handle it. One is the caching technique; another is configuring map and reduce with the static data. Some other frameworks do iterative map-reduce jobs this way:

• MapReduce++ — an iterative MapReduce variant. A configure step is added, so loading the static data is optimized; some of the results are cached to enable faster performance; a combiner operation is introduced to collect all the reduced outputs, so you do not spend much time combining the results of different reducers; and caching and indexing techniques are used for faster retrieval of the data — it is not like reading from the files time and again. MapReduce++ provides configure-map and configure-reduce functions that optimize the loading of the static data, and an efficient combine operation that runs every time a reducer produces results.
• Twister — a framework used for iterative MapReduce.
• HaLoop — an extension of Hadoop that supports iterative jobs.
• Spark — not discussed in detail in this course (it comes later), but for reading: Spark uses in-memory computing to speed up the iterations; the data remains in memory, and Spark makes use of something called an RDD — resilient distributed dataset — to make the computation faster. A link to example code is provided for anyone who wants to go through it.

The performance comparison for running k-means clustering for multiple iterations — 16 iterations, average time taken — ranks the frameworks in order: the time is maximum for Hadoop, then DryadLINK, then Twister, and then MPI, which is the fastest. (The four bars on the comparison chart are Hadoop, DryadLINK, Twister, and MPI — the stated ordering is consistent: the first two take the longest and the last two are more optimized.) So if you want to go for iterative MapReduce, Twister and MPI are more optimized than Hadoop and DryadLINK. The material links further reading on HaLoop and the Twister architecture.

3.11.8 Student Questions and Answers

Exam Guidance Summary

No explicit mark distribution or question-pattern guidance was given in this session, but the following study signals were explicit and worth carrying into revision:

• Asymptotic notation is assumed knowledge. The complexity arguments — \(O(n)\) for linear search, \(O(\log_2 n)\) for binary search, and the \(O(\log(N/P))\), \(O(N/P)\) parallel forms with a theoretical speed-up of \(p\) — are the working language of the whole session. If the notation is not yet comfortable, read a page or two on asymptotic notations; no deep dive is needed, but the basics should be solid.
• Exam note: Be able to state why quick sort defeats data parallelism (pivot-dependent, unbalanced partitions; worst case as bad as sequential) and which parallelism model it uses instead (tree parallelism: partitioning at each level runs in parallel).
• Exam note: The three-fold classification — data, tree, and task level parallelism — plus request level parallelism, with the SIMD (single instruction, multiple data) and MISD (multiple instruction, single data) mappings from Flynn's taxonomy, is core material. Expect to compare them: same task on divided data vs different tasks on the same data vs parallel partitioning levels.
• Exam note: Know the MapReduce pipeline by heart: input splitting → record readers → map (emits intermediate key-value pairs) → shuffle and sort (runtime-provided hidden step) → reduce (group by key, aggregate) → combine. Know the rule that one unique key must go to exactly one reducer.
• Exam note: Know the static-data vs variable-data structure of iterative MapReduce, the two example algorithms (k-means clustering and PageRank), why a naive loop over map and reduce is not performance-optimized (reloading static data from files, re-combining reducer outputs, reinitializing tasks every iteration), and the frameworks that fix it (MapReduce++ with configure-map/configure-reduce and caching, Twister, HaLoop, Spark with RDDs).
• The level of depth requested for supporting algorithms: understand what k-means does (a quick look at k-means and k-means++ suffices — no deep dive), and work through one PageRank iteration example on Wikipedia for the iterative update mechanism.

Key Industry Applications

• Real-world: Fingerprint matching — biometric identity search over a partitioned, distributed fingerprint database; linear search per partition, with repartitioning only when the database grows substantially. The partition-once, search-often trade-off is the design rule: a few new fingerprints never justify redistribution; many do.
• Real-world: Distributed document search, and MongoDB as a document-oriented database hosted on a cluster: queries execute in parallel on the different nodes of the cluster and run much faster than on a single node. MongoDB's sharding is exactly the partitioning idea of Section 3.3 wearing a database name.
• Real-world: Word processing on multi-core machines — editing and background spell-checking running as different tasks on the same document (task level parallelism). The same shape appears in dashboards computing several independent aggregates over one data frame at the same time.
• Real-world: Email servers, API interfaces, reservation systems, and online banking — request level parallelism, with throughput (requests per unit time) as the scalability metric. Server capacity planning is a throughput question: concurrency divided by latency sets the request rate.
• Real-world: Google File System (GFS) and Apache Hadoop — the original open-source MapReduce stack; Hadoop performs map and reduce over many machines and handles distribution and fault tolerance itself, scheduling tasks onto the nodes that already hold the data (locality of reference).
• Real-world: Google's PageRank — web search result ordering driven by an iterative rank computation over inbound/outbound link structure; inbound links matter more than outbound. The rank vector is recomputed iteratively until the per-iteration differences fall below the threshold.
• Real-world: Iterative MapReduce frameworks in production and research: MapReduce++, Twister, HaLoop, DryadLINK, MPI, and Spark (in-memory computing over resilient distributed datasets, RDDs) — benchmarked on 16 iterations of k-means, with Hadoop slowest and MPI fastest among the compared frameworks. Spark's in-memory RDDs are the direct answer to the reload-static-data cost the naive loop pays.
• Real-world: SQL GROUP BY and COUNT remain the right tool for count-by-key queries on data sets that fit in an RDBMS; MapReduce is the choice at billions of records that cannot fit in main memory — the same aggregation, executed as partition → map → shuffle and sort → reduce → combine.
• Reference: The key-value view of MapReduce (document ID → content as input pairs, intermediate pairs, group-by-key, aggregation) is adapted from Jeff Ullman's course slides.