This paper explores the problem of \stragglers" in MapReduce: a common phenomenon where a small number of mappers or reducers takes signi cantly longer than the others to complete. The e ects of these stragglers include unnecessarily long wall-clock running times and sub-optimal cluster utilization. In many cases, this problem cannot simply be attributed to hardware idiosyncrasies, but is rather caused by the Zip an distribution of input or intermediate data. I present a simple theoretical model that shows how such distributions impose a fundamental limit on the amount of parallelism that can be extracted from a large class of algorithms where all occurrences of the same element must be processed together. A case study in parallel ad hoc query evaluation highlights the severity of the stragglers problem. Fortunately, a simple modi cation of the input data cuts end-to-end running time in half. This example illustrates some of the issues associated with designing e cient MapReduce algorithms for real-world datasets.
Categories and Subject Descriptors: H.3.3 [Information Storage and Retrieval]: Information Search and Retrieval
The only practical solution to large data problems
today is to distribute computations across multiple machines
in a cluster. With traditional parallel programming
models (e.g., MPI, pthreads), the developer shoulders the
burden of explicitly managing concurrency. As a result, a
signi cant amount of the developer's attention must be
devoted to managing system-level details (e.g.,
synchronization primitives, inter-process communication, data
transfer, etc.). MapReduce [
Taking inspiration from higher-order functions in functional programming, MapReduce provides an abstraction for Copyright c 2009 for the individual papers by the papers’ authors. Copying permitted for private and academic purposes. Re-publication of material from this volume requires permission by the copyright owners. This volume is published by its editors.
LSDS-IR Workshop. July 2009. Boston, USA. programmer-de ned \mappers" and \reducers". Key-value pairs form the processing primitives in MapReduce. The mapper is applied to every input key-value pair to generate an arbitrary number of intermediate key-value pairs. The reducer is applied to all values associated with the same intermediate key to generate output key-value pairs. This two-stage processing structure is illustrated in Figure 1.
Under the MapReduce framework, a programmer needs
only to provide implementations of the mapper and reducer.
On top of a distributed le system [
This paper explores a performance issue frequently encountered with MapReduce algorithms on natural language text and other large datasets: the \stragglers problem", where the distribution of running times for mappers and reducers is highly skewed. Often, a small number of mappers takes signi cantly longer to complete than the rest, thus blocking the progress of the entire job. Since in MapReduce the reducers cannot start until all the mappers have nished, a few stragglers can have a large impact on overall end-to-end running time. The same observation similarly applies to the reduce stage, where a small number of long-running reducers can signi cantly delay the completion of a MapReduce job. In addition to long running times, the stragglers phenomenon has implications for cluster utilization|while a submitted job waits for the last mapper or reducer to complete, most of the cluster is idle. Of course, interleaving the execution of multiple MapReduce jobs alleviates this problem, but presently, the scheduler in the open-source Hadoop implementation remains relatively rudimentary.
There are two main reasons for the stragglers problem.
The rst is idiosyncrasies of machines in a large cluster|
this problem is nicely handled by \speculative execution" [
map
Barrier: group values by keys reduce reduce
reduce output output output
rst instance to nish are used (the remaining results are discarded). This, however, does not address skewed running times caused by the distribution of input or intermediate data. To illustrate this, consider the simple example of word count using MapReduce (for example, as the rst step to computing collection frequency in a language-modeling framework for information retrieval): the mappers emit keyvalues pairs with the term as the key and the local count as the value. Reducers sum up the local counts to arrive at the global counts. Due to the distribution of terms in natural language, some reducers will have more work than others (in the limit, imagine counting stopwords)|this yields potentially large di erences in running times, independent of hardware idiosyncrasies.
This paper explores the stragglers problem in MapReduce, starting rst with a theoretical analysis in Section 2. A speci c case study in parallel ad hoc query evaluation is discussed in Section 3; for that speci c task, a simple manipulation of the input data yields a signi cant decrease in wallclock running time. I conclude with some thoughts on the design of MapReduce algorithms in light of these ndings.
It is a well-known observation that word occurrences in
natural language, to a rst approximation, follow a Zip an
distribution. Many other naturally-occurring phenomena,
ranging from website popularity to sizes of craters on the
moon, can be similarly characterized [
k s PN n=1 1=ns (1) where s is the characteristic exponent. The denominator is known as the N th generalized harmonic number, often written as HN;s. For many types of algorithms, all occurrences of the same type of element must be processed together. That is, the element type represents the nest grain of parallelism that can be achieved. Let us assume that the amount of \work" associated with processing a type of element is 2 y = 1 0 0 1
2 3 s (characteristic exponent) 4 5 O(n) with respect to the number of element instances. In other words, the amount of work necessary to process an element is proportional to its frequency. If we normalize the total amount of work to one, then the fraction of the total work that must be devoted to processing the most common element is:
Even if we assume unlimited resources and the ability to
process all element types in parallel, the running time of
an algorithm would still be bound by the time required to
process the most frequent element. In the same spirit as
Amdahl's Law [
As a concrete example, suppose N = 106 and s = 1. The most frequently-occurring element would be observed about 6.95% percent of the time|which means that the maximum theoretical speedup of any parallel algorithm is about a factor of 14 (assuming that all instances of the same element must be processed together). For this class of algorithms, the maximum theoretical speedup is simply HN;s. Figure 2 plots HN;s with varying values of s for N = 106. This simple theoretical model shows that Zip an distributions present a serious impediment to the development of e cient parallel algorithms for a large class of real-world problems.
This analysis can be directly applied to MapReduce algorithms. In a common scenario, each key corresponds to a single type of element, and the fraction of total values associated with each key can be characterized by a Zip an distribution. In other words, a few keys have a large number of associated values, while a very large number of keys have only a few values. Take the two following examples:
Inverted indexing. The well-known MapReduce algorithm for inverted indexing begins by mapping over input documents and emitting individual postings as intermediate key-value pairs. All postings associated with each term are gathered in the reducer, where the nal postings lists are created and written to disk. The most straightforward implementation of the algorithm divides the term space into roughly equal-sized partitions and assigns each to a reducer (typically, through hashing). This, however, does not take into account the inherent distribution of document frequencies (which characterizes the length of each postings list, and hence the amount of work that needs to be done for each term). Due to the Zip an distribution of term occurrences, the reducers assigned to very frequent terms will have signi cantly more work than the other reducers, and therefore take much longer to complete. These are the stragglers observed in the execution of the algorithm.
Computing PageRank over large graphs. Although
Google recently revealed that it has developed
infrastructure speci cally designed for large-scale graph algorithms
(called Pregel), the implementation of PageRank [
In practice, faster speed-ups than predicted are possible because in most cases some amount of local aggregation can be performed, thus making the skewed key-value distributions less pronounced. In MapReduce, this is accomplished with \combiners", which can dramatically increase e ciency. Nevertheless, the stragglers problem remains both common and severe.
In both of the examples presented above, the stragglers problem is most severe in the reduce stage of processing, since the distribution of the intermediate data can be characterized as Zip an. However, depending on the distribution of input key-value pairs, it is also possible to have the stragglers problem in the map phase|in fact, the case study presented in the next section examines such a case.
This paper presents a case study of the stragglers
problem that builds on the parallel queries algorithm described in
SIGIR 2009 [
Computing pairwise similarity on document collections is
a task common to a variety of problems such as clustering,
unsupervised learning, and text retrieval. One algorithm
presented in [
In the reduce phase, all associative arrays belonging to the same query are brought together. The reducer performs an element-wise sum of all the associative arrays (denoted by Merge in the pseudo-code): this adds up the contributions for each query term across all documents. The nal result is an associative array holding complete querydocument scores. In e ect, this algorithm replaces random access to the postings with a parallel scan of all postings. In processing a set of queries, each postings list is accessed only once|each mapper computes partial score contributions for all queries that contain the term. Pairwise similarity for the entire collection can be computed by running this algorithm over all blocks in the collection. For more details, please refer to additional discussions of this algorithm in the SIGIR 2009 main proceedings paper. 3.2
Experiments using the algorithm presented in Figure 3
were conducted on a collection of 4.59m MEDLINE
abstracts (from journals in the life and health sciences domain),
used in the TREC 2005 genomics track [
An inverted index was built from the entire collection.3 Figure 4 shows the document frequencies of all 1.3m unique terms in the collection: on the x-axis, the document frequency of a term, and on the y-axis, the number of terms that have that df. Approximately 753k terms appear only once, and approximately 168k terms appear only twice. The term with the largest df appeared in 1.62m documents (or about one in three documents). This distribution is fairly typical of information retrieval text collections.
The thick line in Figure 5 plots the progress of the MapReduce job on the unmodi ed index of the document collection: wall-clock time (in seconds) on the x-axis and number of active workers (either mappers or reducers) on the y-axis. At the beginning of the job, 500 mappers are started by the MapReduce runtime; the number of running mappers decreases as the job progresses. When all the mappers have nished, 200 reducers start up after a short lull (during this time the framework is copying intermediate key-value pairs from mappers to reducers). Finally, the job ends when all reducers complete and the results have been written to disk.
The map phase of execution completes in 1399s; the
reduce phase begins at the 1418s mark, and the entire job
nishes in 1675s. The long tail of the plot in the map phase
graphically illustrates the stragglers problem. Consider the
following statistics, which can be interpreted as checkpoints
corresponding to 98.0%, 99.0%, and 99.8% mapper progress:
At the 467s mark, all except for 10 mappers (2% of
the total) have completed.
2The hardware con guration is the same as the setup in [
At the 532s mark, all except for 5 mappers (1% of the total) have completed.
At the 1131s mark, all except for one mapper have nished|the last mapper would run for about another 4.5 minutes before nishing.
Since all mappers must complete before the reducers can begin processing intermediate key-value pairs, the running time of the map phase is dominated by the slowest mapper, which as shown here takes signi cantly longer than the rest of the mappers. Repeated trials of the same experiment produced essentially the same behavior (not shown).
The behavior of the parallel queries algorithm is explained by the empirical distribution of the length of the postings lists (see Figure 4). There are a few very long postings lists (corresponding to common terms such as \gene" or \patient"), while the vast majority of the postings lists are very short. Since for each query term the postings must be traversed to compute partial document score contributions, the amount of work involved in processing a query term is on the order of the length of the postings list. Therefore, the mappers assigned to process common query terms will run for a disproportionately long time|these are exactly the stragglers observed in Figure 5. These empirical results are consistent with the theoretical model presented in Section 2. However, the situation is not quite as grim in some cases|in a retrieval application, user queries are less likely to contain common terms since they are typically less helpful in specifying an information need.
A natural solution to the stragglers problem, in this case, is to break long postings lists into shorter ones. Exactly such a solution was implemented: postings lists were split into 100k segments, so that terms contained in more than 100k documents were associated with multiple postings lists. Furthermore, the ordering of all the postings segments were randomized (as opposed to alphabetically sorted by term, as in the original inverted index).4 Note that this modi cation to the inverted index did not require any changes to 4This algorithm was straightforwardly implemented in MapReduce, by mapping over the original inverted index and writing a new copy. the parallel queries algorithm. Although partial score contributions for a single query term might now be computed in di erent mappers, all partial scores will still be collected together in the same reducer during the element-wise sum across associative arrays keyed by the same query id.
The thin (red) line in Figure 5 plots the progress of the MapReduce job on the modi ed index. The map phase of execution completes in 636s; the reduce phase begins at the 665s mark, and the entire job nishes in 831s. With this simple modi cation to the inverted index, the same algorithm runs in about half the wall-clock time. Table 1 compares the 98.0%, 99.0%, 99.8%, and 100.0% progress of the mappers for both the original and segmented postings list. Multiple trials of the same experiment gave rise to similar results.
As a follow up, additional experiments were conducted with even smaller postings segments, but ner splits did further decrease running time. This is likely due to the distribution of query terms|some postings lists will never be traversed no matter how nely segmented they are, since the query documents contain only a subset of all terms in the collection (and by implication, some mappers will do little work regardless).
Why does this simple manipulation of the input data work so well? It is important to note that the technique does not actually reduce the number of computations required to produce query-document scores. The total amount of work (i.e., area under the curve) is the same|however, with the segmented index, work is more evenly distributed across the mappers. In essence, splitting long postings lists is a simple, yet e ective, approach to \defeating" Zip an distributions for this particular application.
The limits on parallelization imposed by Zip an distributions is based on one important assumption|that all instances of the same type of element must be processed together. The simple approach to overcoming the stragglers problem is to devise algorithms that do not depend on this assumption. In the parallel queries case, this required only manipulating input data and no modi cations to the algorithm: query-document scores could be computed independently, since the associative arrays in which they are held would eventually be brought together and merged in the reduce stage.
Although the parallel queries case study dealt with stragglers in the map phase due to distributional characteristics of the input data, the same principles can be applied to stragglers in the reduce phase as well. However, there are additional complexities that need to be considered. Often, it is known in advance that intermediate data can be characterized as Zip an|but devising algorithms to address the issue may require actually knowing which elements occupy the head of the distribution. This in turn requires executing the algorithm itself, which may be slow precisely because of the stragglers problem. This chicken-and-egg dependency can be broken by sampling strategies, but at the cost of greater complexity in algorithm design.
For solving reduce-phase stragglers, once the distribution of intermediate data has been characterized, a more intelligent partitioner can better distribute load across the reducers. Unfortunately, this may still be insu cient in some cases, for example, PageRank computations. Recall that in computing PageRank all contributions from inbound links must be summed, thereby creating the requirement that all values associated with the same key (i.e., vertex in graph) be processed together. The only recourse here is a modi cation of the original algorithm (e.g., a multi-stage process for totaling the PageRank contributions of pages with large in-degrees).
The upshot of this discussion is that to overcome the e ciency bottleneck imposed by Zip an distributions, developers must apply application-speci c knowledge to parallelize the processing of common elements. This, in turn, depends on the ingenuity of the individual developer and requires insight into the problem being tackled. 5.
This paper explores the stragglers problem in MapReduce caused by Zip an distributions common in many real-world datasets. What are the broader implications of these ndings? I believe that two lessons are apparent:
First, e cient MapReduce algorithms are not quite as easy to design as one might think. The allure of MapReduce is that it presents a simple programming model for designing scalable algorithms. While this remains an accurate statement, the deeper truth is that there is still quite a bit of \art" in designing e cient MapReduce algorithms for non-toy problems|witness the case study presented in this paper, where a simple tweak to the input data cut running time in half.5 Second, MapReduce algorithms cannot be studied in isolation, divorced from real-world applications|the stragglers problem is caused by properties of datasets (critically, not from the processing architecture or inherent bottlenecks in the algorithm). Furthermore, since there does not appear to be a general-purpose, universally-applicable solution to the problem, it is of limited value to discuss algorithmic performance without reference to speci c applications.
Hopefully, these two lessons will be useful to future designers of MapReduce algorithms. 6.
This work was supported by the following sources: the Intramural Research Program of the NIH, National Library of Medicine; NSF under awards IIS-0705832 and IIS-0836560 (under the CLuE program); Google and IBM, via the Academic Cloud Computing Initiative. I am grateful to Esther and Kiri for their love. 5Although one might argue how obvious this solution is, I think it is safe to say that the solution requires a degree of understanding of both information retrieval algorithms and MapReduce that a novice would be unlikely to have.