WIXLIB

Index filethe map function to each record.Partition 0Offset2. After the map function has been applied to each input record, the commit phase registers the final output with the TaskTracker, which then informs theJobTracker that the task has finished executing.Partition 1OffsetFigure 1 contains the interface that must be implemented by user-defined map functions. After the mapfunction has been applied to each record in the split, theclose method is invoked.The third argument to the map method specifies anOutputCollector instance, which accumulates the outputrecords produced by the map function. The output of themap step is consumed by a reduce function, so the OutputCollector must store the map output in a format thatis easy for the reducer to consume. Intermediate keys areassigned to reducers by applying a partitioning function,so the OutputCollector applies that function to each keyproduced by the map function, and stores each recordand partition number in an in-memory buffer. The OutputCollector is responsible for spilling this buffer to diskwhen it reaches capacity.A spill of the in-memory buffer involves first sortingthe records in the buffer by partition number and then bykey. The buffer content is written to the local file systemas an index file and a data file (Figure 2). The index filepoints to the offset of each partition in the data file. Thedata file contains only the records, which are sorted bythe key within each partition segment.After a map task has applied the map function to eachinput record, it enters the commit phase. To generatethe task’s final output, the in-memory buffer is flushedto disk, and all of the spill files generated during the mapphase are merged into a single data and index file. Theseoutput files are registered with the TaskTracker beforethe task completes. The TaskTracker will read these fileswhen servicing requests from reduce tasks.2.4Data fileKey SizeValue SizeKey BytesValue BytesKey SizeValue SizeKey BytesValue BytesKey SizeValue SizeKey BytesValue BytesKey SizeValue SizeKey BytesValue Bytespartition 0recordpartition 1Figure 2: Map task index and data file format (2 partition/reduce case).public interface Reducer K2, V2, K3, V3 {void reduce(K2 key, Iterator V2 values,OutputCollector K3, V3 output);void close();}Figure 3: Reduce function interface.In the shuffle phase, a reduce task fetches data fromeach map task by issuing HTTP requests to a configurable number of TaskTrackers at once (5 by default).The JobTracker relays the location of every TaskTrackerthat hosts map output to every TaskTracker that is executing a reduce task. In traditional batch-oriented Hadoop,a reduce task cannot fetch the output of a map task untilthe map has finished executing and committed its finaloutput to disk.After receiving its partition from all map outputs, thereduce task enters the sort phase. The map output foreach partition is already sorted by key. The reduce taskmerges these runs together to produce a single run that issorted by the key. The task then enters the reduce phase,in which it invokes the user-defined reduce function foreach distinct key in sorted order, passing it the associated list of values. The output of the reduce function iswritten to a temporary location on HDFS. After the reduce function has been applied to each key in the reducetask’s partition, the task’s HDFS output file is atomicallyrenamed from its temporary location to its final location.In this design, the output of both map and reduce tasksis written to disk before it can be consumed. This is particularly expensive for reduce tasks, because their output is written to HDFS. By default, this requires a synchronous write operation that must store three copies ofReduce Task ExecutionThe execution of a reduce task is divided into threephases.1. The shuffle phase fetches the reduce task’s inputdata. Each reduce task is assigned a partition of thekey range produced by the map step, so the reducetask must fetch the content of this partition from every map task’s output.2. The sort phase groups records with the same keytogether.3. The reduce phase applies the user-defined reducefunction to each key and corresponding list of values.3

