WIXLIB

Google MapReduce Data parallel framework for processing BigData on large commodity hardware Transparently tolerates– Data faults– Computation faults Achieves– Scalability and fault tolerance

Commodity ClustersMapReduce is designed to efficiently process big data using regularcommodity computersA theoretical 1000-CPU machine would cost a very large amount ofmoney, far more than 1000 single-CPU or 250 quad-corecommodity machinesPremise: MapReduce ties smaller and more reasonably pricedmachines together into a single cost-effective commodity cluster tosolve Big Data problems5

Three Strategies Parallelism– Break down jobs into distributed independenttasks to exploit parallelism Scheduling– Consider data-locality and variations in overallsystem workloads for scheduling Fault Tolerance– Transparently tolerate data and task failures

Hadoop MapReduce Hadoop is an open source implementation of MapReduce– 2006 Hadoop presents MapReduce as an analytics engine and underthe hood uses a distributed storage layer referred to as HadoopDistributed File System (HDFS) HDFS mimics Google File System (GFS) Applications in MapReduce are represented as jobs Each job encompasses several map and reduce tasks Map and reduce tasks operate on data independently and in parallel7

MapReduce In a Nutshell MapReduce incorporates two phases Map Phase Reduce nReduceTaskPartitionPartitionShuffle StageMap PhaseMerge &SortStageReduce PhaseTo HDFSReduce Stage

Data Distribution In a MapReduce cluster, data is distributed to all the nodes of the clusteras it is being loaded An underlying distributed file systems (e.g., GFS, HDFS) splits large datafiles into chunks which are managed by different nodes in the clusterInput data: A large fileNode 1Node 2Node 3Chunk of input dataChunk of input dataChunk of input data Even though the file chunks are distributed across several machines, theyform a single namespace9

Network Topology In MapReduce MapReduce assumes a tree style network topology Nodes are spread over different racks embraced in one or many data centers The bandwidth between two nodes is dependent on their relative locations in thenetwork topology The assumption is that nodes that are on the same rack will have higher bandwidthbetween them as opposed to nodes that are off-rack

Computing Units: TasksMapReduce divides the workload into multiple independent tasks andautomatically schedule them on cluster nodesA work performed by each task is done in isolation from one anotherThe amount of communication which can be performed by tasks is limitedmainly for scalability and fault-tolerance reasons11

MapReduce Phases The output from the mappers is denoted asintermediate output and broughtinto a second set of tasks called Reducers The process of reading intermediate output intoa set of Reducers is known as shufflingThe Reducers produce the final outputC0C1C2C3mappers M0M1M2M3IOIOIOIOsplitsMap PhaseIn MapReduce, splits are processed inisolation by tasks called MappersReduce Phase Shuffling DataMerge & SortOverall, MapReduce breaks the data flow into two phases,map phase and reduce phaseReducersR0R1FOFO

Keys and Values The programmer in MapReduce has to specify two functions, themap function and the reduce function that implement the Mapperand the Reducer in a MapReduce program In MapReduce data elements are always structured as key-value(i.e., (K, V)) pairs The map and reduce functions receive and emit (K, V) pairsInput Splits(K, V)PairsIntermediate OutputMapFunction(K’, V’)PairsFinal OutputReduceFunction(K’’, V’’)Pairs

Splits A split can contain reference to one or more HDFS blocks Configurable parameter Each map task processes one split# of splits dictates the # of map tasksBy default one split contains reference to one HDFS blockMap tasks are scheduled in the vicinity of HDFS blocks toreduce network traffic

Partitions Map tasks store intermediate output on local disk (not HDFS) A subset of intermediate key space is assigned to each Reducer hash(key) mod R These subsets are known as partitionsDifferent colors representdifferent keys (potentially)from different MappersPartitions are the inputto Reducers

Hadoop MapReduce: A Closer LookNode 1Node 2Files loaded from local HDFS storeFiles loaded from local HDFS put (K, V) pairsInput (K, V) pairsMapMapIntermediate (K, V) termediate(K,V) pairsexchanged byall nodesMapMapIntermediate (K, V) pairsPartitionerSortReduceFinal (K, V) pairsWriteback to localHDFS storeRecordReadersFinal (K, V) pairsOutputFormatOutputFormatWriteback to localHDFS store

Input Files Input files are where the data for a MapReduce task isinitially stored The input files typically reside in a distributed file system(e.g. HDFS) The format of input files is arbitrary Line-based log filesBinary filesMulti-line input recordsOr something else entirelyfilefile17

InputFormat How the input files are split up and read is definedby the InputFormat InputFormat is a class that does the following: Selects the files that should be used for inputDefines the InputSplits that break a fileProvides a factory for RecordReader objects that read the fileFiles loaded from local HDFS storeInputFormatfilefile18

