Handling Partitioning Skew in MapReduce using LEEN Shadi Ibrahim, Hai Jin, Lu Lu, Bingsheng He, Gabriel Antoniu, Song Wu To cite this version: Shadi Ibrahim, Hai Jin, Lu Lu, Bingsheng He, Gabriel Antoniu, et al.. Handling Partitioning Skew in MapReduce using LEEN. Peer-to-Peer Networking and Applications, 2013. ￿hal-00822973￿ HAL Id: hal-00822973 https://hal.inria.fr/hal-00822973 Submitted on 28 Jun 2016 HAL is a multi-disciplinary open access L’archive ouverte pluridisciplinaire HAL, est archive for the deposit and dissemination of sci- destinée au dépôt et à la diffusion de documents entific research documents, whether they are pub- scientifiques de niveau recherche, publiés ou non, lished or not. The documents may come from émanant des établissements d’enseignement et de teaching and research institutions in France or recherche français ou étrangers, des laboratoires abroad, or from public or private research centers. publics ou privés. Noname manuscript No. (will be inserted by the editor) Handling Partitioning Skew in MapReduce using LEEN Shadi Ibrahim⋆ Hai Jin Lu Lu · · · Bingsheng He Gabriel Antoniu Song · · Wu Received:date/Accepted:date Abstract MapReduce is emerging as a prominent tool for big data process- ing. Data locality is a key feature in MapReduce that is extensively leveraged in data-intensive cloud systems: it avoids network saturation when processing large amounts of data by co-allocating computation and data storage, partic- ularly for the map phase. However, our studies with Hadoop, a widely used MapReduce implementation, demonstrate that the presence of partitioning skew1 causes a huge amount of data transfer during the shuffle phase and leadstosignificantunfairnessonthereduceinputamongdifferentdatanodes. As a result, the applications severe performance degradation due to the long data transfer during the shuffle phase along with the computation skew, par- ticularly in reduce phase. In this paper, we develop a novel algorithm named LEEN for locality- aware and fairness-aware key partitioning in MapReduce. LEEN embraces an asynchronousmapandreducescheme.Allbufferedintermediatekeysarepar- ⋆ correspondingauthor ShadiIbrahim,GabrielAntoniu INRIARennes-BretagneAtlantique Rennes,France E-mail: shadi.ibrahim,gabriel.antoniu @inria.fr { } HaiJin,LuLu,SongWu ClusterandGridComputingLab ServicesComputingTechnologyandSystemLab HuazhongUniversityofScienceandTechnology Wuhan,China E-mail:[email protected] BingshengHe SchoolofComputerEngineering NanyangTechnologicalUniversitySingapore E-mail:[email protected] 1 Partitioning skew refers to the case when a variation in either the intermediate keys’ frequenciesortheirdistributionsorbothamongdifferentdatanodes. 2 ShadiIbrahimetal. titioned according to their frequencies and the fairness of the expected data distribution after the shuffle phase. We have integrated LEEN into Hadoop. Our experiments demonstrate that LEEN can efficiently achieve higher local- ity and reduce the amount of shuffled data. More importantly, LEEN guar- antees fair distribution of the reduce inputs. As a result, LEEN achieves a performance improvement of up to 45% on different workloads. Keywords MapReduce Hadoop cloud computing skew partitioning · · · · intermediate data; 1 Introduction MapReduce [1], due to its remarkable features in simplicity, fault tolerance, andscalability,isbyfarthemostsuccessfulrealizationofdataintensivecloud computing platforms[2]. It is often advocated as an easy-to-use, efficient and reliable replacement for the traditional programming model of moving the data to the cloud[3]. Many implementations have been developed in different programminglanguagesforvariouspurposes[4][5][6].Thepopularopensource implementationofMapReduce,Hadoop[7],wasdevelopedprimarilybyYahoo, whereitprocesseshundredsofterabytesofdataontensofthousandsofnodes [8],andisnowusedbyothercompanies,includingFacebook,Amazon,Last.fm, and the New York Times [9]. The MapReduce system runs on top of the Google File System (GFS) [10], within which data is loaded, partitioned into chunks, and each chunk replicated across multiple machines. Data processing is co-located with data storage:whenafileneedstobeprocessed,thejobschedulerconsultsastorage metadata service to get the host node for each chunk, and then schedules a “map” process on that node, so that data locality is exploited efficiently. The map function processes a data chunk into key/value pairs, on which a hash partitioningfunctionisperformed,ontheappearanceofeachintermediatekey produced by any running map within the MapReduce system: hash (hash code (Intermediate-Keys) module ReduceID) The hashing results are stored in memory buffers, before spilling the in- termediate data (index file and data file) to the local disk [11]. In the reduce stage, a reducer takes a partition as input, and performs the reduce function on the partition (such as aggregation). Naturally, how the hash partitions are storedamongmachinesaffectsthenetworktraffic,andthebalanceofthehash partition size is an important indicator for load balancing among reducers. In this work, we address the problem of how to efficiently partition the intermediate keys to decrease the amount of shuffled data, and guarantee fair distribution of the reducers’ inputs, resulting in improving the overall perfor- mance. While, the current Hadoop’s hash partitioning works well when the keys are equally appeared and uniformly stored in the data nodes, with the presence of partitioning skew, the blindly hash-partitioning is inadequate and can lead to: HandlingPartitioningSkewinMapReduceusingLEEN 3 1. Networkcongestioncausedbythehugeamountofshuffleddata,(forexam- ple,inwordcountapplication,theintermediatedataare1.7timesgreaterin sizethanthemapsinput,thustacklingthenetworkcongestionbylocality- aware map executions in MapReduce systems is not enough); 2. unfairness of reducers’ inputs; and finally 3. severe performance degradation [12] (i.e. the variance of reducers’ inputs, in turn, causes a variation in the execution time of reduce tasks, resulting in longer response time of the whole job, as the job’s response time is dominated by the slowest reduce instance). Recentresearchhasreportedontheexistenceofpartitioningskewinmany MapReduce applications [12][13][14], but none of the current MapReduce im- plementations have overlooked the data skew issue[15]. Accordingly, in the presenceofpartitioningskew,theexistingshufflestrategyencounterstheprob- lems of long intermediate data shuffle time and noticeable network overhead. To overcome the network congestion during the shuffle phase, we propose to expose the locality-aware concept to the reduce task; However, locality-aware reduce execution might not be able to outperform the native MapReduce due tothepenaltiesofunfairnessofdatadistributionaftertheshufflephase,result- inginreducecomputationskew.Toremedythisdeficiency,wehavedeveloped an innovative approach to significantly reduce data transfer while balancing the data distribution among data nodes. Recognizing that the network congestion and unfairness distribution of reducers’ inputs, we seek to reduce the transferred data during the shuffle phase,aswellasachievingamorebalancedsystem.Wedevelopanalgorithm, locality-aware and fairness-aware key partitioning (LEEN), to save the net- work bandwidth dissipation during the shuffle phase of the MapReduce job along with balancing the reducers’ inputs. LEEN is conducive to improve the data locality of the MapReduce execution efficiency by the virtue of the asynchronous map and reduce scheme, thereby having more control on the keys distribution in each data node. LEEN keeps track of the frequencies of buffered keys hosted by each data node. In doing so, LEEN efficiently moves buffered intermediate keys to the destination considering the location of the high frequencies along with fair distribution of reducers’ inputs. To quan- tify the locality, data distribution and performance of LEEN, we conduct a comprehensive performance evaluation study using LEEN in Hadoop 0.21.0. OurexperimentalresultsdemonstratethatLEEN interestinglycanefficiently achieve higher locality, and balance data distribution after the shuffle phase. In addition, LEEN performs well across several metrics, with different parti- tioningskewdegrees,whichcontributetotheperformanceimprovementupto 45%. LEEN is generally applicable to other applications with data partitioning andthiswillresultinguaranteedresourceloadbalancingwithasmalloverhead duetotheasynchronousdesign.Themainfocusofthispaperandtheprimary usage for LEEN is on MapReduce applications where partitions skew exists (e.g., many scientific applications [12][13][14][16] and graph applications [17]). 4 ShadiIbrahimetal. We summarize the contributions of our paper as follows: – An in-depth study on the source of partitioning skew in MapReduce and its impacts on application performance. – Anaturalextensionofthedata-awareexecutionbythenativeMapReduce model to the reduce task. – A novel algorithm to explore the data locality and fairness distribution of intermediate data during and after the shuffle phase, to reduce network congestion and achieve acceptable data distribution fairness. – Practical insight and solution to the problems of network congestion and reduce computation skew, caused by the partitioning skew, in emerging Cloud. The rest of this paper is organized as follows. Section 2 briefly introduces MapReduce and Hadoop, and illustrates the recent partitioning strategy used inHadoop.Thepartitioningskewissueisexploredandempiricallyanalyzedin sections3.ThedesignandimplementationoftheLEEN approachisdiscussed in section 4. Section 5 details the performance evaluation. Section 6 discusses therelatedworks.Finally,weconcludethepaperandproposeourfuturework in section 7. 2 Background In this section, we briefly introduce the MapReduce model and its widely used implementation, Hadoop. Then we briefly zoom on the workflow of job execution in Hadoop introducing side by side the map, reduce and partition functions. 2.1 MapReduce Model The MapReduce [1] abstraction is inspired by the Map and Reduce functions, which are commonly used in functional languages such as Lisp. Users express the computation using two functions, map and reduce, which can be carried out on subsets of the data in a highly parallel manner. The runtime system is responsible for parallelizing and fault handling. The steps of the process are as follows: – The input is read (typically from a distributed file system) and broken up into key/value pairs. The key identifies the subset of data, and the value will have computation performed on it. The map function maps this data into sets of key/value pairs that can be distributed to different processors. – The pairs are partitioned into groups for processing, and are sorted ac- cording to their key as they arrive for reduction. The key/value pairs are reduced,onceforeachuniquekeyinthesortedlist,toproduceacombined result. HandlingPartitioningSkewinMapReduceusingLEEN 5 kkk123 Node1 Node2 kk45 Node3 k6 Imply hash function on keys in the k7 Maps output by appearance: k8 k9 hash(Hash code (Intermidiate Key) Module ReduceID) Keys ou tput in order of appearance Merge all the files in one indexed file Set of records RReecaodrd reprBelsoecnkt1 the Map FunctPioanr tition Function OBuuftfpeurte ids Sort and spill to disk Local (Mem) in Mem Disk HDFS Merge all the files Map in one indexed file Set of records reprBelsoecnkt2 the FunctionPartition Function OBuuftfpeurte ids Sort and sp ill to disk Local (Mem) in Mem Disk Fig. 1 The workflow of the two phases in MapReduce job: the map phase and reduce phase 2.2 Hadoop Hadoop [7] is a java open source implementation of MapReduce sponsored by Yahoo! The Hadoop project is a collection of various subprojects for re- liable, scalable distributed computing. The two fundamental subprojects are the Hadoop MapReduce framework and the HDFS. HDFS is a distributed file system that provides high throughput access to application data [7]. It is in- spired by the GFS. HDFS has master/slave architecture. The master server, called NameNode, splits files into blocks and distributes them across the clus- ter with replications for fault tolerance. It holds all metadata information about stored files. The HDFS slaves, the actual store of the data blockscalled DataNodes, serve read/write requests from clients and propagate replication tasks as directed by the NameNode. TheHadoopMapReduceisasoftwareframeworkfordistributedprocessing oflargedatasetsoncomputeclusters[7].ItrunsonthetopoftheHDFS.Thus dataprocessingiscollocatedwithdatastorage.Italsohasmaster/slavearchi- tecture. The master, called Job Tracker (JT), is responsible of : (a) Querying the NameNode for the block locations, (b) considering the information re- trieved by the NameNode, JT schedule the tasks on the slaves, called Task Trackers (TT), and (c) monitoring the success and failures of the tasks. 2.3 Zoom on job execution in Hadoop The MapReduce program is divided into two phases, map and reduce. For the map side, it starts by reading the records in the Map process, then the map function processes a data chunk into key/value pairs, on which the hash partitioningfunctionisperformedasshowninFig1.Thisintermediateresult, 6 ShadiIbrahimetal. Node1 Node2 Node3 Node1 Node2 Node3 Key 1 1 1 1 Key 1 4 1 1 Key 2 2 2 2 Key 2 1 4 1 Key 3 3 3 3 We assume balance Key 3 1 1 4 We assume balance map execution map execution Key 4 4 4 4 Key 4 3 2 1 Key 5 5 5 5 Key 5 1 3 2 Key 6 6 6 6 Key 6 2 1 3 21 21 21 12 12 12 Best Case Worst Case Best Case Worst Case (a) Keys’FrequenciesVariation (b) InconsistencyinKey’sDistribution Fig. 2 Motivational Example: demonstrates the worst and best partitioning scenarios when applying the current blindly key partitioning in MapReduce in the presence of Par- titioning skew. The keys are ordered by their appearance while each value represents the frequencyofthekeyinthedatanode. refereed as record, is stored with its associate partition in the buffer memory (100 MB for each map by default). If the buffered data reaches the buffer threshold(80%ofthetotalsize),theintermediatedatawillbesortedaccording to the partition number and then by key and spilled to the local disk as an indexfileandadatafile.Allfileswillbethenmergedasonefinalindexedfile- byindexedwemeanindexedaccordingtothepartitionnumberthatrepresents the target reduce. The reduce case is starting as soon as the intermediate indexed files are fetched to the local disk; the files from multiple local map outputswillbewrittenatthesametime(bydefaultfivepipeswillbeavailable for the different nodes). The files will be buffered in the memory in a “shuffle buffer”; when the shuffle buffer reaches a threshold the files will be redirected to the local disk, then the different files will be merged according to the user specific application, and merged files from the shuffle buffer will be tailed in the local disk. Finally the merged data will be passed to the reduce function and then the output will be written to the HDFS or elsewhere according to the user specific application. 3 Partitioning Skew in MapReduce The outputs of map tasks are distributed among reduce tasks via hash par- titioning. The default hash-partitioning, however, is designed to guarantee evenly distribution of keys amongst the different data nodes, that is, if we have n data nodes and k different keys then the number of keys which will be partitionedtoeachdatanodeis k,regardlessofthefrequenciesofeachdistinct n key (usually the number of records are associated with one key). The default hash-partitioningthereforeisonlyadequatewhenthenumberofrecordsasso- ciated with each key are relatively equal and the key’s records are uniformly distrusted amongst data nodes. HandlingPartitioningSkewinMapReduceusingLEEN 7 However, in the presence of partitioning skew the hash-partitioning as- sumption will break and therefore reduce-skew and network congestion can arise in practice[13][14][12] [18]. As we earlier stated the partition skew phe- nomena refereed to the case when the keys’ frequencies vary and/or the key’s records among data node are not uniformly distributed. Consider the two ex- amples which represent each factor separately: – Keys’ Frequencies Variation: Although the partitioning function perfectly distributes keys across reducers, some reducers may still be assigned more data simply because the key groups they are assigned to contain signifi- cantly more values. Fig 2-a presents the first example considering three data nodes and six keys. We vary keys frequencies to 3,6,9,12,15,and18 records per key, accordingly using the blindly hash-partitioning which is based on the sequence of the keys appearance during the map phase, the distribution of reducers’ inputs will vary between the best partitioning: 21 records for each reducer, and the worst case partitioning: the input of the reducers in node1, node2, and node3 will be 9, 21 and 33 records respec- tively.Despitethatinbothcasesthenumberofkeysassignedtoeachdata node is the same, two keys per node in our example. Accordingly reduce-skew will occur, in our example, node3 will finish its reduce nearly four times slower than node1; consequently, heavy reduce execution on some nodes. Thus performance experiences degradation (i.e. waiting the last subtask to be finished), and less resource utilization (i.e. node1 will be idle while node3 is overloaded). [18] and [12] have demon- strated the existence of this phoneme in some biological applications, for example, [12] has demonstrated that because of the keys’ frequencies vari- ation, in CloudBurst [18] application, some reducers will finish their task four times longer than other reduces. – Inconsistency in Key’s Distribution: As a second example, even when the keyshavethesamefrequenciesandthereforethepartitioningfunctionper- fectly distributes keys across reducers – all reducers inputs are relatively equal–.But,however,theblindhash-partitioningmayleadtohighnetwork congestion,especiallywhenthekey’srecodesarenotuniformlydistributed among data nodes. Fig 2-b presents the second example considering three data nodes and six keys. All keys have the same frequents, 6 records per key but the key’s distribution is inconsistent among the nodes. Applying the blindly hash-partitioning will result with evenly reducers’s inputs, but the data transfer, in contrast with the total map output during the shuf- fle phase will vary from 41.6%1, in the best case, to 83.3% in the worst case. Accordingly network congestion during the shuffle phase is strongly depending on the hash-partitioning. However, in the case of partitioning skew, when both factors, keys’ fre- quenciesvariation and inconsistency in key’sdistribution, will occur the blind hash-partitioningmayresultwillbothskew-reduceandnetworkcongestionas demonstrated in section 5. 1 Thisvaluerepresentstheratio= transferreddataduring shuffle mapphaseoutput 8 ShadiIbrahimetal. Table 1 MapReduceApplications’Classification MapReducewith MapReducewithoutCombiner MapOnly Combiner SingleRecord Multirecord Distributed Wordcount DistributedSort Wordcount without Com- Grep biner, Count of URL Graphprocessing[21], Access Fre- quency MachineLearning[20], Scientific application [18][16][19] 3.1 Partitioning Skew in MapReduce Applications MapReduce has been applied widely in various fields including data- and compute- intensive applications, machine learning, and multi-core program- ming. In this sub-section we intend to classify the MapReduce application in term of skewed intermediate data. A typical MapReduce application includes four main functions: map, re- duce,combinerandshufflefunctions.AccordinglywecouldclassifyMapRduce applications in respect to the main applied function in these applications into: map-oriented, combiner-oriented, map/reduce-oriented , shuffle-oriented as shown in table 1. – Map-oriented.Themapfunctionisthemainfunctionintheapplication, while the reduce function is only an identity function. An example of this type of applications is the Distributed Grep application2. – Combiner-oriented. The combiner function is applied in such applica- tions. The combiner performs as a map-based pre-reducer which signifi- cantly reduces the network congestion as in wordcount3 applications and Count of URL Access Frequency4. – Map/Reduce-oriented. These applications are typical map and reduce jobswherenocombinecanbeapplied.Alsointhistypeofapplications,all the keys is associated with only one unique value as in distributed Sort5. – Shuffle-oriented. In these applications both map and reduce functions are applied. However, they differ from the previous application in that multi record are associated with the same key and they differ from the secondtypeinthatnocombinercouldbeused.Herewhenthemapoutput is shuffled to the reducer, this may cause a network bottleneck. There is a wide range of applications in this category as graph processing, machine learning and scientific application [18][12][16][19][20]. It is important to note that many optimizations could be applied in this category. 2 http://wiki.apache.org/hadoop/Grep 3 http://wiki.apache.org/hadoop/WordCount 4 http://code.google.com/intl/fr/edu/parallel/mapreduce-tutorial.html 5 http://wiki.apache.org/hadoop/Sort HandlingPartitioningSkewinMapReduceusingLEEN 9 Keys Frequencies Variation 0 1. 8 0. 6 0. F D C 4 0. 2 0. 0 0. 1e+00 1e+02 1e+04 1e+06 1e+08 Key Frequency Fig. 3 Experiment Setup:CDFoftheKeys’frequencies.Thekeyfrequenciesvaryfrom 60to79860000recordsperkey. 3.2 Empirical Study on Partitioning Skew in Hadoop Inthissectionweempiricallydemonstratetheimpactsofthepartitionskewon MapReduceapplications.Forsimplicity,wemimicthefirsttypeofpartitioning skew, frequencies variation, which was in practise in some real applications. We use wordcount benchmark but after disabling the combiner function. 3.2.1 Experimental environment Our experimental hardware consists of a cluster with four nodes. Each node is equipped with four quad-core 2.33GHz Xeon processors, 24GB of memory and 1TB of disk, runs RHEL5 with kernel 2.6.22, and is connected with 1GB Ethernet. In order to extend our testbed, we use a virtualized environment, usingXen[22].Inthevirtualizedenvironment,onevirtualmachine(VM)was deployed on one physical machine (PM) to act as master node (Namenode). We also deployed two VMs on each of the three left PM, reaching a cluster sizeof6datanodes.Eachvirtualmachineisconfiguredwith1CPUand1GB memory. All results described in this paper are obtained using Hadoop-0.21.0. In order to show the case of partitioning skew, we perform the wordcount applications without combiner function. Moreover, we have used up to 100 differentkeysreachinganinputdatasizeof6GB:representingdifferentwords withthesamelength(toavoidvariationinvaluessize),withdifferentfrequen- cies as shown in Fig. 3 (we vary the keys frequencies between 60 to 79860000