HDFSHDFSPush(Final Result)Push(Final ous)performance of stock Hadoop with our pipelining implementation in Section 3.5.3.1As described in Section 2.4, reduce tasks traditionallyissue HTTP requests to pull their output from each TaskTracker. This means that map task execution is completely decoupled from reduce task execution. To support pipelining, we modified the map task to instead pushdata to reducers as it is produced. To give an intuition forhow this works, we begin by describing a straightforwardpipelining design, and then discuss the changes we hadto make to achieve good performance.PushLocal Memory(Filesystem)Local Memory(Cache chronous)HDFSHDFSMapReduceBatch DataflowMapReduceOnline Dataflow3.1.1Naı̈ve PipeliningIn our naı̈ve implementation, we want to directly connectmappers to reducers and pipe data between them. Whena client submits a new job to Hadoop, the JobTracker assigns the map and reduce tasks associated with the job tothe available TaskTracker slots. For purposes of discussion, let us assume that there are enough free slots to assign all the tasks for each job. The JobTracker also communicates the location of each map task to every reducetask. We modified Hadoop so that each reduce task contacts every map task upon initiation of the job, and opensa socket which will be used to pipeline the output of themap function. As each map output record is produced,the mapper determines which partition (reduce task) therecord should be sent to, and immediately sends it via theappropriate socket.A reduce task accepts the pipelined data it receivesfrom each map task and stores it an in-memory buffer,spilling sorted runs of the buffer to disk as needed. Oncethe reduce task learns that every map task has completed,it performs a final merge of all its sorted runs and appliesthe user-defined reduce function as normal, writing theoutput to HDFS.Figure 4: Hadoop dataflow for batch (left) and pipelined(right) processing of MapReduce computations.each output block on different nodes (to ensure fault tolerance).Output materialization simplifies fault tolerance, because it reduces the amount of state that must be restoredto consistency after a node failure. If any task (either mapor reduce) fails, the JobTracker simply schedules a newtask to perform the same work as the failed task. Sincea task never exports any data other than its final answer,no further recovery steps are needed.3Pipelining Within A JobPipelined MapReduceIn this section we discuss our extensions to Hadoop tosupport pipelining. We focus here on batch-processingtasks, postponing the discussion of the changes requiredfor online aggregation and continuous queries until thenext sections.Figure 4 depicts the dataflow of two MapReduce implementations. The dataflow on the left correspondsto the output materialization approach used by stockHadoop; the dataflow on the right allows pipelining. Inthe remainder of this section, we present our design andimplementation for the pipelined Hadoop dataflow. Wedescribe how our design supports fault tolerance (Section 3.3), and discuss the interaction between pipelining and task scheduling (Section 3.4). We compare the3.1.2RefinementsWhile the algorithm described above is straightforward,it suffers from several practical problems. First, it ispossible that there will not be enough slots available toschedule every task in a new job. Opening a socket between every map and reduce task also requires a largenumber of TCP connections. A simple tweak to the naı̈vedesign solves both problems: if a reduce task has not yetbeen scheduled, any map tasks that produce records forthat partition simply write them to disk. Once the reducetask is assigned a slot, it can then fetch the records fromthe map task, as in stock Hadoop. To reduce the number of concurrent TCP connections, each reducer can be4