InputFormat Types Several InputFormats are provided with atDefault format;reads lines of textfilesThe byte offsetof the lineThe line contentsKeyValueInputFormatParses lines into(K, V) pairsEverything upto the first tabcharacterThe remainder ofthe lineSequenceFileInputFormatA Hadoop-specifichigh-performancebinary formatuser-defineduser-definedMyInputFormatA user-specifiedinput formatuser-defineduser-defined19

Input Splits An input split describes a unit of data that a single map task in aMapReduce program will process By dividing the file into splits, we allowseveral map tasks to operate on a singlefile in parallelFiles loaded from local HDFS storeInputFormat If the file is very large, this can improveperformance significantly through parallelismfileSplitfile Each map task corresponds to a single input splitSplitSplit

RecordReader The input split defines a slice of data but does not describe howto access it The RecordReader class actually loads data from its source and converts itinto (K, V) pairs suitable for reading by Mappers The RecordReader is invoked repeatedlyon the input until the entire split is consumed Each invocation of the RecordReader leadsto another call of the map function definedby the programmerFiles loaded from local HDFS storeInputFormatfileSplitSplitSplitRRRRRRfile

Mapper and Reducer The Mapper performs the user-defined work of thefirst phase of the MapReduce programFiles loaded from local HDFS storeInputFormatA new instance of Mapper is created for each splitfileThe Reducer performs the user-defined work ofthe second phase of the MapReduce programSplitSplitSplitRRRRRRMapMapMapfileA new instance of Reducer is created for each partition For each key in the partition assigned to a Reducer, theReducer is called oncePartitionerSortReduce

Partitioner Each mapper may emit (K, V) pairs to any partitionFiles loaded from local HDFS store Therefore, the map nodes must all agree onwhere to send different pieces ofintermediate ile The partitioner class determines whichpartition a given (K,V) pair will go to The default partitioner computes a hash value for agiven key and assigns it to a partition based onthis resultPartitionerSortReduce

Sort (merge) Each Reducer is responsible for reducingthe values associated with (several)intermediate keys The set of intermediate keys on a singlenode is automatically sorted (merged) byMapReduce before they are presentedto the ReducerFiles loaded from local HDFS filePartitionerSortReduce

OutputFormatFiles loaded from local HDFS store The OutputFormat class defines the way (K,V) pairsproduced by Reducers are written to output filesInputFormatfile The instances of OutputFormat provided byHadoop write to files on the local disk or in HDFSSplitSplitSplitRRRRRRMapMapMapfileSeveral OutputFormats are provided by ult; writes lines in "key \tvalue" formatSequenceFileOutputFormatWrites binary files suitable forreading into subsequentMapReduce jobsNullOutputFormatGenerates no output filesPartitionerSortReduceOutputFormat

Combiner Functions MapReduce applications are limited by the bandwidth available on the clusterIt pays to minimize the data shuffled between map and reduce tasksHadoop allows the user to specify a combiner function (just like the reducefunction) to be run on a map output only if the Reduce function is commutativeand associative.(Y, T) CombinerMapoutputMTMTN(1950, 0)(1950, 20)(1950, 10)output(1950, 20)MTRNMTMTNMTRNMTLEGEND:RT R Rack N Node MT Map Task RT Reduce Task Y Year T Temperature

MapReduce In a Nutshell MapReduce incorporates two phases, Map and Reduce.

The Shuffle in MapReduce

Job Scheduling in MapReduce In MapReduce, an application is represented as a job A job encompasses multiple map and reduce tasks Job schedulers in MapReduce are pluggable Hadoop MapReduce by default FIFO scheduler for jobs Schedules jobs in order of submissionStarvation with long-running jobsNo job preemptionNo evaluation of job priority or size29

Multi-user Job Scheduling in MapReduce Fair scheduler (Facebook)– Pools of jobs, each pool is assigned a set of shares– Jobs get (on ave) equal share of slots over time– Across pools, use Fair scheduler, within pool, FIFOor Fair scheduler Capacity scheduler (Yahoo!)– Creates job queues– Each queue is configured with # (capacity) of slots– Within queue, scheduling is priority based

Task Scheduling in MapReduce MapReduce adopts a master-slave architectureTTTask Slots The master node in MapReduce is referredto as Job Tracker (JT) Each slave node in MapReduce is referredto as Task Tracker (TT)JTTasks QueueT0T0T1T1T2TT MapReduce adopts a pull scheduling strategy rather thana push one Task SlotsI.e., JT does not push map and reduce tasks to TTs but rather TTs pull them bymaking requests31

Map and Reduce Task Scheduling Every TT sends a heartbeat message periodically to JT encompassing arequest for a map or a reduce task to runI. Map Task Scheduling: JT satisfies requests for map tasks via attempting to schedule mappers in thevicinity of their input splits (i.e., it considers locality)II. Reduce Task Scheduling: However, JT simply assigns the next yet-to-run reduce task to a requesting TTregardless of TT’s network location and its implied effect on the reducer’sshuffle time (i.e., it does not consider locality)32

Task Scheduling