configured to pipeline data from a bounded number ofmappers at once; the reducer will pull data from the remaining map tasks in the traditional Hadoop manner.Our initial pipelining implementation suffered froma second problem: the user-defined map function wasinvoked by the same thread that wrote output recordsto the pipeline sockets. This meant that if a networkI/O blocked (e.g. because the reducer was over-utilized),the mapper was prevented from doing useful work.Pipeline stalls should not prevent a map task from making progress—especially since, once the task has completed, it frees a TaskTracker slot that can be used forother purposes. We solved this problem by running themap function in a separate thread that stores its output inan in-memory buffer, and then having another thread periodically send the contents of the buffer to the pipeliningreducers.spill file will likely remain in the mapper machine’s kernel buffer cache). However, if the reducer begins to trailthe mapper, the number of unsent spill files will grow.In this case, the mapper periodically applies the combiner function to the spill files, merging multiple spillfiles together into a single larger file. This has the effectof adaptively moving load from the reducer to the mapper or vice versa, depending on which node is the currentbottleneck.The connection between pipelining and adaptive queryprocessing techniques has been observed elsewhere(e.g. [2]). The adaptive scheme outlined above is relatively simple, but we believe that adapting to feedbackalong pipelines has the potential to significantly improvethe utilization of MapReduce clusters.3.1.3Many practical computations cannot be expressed as asingle MapReduce job, and the outputs of higher-levellanguages like Pig [19] typically involve multiple jobs.In the traditional Hadoop architecture, the output of eachjob is written to HDFS in the reduce step, and then immediately read back from HDFS by the map step of thenext job. In fact, the JobTracker cannot even schedule aconsumer job until the producer job has completed, because scheduling a map task requires knowing the HDFSblock locations of the map’s input split.In our modified version of Hadoop, the reduce tasks ofone job can optionally pipeline their output directly to themap tasks of the next job, sidestepping the need for expensive fault-tolerant storage in HDFS for what amountsto a temporary file. Unfortunately, the computation ofthe reduce function from the previous job and the mapfunction of the next job cannot be overlapped: the final result of the reduce step cannot be produced untilall map tasks have completed, which prevents effectivepipelining. However, in the next sections we describehow online aggregation and continuous query pipelinescan publish “snapshot” outputs that can indeed pipelinebetween jobs.3.2Granularity of Map OutputAnother problem with the naı̈ve design is that it eagerlysends each record as soon as it is produced, which prevents the use of map-side combiners. Imagine a jobwhere the reduce key has few distinct values (e.g., gender), and the reduce applies an aggregate function (e.g.,count). As discussed in Section 2.1, combiners allow“map-side preaggregation”: by applying a reduce-likefunction to each distinct key at the mapper, networktraffic can often be substantially reduced. By eagerlypipelining each record as it is produced, there is no opportunity for the map task to apply a combiner function.A related problem is that eager pipelining moves someof the sorting work from the mapper to the reducer. Recall that in the blocking architecture, map tasks generatesorted spill files: all the reduce task must do is mergetogether the pre-sorted map output for each partition.In the naı̈ve pipelining design, map tasks send outputrecords in the order in which they are generated, so thereducer must perform a full external sort. Because thenumber of map tasks typically far exceeds the number ofreduces [6], moving more work to the reducer increasedresponse time in our experiments.We addressed these issues by modifying the inmemory buffer design we described in Section 3.1.2. Instead of sending the buffer contents to reducers directly,we instead wait for the buffer to grow to a thresholdsize. The mapper then applies the combiner function,sorts the output by partition and reduce key, and writesthe buffer to disk, using the spill file format describedin Section 2.3. A second thread monitors the spill files,and sends them to the pipelined reducers. If the reducersare able to keep up with the map task and the network isnot a bottleneck, a spill file will be sent to a reducer veryquickly after it has been produced (in which case, the3.3Pipelining Between JobsFault ToleranceOur pipelined Hadoop implementation is robust to thefailure of both map and reduce tasks. To recover frommap task failures, we added some bookkeeping to thereduce task to record which map task produced eachpipelined spill file. To simplify fault tolerance, the reducer treats the output of a pipelined map task as “tentative” until the JobTracker informs the reducer that themap task has committed successfully. The reducer canmerge together spill files generated by the same uncommitted mapper, but won’t combine those spill files with5

the output of other map tasks until it has been notifiedthat the map task has committed. Thus, if a map taskfails, each reduce task can ignore any tentative spill filesproduced by the failed map attempt. The JobTracker willtake care of scheduling a new map task attempt as instock Hadoop.If a reduce task fails and a new copy of the task isstarted, the new reduce instance must be sent all the input data that was sent to the failed reduce attempt. Ifmap tasks operated in a purely pipelined fashion anddiscarded their output after sending it to a reducer, thiswould be difficult. Therefore, map tasks retain their output data, and write a complete output file to disk beforecommitting as in normal Hadoop. This allows the map’soutput to be reproduced if any reduce tasks fail. Forbatch jobs, the key advantage of our architecture is thatreducers are not blocked waiting for the complete outputof the task to be written to disk.Our technique for recovering from map task failure isstraightforward, but places a minor limit on the reducer’sability to merge spill files. To avoid this, we envision introducing a “checkpoint” concept: as a map task runs, itwill periodically notify the JobTracker that it has reachedoffset x in its input split. The JobTracker will notify anyconnected reducers; map task output that was producedbefore offset x can then be merged by reducers with othermap task output as normal. To avoid duplicate results, ifthe map task fails, the new map task attempt resumesreading its input at offset x. This technique also has thebenefit of reducing the amount of redundant work doneafter a map task failure.3.4tion 7, there are many interesting options for schedulingpipelines or even DAGs of such jobs that we plan to investigate in future.3.5Performance EvaluationWe conducted a series of performance experiments usinga 60-node cluster on Amazon EC2. One node executedthe Hadoop JobTracker and the HDFS NameNode, whilethe remaining 59 nodes served as slaves for running theTaskTrackers and HDFS DataNodes. All nodes executedon “high-CPU medium” EC2 instances with 1.7GB ofmemory and 2 virtual cores. Each virtual core is theequivalent of a 2007-era 2.5Ghz Intel Xeon processor.We began by measuring the performance of a simpleMapReduce job that does not use a combiner. Sortingis commonly used as a benchmark for basic MapReduceperformance, because of the implicit sort done by the reduce phase. We sorted 5.5GB of article text extractedfrom Wikipedia; each word from the text was parsed asa separate record. Figure 5 describes sort performanceon the EC2 cluster using an HDFS block size of 128MB(yielding 40 map tasks). We configured the system to use59 reducers. In each graph, we give the CDF of map andreduce task completion. The left and right graphs depictblocking and pipelined performance, respectively.Pipelining dominates blocking for this configuration,in part because it achieves better cluster utilization: thereduce tasks in the blocking job were idle for the first192 seconds of the experiment, whereas for the pipelinedjob, reducers began doing useful work within 20 seconds.Note that in a highly-utilized cluster, increased pipelineparallelism would not necessarily lead to an improvement in total throughput. However, these results suggestthat pipelining can substantially reduce the response timeof an individual job, which can often be important (e.g.quickly executing high-priority jobs).Task SchedulingThe Hadoop JobTracker had to be retrofitted to be awareof the pipelined execution of jobs. Intra-job pipeliningwas partially handled by the baseline version of Hadoop,which co-schedules map and reduce tasks that are part ofthe same job. However, the JobTracker was not awareof inter-job dependencies. In stock Hadoop, a client thatneeds to submit a series of jobs (that possibly consist of alarger query) must do so in order of their dataflow dependencies. That is, a job that consumes the output of one ormore other jobs cannot be submitted until the producerjobs have executed to completion.Hadoop exports an interface to client applications forsubmitting jobs. We extended this interface to accepta list (pipeline) of jobs, where each job in the list depends on the job before it. The client interface traversesthis list annotating each job with the identifier of the jobthat it depends on. The JobTracker scheduler looks forthis annotation and co-schedules jobs with their dependencies, giving slot preference to “upstream” jobs overthe “downstream” jobs they feed. As we note in Sec-4Online AggregationAlthough MapReduce was originally designed as abatch-oriented system, it is often used for interactive dataanalysis: a user submits a job to extract information froma data set, and then waits to view the results before proceeding with the next step in the data analysis process.This trend has accelerated with the development of highlevel query languages that are executed as MapReducejobs, such as Hive [25], Pig [19], and Sawzall [22].Traditional MapReduce implementations provide apoor interface for interactive data analysis, because theydo not emit any output until the job has been executed tocompletion. However, in many cases, an interactive userwould prefer a “quick and dirty” approximation over acorrect answer that takes much longer to compute. In the6

BlockingPipeliningReduce progressMap progress100%100%80%80%ProgressProgressMap progress60%40%20%Reduce 100200300Time (seconds)400500600700800900Time (seconds)Figure 5: CDF of map and reduce task completion times for a sort job on 5.5GB of text extracted from Wikipedia.The total job runtimes were 927 seconds for blocking, and 610 seconds for pipelining.database literature, online aggregation has been proposedto address this problem [11], but the batch-oriented nature of traditional MapReduce implementations makesthese techniques difficult to apply. In this section, weshow how we extended our pipelined Hadoop implementation to support online aggregation within a single job(Section 4.1) and between multiple jobs (Section 4.2).We show that online aggregation has a minimal impacton job completion times, and can often yield an accurate approximate answer long before the job has finishedexecuting.4.1to include data from tentative (unfinished) map tasks.This option does not affect the fault tolerance design described in Section 3.3. In the current prototype, eachsnapshot is stored in a directory on HDFS. The name ofthe directory includes the progress value associated withthe snapshot. Each reduce task runs independently, andat a different rate; once a reduce task has made sufficientprogress, it writes a snapshot to a temporary directory onHDFS, and then atomically renames it to the appropriatesnapshot directory.Applications can consume snapshots by polling HDFSin a predictable location. An application knows that agiven snapshot has been completed when every reducetask has written a file to the snapshot directory. Atomicrename is used to avoid applications mistakenly readingincomplete snapshot files (the same technique is used bystock Hadoop when writing the final output of each reduce task).Note that if there are not enough free slots to allow allthe reduce tasks in a job to be scheduled, snapshots willnot be available for reduce tasks that are still waiting tobe executed.

2.2 Hadoop Architecture Hadoop is composed of Hadoop MapReduce, an im-plementation of MapReduce designed for large clusters, and the Hadoop Distributed File System (HDFS), a file system optimized for batch-oriented workloads such as MapReduce. In most Hadoop jobs, HDFS is used to store both the input to the map step and the output of the .