Task Scheduling in Hadoop A golden principle adopted by Hadoop is: “Moving computation towards datais cheaper than moving data towards computation”– Hadoop applies this principle to Map task scheduling With map task scheduling, once a slave (or a TaskTracker- TT) polls for a maptask, M, at the master node (or the JobTracker- JT), JT attempts to assign TTan M that has its input data local to TTCore SwitchRack Switch 1TaskTracker1TaskTracker2Rack Switch 2TaskTracker3TaskTracker4TaskTracker5MT2MT3Request a Map TaskSchedule a Map Task at an Empty Map Slot on TaskTracker1JobTrackerMT1 MT2 MT3

Task Scheduling in Hadoop Hadoop does not apply the locality principle to Reduce task scheduling With reduce task scheduling, once a slave (or a TaskTracker- TT) polls for areduce task, R, at the master node (or the JobTracker- JT), JT assigns TT any RCSRS1TT1TT2TT4TT3Request Reduce Task RAssign R to TT1Shuffle PartitionsA locality problem,where R is scheduledat TT1 while itsRS2partitions existat TT4TT5JT35CS Core Switch & RS Rack Switch

Fault Tolerance in Hadoop Data redundancy Achieved at the storage layer through replicas (default is 3) Stored at physically separate machines Can tolerate– Corrupted files– Faulty nodes HDFS:– Computes checksums for all data written to it– Verifies when reading Task Resiliency (task slowdown or failure) Monitors to detect faulty or slow tasks Replicates tasks36

Task Resiliency MapReduce can guide jobs toward a successful completion even when jobs arerun on a large cluster where probability of failures increases The primary way that MapReduce achieves fault tolerance is throughrestarting tasks If a TT fails to communicate with JT for a period of time (by default, 1 minute inHadoop), JT will assume that TT in question has crashed If the job is still in the map phase, JT asks another TT to re-execute allMappers that previously ran at the failed TT If the job is in the reduce phase, JT asks another TT to re-execute allReducers that were in progress on the failed TT37

Speculative Execution A MapReduce job is dominated by the slowest task MapReduce attempts to locate slow tasks (stragglers) and run redundant(speculative) tasks that will optimistically commit before the correspondingstragglers This process is known as speculative execution Only one copy of a straggler is allowed to be speculated Whichever copy of a task commits first, it becomes the definitive copy,and the other copy is killed by JT

Locating Stragglers How does Hadoop locate stragglers? Hadoop monitors each task progress using a progress score between0 and 1 based on the amount of data processed If a task’s progress score is less than (average – 0.2), and the task hasrun for at least 1 minute, it is marked as a stragglerNot a stragglerT1PS 2/3A stragglerT2PS 1/12Time

Issues with Speculative Execution Susceptible in heterogeneous environments– If transient congestion, lots of speculative tasks Launches speculative tasks without checking speed ofTT or load of speculative task– Slow TT will become slower Locality always trumps– If 2 speculative tasks ST1 & ST2 With stragglers T1@70% and T2@20% If task slot is local to ST2’s HDFS block, ST2 gets scheduled Three reduce stages treated equally– Shuffle stage is typically slower than the merge & sort andreduce stages

MapReduce ApplicationsMapInput DataShuffledDataLocal Disk orNetworkData PatternMap InputShuffle DataOutputDataReduceNetworkNetworkShuffle Data/MapInput RatioExample0Sobel EdgeDetection 1Grep 1Sort 1Some IR apps

Grep ExampleGrep Search Map & Reduce task detailsReduce Task 1 - Sorting0:00:36Reduce Task 1 - Shuffling0:00:35Reduce Task 10:00:37Map task 150:00:05Map task 140:00:09Map task 130:00:08Map task 120:00:09Map task 110:00:09Map task 100:00:11Map task 90:00:09Map task 80:00:08Map task 70:00:09Map task 60:00:10Map task 5Map task 4Map task 3Map task 2Map task 0900:00:1700:00:2600:00:3500:00:4300:00:52

TeraSort ExampleTeraSort Map & Reduce Task DetailsReduce Task 1 - Sorting0:01:50Reduce Task 1 - Shuffling0:01:28Reduce Task 10:02:54Map task 160:00:10Map task 150:00:09Map task 140:00:22Map task 130:00:13Map task 120:00:15Map task 110:00:16Map task 100:00:15Map task 90:00:18Map task 80:00:13Map task 70:00:12Map task 60:00:31Map task 5Map task 40:00:130:00:16Map task 30:00:13Map task 20:00:12Map task :03:36

What Makes MapReduce Popular? MapReduce is characterized by:1. Its simplified programming model which allows the user toquickly write and test distributed systems2. Its efficient and automatic distribution of data and workloadacross machines3. Its flat scalability curve. Specifically, after a Mapreduceprogram is written and functioning on 10 nodes, very little-ifany- work is required for making that same program run on1000 nodes4. Its fault tolerance approach44

Programming Models and Frameworks Advanced Cloud Computing 15-719/18-847b Garth Gibson Greg Ganger M