SQL versus MapReduce

SQL versus MapReduceA Comparison between two Approaches to Large Scale Data AnalysisLivio HobiUniversity of Zurichlivio.hobi@uzh.chMay30,2013AbstractThis paper compares two different approaches to large scale data anal-ysis,namely MapReduce and parallel database management systems.Bothapproaches use a cluster of nodes to compute expensive tasks in parallel.The strengths of MapReduce are fault tolerance,storage-system indepen-dence,flexibility and simplicity.While MapReduce can process the dataon-the-fly by simply loading it into a distributedfile system,the load-ing phase of parallel database management systems can take a long time.Once the data is loaded,parallel database management systems are ro-bust,support a high degree of parallelism and provide high-performance,since they have been developed and improved for over two decades.Theuser has to think about the fundamental trade-offbetween data loadingcosts and task response time.If the dataset is static rather than dynamicand the data has to be queried frequently,a parallel DBMS may be prefer-able.In contrast,if the data is dynamic and the user is interested in a fewqueries,the approach of MapReduce can be very useful.Another interest-ing difference is the failure model.While MapReduce makes extensivelyuse of afine grained failure model,parallel DBMSs prefer performanceat the cost of possible re-executions of the whole task.If the computa-tion takes a long time or the cluster consists of hundreds or thousands ofmachines,the approach of MapReduce may be preferable.1IntroductionAs datasets become larger and larger,there emerges a need to handle such data in parallel over a cluster of machines.Since over two decades parallel database management systems(DBMSs)have existed.They use a cluster of interconnected computing machines to perform queries over large datasets in parallel[PPR09].A DBMS is installed on each node,which is connected to the other members of the cluster.Once the system has to process a query,it automatically computes a query plan,distributes it and partially executes the query in parallel over the cluster.Parallel DBMSs are robust,support a high1degree of parallelism and provide high-performance.They use the declarative, high-level query language SQL(Structured Query Language).Parallel DBMS perform well because they load and index the data before they process queries.Since this loading phase can take a long time,there arises the question if this loading phase is worth it to only process one or two queries over the data.This is where MapReduce comes into play.MapReduce is a programming model for processing and generating large datasets.MapReduce was introduced in2004by Dean and Ghemawat[DG08]and embraced by a ever growing community.Since the launch of MapReduce,users have developed thousands of MapReduce programs.So what makes this programming model so successful?The main benefit of the MapReduce paradigm is simplicity.MapReduce consists basically only of two functions:Map and Reduce.These functions have to be defined by the user.They take as input a key/value pair and output a key/value pair.Hidden from the user,the MapReduce implementation handles parallelism,failures of nodes,data distribution and load balancing.In contrast to the loading and indexing phase of parallel DBMSs,MapReduce only loads the data in a distributedfile system and then processes the data on-the-fly. MapReduce divides the dataset into smaller chunks of data and distributes them among to the cluster.While parallel DBMS make use of the declarative language SQL,MapReduce tasks can be programmed using procedural,object-oriented languages,such as Java.It is possible to write almost every parallel database task as an equivalent MapReduce task and vice-versa(using user defined functions(UDFs)in DBMSs) [PPR09].An interesting question is whether MapReduce should and could replace parallel DBMSs.What are the benefits of each approach?How do they perform?What can they learn from each other?2Problem StatementThis section introduces a problem from the banking industry that will be solved with both techniques,parallel DBMS and MapReduce.The following sections will refer to this problem to show how the two approaches work and where their differences are.Let us consider we are a bank(e.g.UBS).As in most companies,one problem is to know our customers.Which customers should be involved in which marketing campaign?Where are potential customers?Where are their interests?In this example,our goal is tofind customers which have a capital of over one million and are additionally interested in share transactions.The data we have are information about the capital of a customer and a log of clicks of our online banking system.We assume that a customer is interested in shares,when he or she clicks more than ten times in a month on the according”Shares”-button in the online banking system.The difficulty in this scenario is that the log of clicks in our2online banking system grows very fast and changes over time.Therefore we have to deal with a big amount of data.3The two ApproachesIn this section the details of the two approaches are presented.3.1MapReduceMapReduce is a programming model,which allows straightforward implemen-tation of parallel tasks,but hides the details of parallelism,fault tolerance,data distribution and load balancing[DG08].In practice there are a broad range of applications for MapReduce.Examples are graph processing,text processing, data mining,reports of popular queries(Google Trends)and processing of data from web crawls[SAD+10].3.1.1Programming ModelAs mentioned before,MapReduce is a programming model developed for pro-cessing parallel tasks over a large dataset with a cluster of nodes.The two core functions,Map and Reduce,take as input a key/value pairs and produce as output a list of key/value pairs.The Reduce function accepts intermediate key/value pairs,provided by the Map function,performs some computation (such as merging,aggregation)and outputs again key/value pairs.In MapReduce,every intermediate key/value pair with the same key is sent to the same Reduce instance on the same node.An instance means a unique running invocation of a Map or Reduce function,where a node of a MapReduce cluster usually has multiple instances of these functions[PPR09].It is possible to perform complex computations with the MapReduce pro-gramming model by simply chaining multiple Map and Reduce functions.While in SQL,respectively in parallel DBMS,the user has to write a SQL query(what can be hard if the query is complex),MapReduce follows a simple and more un-derstandable step-by-step manner.3.1.2ImplementationIn clusters with hundreds,or even thousands of nodes,failures of some nodes can be very common.Therefore MapReduce uses replication to provide availability and reliability[DG08].Let us look at a simple example:In a cluster of1000machines,where the probability of a failure of a machine is0.001per active hour,and a average task computation time of10hours,the probability of a failure is10%(1000*0.001*10).If this task is running every night and the result is needed the next day, reliability of a programming model can be really useful.Of course,1000nodes and10hour processing are quite big numbers and maybe not often used by3a company nowadays,but with ever-growing data,this scenario may become more realistic.A MapReduce cluster is comprised of one master node and a set of worker nodes.The master node keeps track of the progress of the execution and in case of a failure it re-executes the specific task on another node.A typical execution contains the following steps:1.The inputfiles are split into chunks and distributed among the nodes overthe distributedfile system.Then,copies of the program are started on the cluster.2.The master node picks idle workers and assigns them Map and Reducetasks.3.Workers with Map tasks read the input chunk,parse the data to key/valuepairs and pass them to the Map function.Intermediate key/value pairs are buffered in memory.4.The buffered pairs are written to local disk and the master gets informedabout their location.5.The master informs Reduce workers about the location of the intermediatevalues.The Reduce workers then use remote procedure calls to read the data,where the keys of a Reduce instance are all the same.6.The Reduce worker iterates over the intermediate data and passes thecorresponding data to the Reduce function for each unique key.This generates afinal outputfile for every Reduce instance,which is again stored in the distributedfile system.7.Afterfinishing all tasks,the master wakes up the user program.A MapReduce implementation typically stores one outputfile per Reduce task,since it is normally not necessary to combine the output.The output is often used for a next MapReduce task or the next application can deal with it.For each assigned task,the master node keeps track of the identity of the worker machine and the corresponding state(idle,in-progress,completed).Ad-ditionally it stores the location of the R intermediatefile regions per Map task, where R is the number of unique keys produced by a Map worker.3.1.3Fault ToleranceOne of the key benefits of the MapReduce programming model is its fault toler-ance.It uses replications of data chunks to provide availability and reliability. The master pings workers and in case of no response,the worker is marked as failed.All running Map and Reduce tasks of that machine are then assigned to other pleted Map tasks too,because the intermediate results are stored on local disks,whereas completed Reduce tasks do not have to be re-executed because they are stored in the distributedfile system.Of course there4are some trade-offs between fault tolerance overhead and performance.As data gets larger and hence processing tasks takes more time,failures of nodes in big cluster are more likely and therefore an expensive fault tolerance model can be worth it.We will discuss this aspect later on in section4.3.1.4Solution with MapReduceLet us come back to our problem from section2.Our data is stored in comma-separated values(CSV)files.In thefile of the customers(Customers.csv)every row stores the id,the name and the capital.The log of clicks is stored in one log file per day.As usual in logfiles,the date information is stored in thefilename. Afile of our log could be named”2013-04-02-ClickLog.csv”.Every row stores the customer id and the click event.Examplefiles:2013-04-02-ClickLog.csv Customer.csv CustomerID,ClickEvent ID,Name,Capital3,Shares1,Hans,1’750’0002,Shares2,Peter,25’0002,Home3,Julia,3’500’0003,Payment1,Transfer2,Shares2,Shares1,SharesThe code in MapReduce would look as follows:Map1(key,value){//key:filename of Customer//value:Object,with the entries of a CustomerCustomer c=new Customer()c.parseFrom(value)if(c.capital>1’000’000)Emit(c.id,)}Map2(key,value){//key:filename of ClickLog//value:Object,with the entries of a ClickLogClickLog c=new ClickLog()c.parseFrom(value)if(c.clickEvent=Shares)Emit(c.customerID,c.event)}5Reduce(key,values){//key:CustomerID//value:String shares,String customer;int count=0;if(a value in values!=Shares){customer=that valueFor all other value in valuesCount++}If(count>10AND customer!=null){Emit(key,customer)}}There are two Map functions and one Reduce function.Thefirst Map func-tion takes as input the data from”Customers.csv”and only emits the customer id as key and the customer name as value when the capital of that customer is over one million.This is equivalent to afilter operation in the where clause of a SQL statement.The second Map function processes the data from the log of clicks and emits the customer id as key and the event as shares when the click event is an event of type”Shares”.Intermediate key/value pairs with the same customer id are sent to the same Reduce instance.The Reduce function takes as input the intermediate key/value pairs from both Map functions.It then counts the number of”Shares”events and looks for a customer name.If the variable customer is still null in the end, it means that this customer does not have a capital of more than a million and therefore the Reduce function does not emit a result.After the computation,there will be afile in the distributedfile system for every customer which has more than a million capital and over ten clicks on ”Shares”with its customer id and the customer name.This is exactly what we were looking for.In MapReduce we can make use of the time information stored in thefile-name in order to not scan through all the ClickLogfiles.This reduces the amount of work dramatically(less parsing,reading,processing overhead).Figure1shows an example execution of this MapReduce task.In thefirst column are the input data chunks,stored in the distributedfile system.In this example,every chunk consists of three entries of the Customer data or the ClickLog data.This chunks are assigned to the different Map instances and their output is then stored on local disks on the nodes of the Map instances. Afterwards,the master node informs the Reduce instances where their data is located.They use remote procedure calls to get their intermediate key/value pairs.The output of the Reduce instances is then stored in the distributedfile system.To save space,the example shows a result of the Reduce function if there is at least one customer and one”Shares”event.6Figure1:MapReduce Execution Example3.2Parallel Database Management SystemsA parallel DBMS uses a cluster of nodes with high-speed interconnects to com-pute expensive tasks in parallel.They support standard relational tables and ing horizontal partitioning of relational tables,the system can partially execute SQL queries in parallel.Horizontal partitioning means that the rows of a relational table are distributed across the nodes of the cluster[SAD+10]. Parallel DBMSs use different partitioning ing hash partitioning, a hash function is applied to every row of a relation and according to the value distributed to the nodes.In round-robin partitioning,the rows of a relation are distributed in blocks across the nodes of the cluster.The advantage of round-robin partitioning is better load balancing,since every node stores more or less the same number of records.The system uses an optimizer that translates the query into a query plan, which is then distributed across the nodes of the cluster.In contrast to the approach of MapReduce,there are no additional messages or scheduling al-gorithms necessary,because the optimizer sends the whole query plan at the beginning of the computation and therefore every node knows exactly what and when to compute.A key benefit of parallel DBMSs is the automatic generation of the query plan by the optimizer,which takes into account how the data is partitioned.The user doesn’t have to think about how it is executed.There are many commercial implementations available,including Teradata, Netezza,DataAllegro,ParAccel,Greenplum,Vertica and DB2[SAD+10].73.2.1Solution with parallel DBMSsThis subsection describes how to solve the problem from section2with parallel DBMSs.Our data is stored in two relations,Customer and ClickLog.Customer consists of three attributes,the ID,Name and the Capital.The attributes of ClickLog are CustomerID,Date and ClickEvent.The two example relations could look as follows:ClickLog CustomerCustomerID TimeStamp ClickEvent ID Name Capital 32013-02-15Shares1Hans1’750’000 22013-03-07Shares2Peter25’000 22013-04-02Home3Julia3’500’000 32013-04-08Payment12013-04-17Transfer22013-04-20Shares22013-04-25Shares12013-04-26SharesThis SQL statement solves our problem:SELECT Customer.ID,FROM Customer,ClickLogWHERE Customer.ID=ClickLog.CustomerIDAND Customer.Capital>1000000AND ClickLog.ClickEvent=SharesAND ClickLog.Date BETWEEN’2013-04-01’AND’2013-04-30’GROUP BY Customer.ID,HAVING COUNT(ClickLog.ClickEvent)>10This SQL query basically joins the two relations on the customer id and filters them on the amount of capital,type of event and the date.It then groups by the id and the name and returns the results which have a count of the event type higher than ten.Let us now look at three different execution scenarios:Scenario one:Both relations are hash partitioned on the customer id.If the two relations are hash partitioned on the join attribute,the query can be processed in parallel over the cluster because all the data of a single join attribute are stored on the same node.Therefore the local joins can be processed in parallel and network traffic is only needed at the beginning,when the query plan is distributed and in the end to send all the results to a single node.8Scenario two:Customer and ClickLog are round-robin partitioned. If both relations are round-robin partitioned,the optimizer can chose between different query plans.If one of the two relations is small(e.g.the Customer relation),it could be sent to all the nodes and every node could then compute their partial results in parallel.If the overhead of sending the whole relation across the cluster is too big,a hash function can be applied to the join attribute of the query(ID,CustomerID)to compute where to send the rows.All the records of specific id are then sent to the same node such that the query can be executed in parallel.Scenario three:ClickLog is hash partitioned on CustomerID and Cus-tomer is round-robin partitioned.In this scenario the optimizer would apply the same hash function used to partition ClickLog to Customer and then distribute the records of Customer to the same node,where the ClickLog entries are located.As mentioned before,the optimizer automatically computes the best query plan,depending on how the data is partitioned.4Some Key Aspects in DetailIn this section some differences between the two approaches are discussed in detail.4.1SchemaMapReduce is schema-free.On one hand this makes the programming model veryflexible but on the other hand the user has to think about how the data is parsed into the Map,respectively Reduce functions.Pavlo et al.stated in their paper that the user often has to write a custom parser to add the required semantics to the data used by the program.This can lead to a significant overhead if the parser is not optimized.For this purpose there exists a Protocol Buffer format,which describes the input and output types,but hides the details of encoding and decoding.It uses an optimized binary representation that is more compact and much faster than the textual formats[DG10].In our example,we can use the input like a normal object,because we provided a protocol buffer description,where one,two and three refer to the position in the input value,separated by a comma:message Customer{required int32ID=1;required string Name=2;required int32Capital=3;}9In contrast,parallel DBMSs use the schema known from the relational paradigm. They separate the schema from the application.Furthermore,parallel DBMSs automatically ensure that the data does not violate integrity or other con-straints,while in MapReduce this is the task of the programmer[PPR09].In MapReduce,this can lead to a problem,when data is shared between programmers.They have to agree on a schema and must ensure that added or modified data is handled correctly by the application.If the data changes, it is possible that all programs have to be changed,which can lead to higher maintenance costs.But if no data is shared,the programmer can make use of theflexibility of the MapReduce paradigm.4.2IndexingParallel DBMSs make extensive use of indices to improve their performance. In contrast,MapReduce does not provide built-in indices and one could think that MapReduce always has to scan through the entire data[PPR09].How-ever,MapReduce is storage-system independent.This means that it could for example use the output of a query in a DBMS as an input of a MapReduce task.Another workaround is to make use of information stored infilenames. Time information for example are often stored in the names of log datafiles. MapReduce can then process data from a specific time period[DG10].This is exactly what we have done to solve our example problem.Of course,this workarounds of MapReduce do not fully compensate for the lack of indices.But the storage-system independence makes it reallyflexible. MapReduce was not built to make extensive use of index structures.This is one of the key differences.Parallel DBMSs use these techniques to improve the performance,but this leads to a longer loading phase.The advantage of MapReduce is the ability to process tasks on-the-fly.4.3Programming LanguageThe programming model of the two approaches are very different.While parallel DBMSs use the high-level,declarative language SQL,MapReduce uses proce-dural languages,such as Java or C++.In SQL,the user has to state what he wants,while the implementation is hidden.On the contrary,in procedural languages the programmer has to implement the algorithm by him self[PPR09]. There are approaches that build a layer on top of MapReduce frameworks and provide some high-level functionalities,such as joins or ing this layer,frequently utilized algorithms do not always have to be implemented by the programmer.Apache Pig[ORS08]is an example.4.4Complex FunctionsUsing MapReduce,the programmer is enjoyingflexibility in how a problem is addressed.In contrast,most of the parallel DBMSs come with afixed set of functionalities,where the usage of user defined functions is limited.Even10though several problems can be solved with these standard functionalities,there are cases where this does not hold.For example processing unstructured input data that does no properlyfit into the relational paradigm(e.g.data-mining or data-clustering,where multiple passes over the same data are required)[PPR09]. In contrast,the programmer doesn’t have to care about the execution order because the optimizer automatically computes a query plan.4.5Execution Strategy and Fault ToleranceThe execution strategy of MapReduce is based on a pull model.Every Map function materializes the output on local disk.Then the Reduce instance gets informed by the master on which node and where on the disk its data is located. The advantage of this materializing is fault tolerance.As mentioned in section 3.1,MapReduce is able to re-execute every failed task.On the other side this can lead to a bottleneck,since a lot of Reduce instances could try to get data from the same node at the same time.Dean and Ghemawat state that as datasets grow and computations take more time,fault tolerance will become more important.In contrast,parallel DBMSs use a push model rather than a pull model. Intermediate results are not stored on local disks,but they are forwarded to the next execution step.This ends up with a much simpler failure model(e.g. transactions),where a failure would result in a re-execution of most parts of a computation.Nevertheless,parallel DBMSs gain performance because they do not store intermediate results[PPR09].4.6PerformancePavlo et al.tried to compare the performance of a MapReduce implementation (Hadoop[had])to two parallel DBMSs(DBMS-X and Vertica[ver]).In thefive tasks they tested,on average the two parallel DBMSs outperform Hadoop by a factor of3.2to7.4on a cluster of100nodes.Reasons for the better perfor-mance of the parallel DBMSs are the usage of indices,better compression and storage techniques and mature algorithms for parallel processing.Additionally the high start-up costs of a Hadoop jop can dominate the response time of a task,especially in short-running queries.They were however impressed,how easy a Hadoop environment is installed and how fast a user can start running a job.Dean and Ghemawat criticized different aspects on the performance com-parison.First of all they stated that for many of the tasks,the loading phase of the parallel DBMSs takes aboutfive to50times the time needed to run the job on Hadoop.If the data is processed only once or twice,a Hadoop job would clearly end before the data is even loaded into the parallel DBMSs.Further-more,they claimed that start-up and parsing overhead are not fundamental differences between two programming models.These two aspects are imple-mentation specific and can be addressed in various ways(e.g.Protocol Buffer Format,pools of worker waiting for a new job).Additionally they falsify the11conclusion of[PPR09]that a MapReduce task always has to scan over all the data.MapReduce can use the output of a DBMS as an input or use natural indices such as information stored infilenames in order to not scan through all the data.Even though the performance benchmark of Pavlo et al.may overstate the benefit of parallel DBMSs,it clearly shows the trade-offbetween data loading in parallel DBMSs and processing data on-the-fly in MapReduce.As MapReduce implementations will get more sophisticated,the probable actual performance advantage may shrink and MapReduce may become an interesting alternative to parallel DBMSs.5ConclusionThis paper simplifies and compares the two different approaches,MapReduce and parallel DBMSs.While MapReduce is excellent in the areas of simplicity, fault tolerance andflexibility,there can be still a lack of performance compared to parallel DBMSs,once the data is loaded.MapReduce consists basically of two functions,Map and Reduce.They have to be defined by the program-mer.MapReduce supports procedural programming languages,such as Java and C++.A speciality of MapReduce is to start computations over data on-the-fly.The data has only to be copied in the distributedfile system and then the processing can start.The possibility of starting a task over a huge dataset without a long loading phase is one of the key features.If the data is only processed once and changes frequently over time,this approach can be very useful and preferable.Its fault tolerance model brings an important advantage, as datasets and clusters get bigger.But this fault tolerance is not for free.The materializing of intermediate results on local disks can have a bad impact on the performance.Nevertheless,such a programming model with this fault tolerance will become more preferable as data sets grow.On the other hand,parallel DBMSs have been developed for over two decades and are thus very mature.If the data is static,rather than dynamic,and the same data has to be processed multiple times,their overall performance is quite ing the well-known declarative language SQL,the user can state what he wants,while the parallelism,optimization and execution of the query is hidden.Nevertheless,parallel DBMSs reach their limitations when the input data does not properlyfit into the relational paradigm(unstructured data such as plain text or content of a html page).Both approaches have their advantages and disadvantages.It is up to the user to decide which solution is when useful for a specific problem. References[DG08]Jeffrey Dean and Sanjay Ghemawat.MapReduce:simplified data processing on large munications of the ACM,1251(1):107–113,2008.[DG10]Jeffrey Dean and Sanjay Ghemawat.MapReduce:aflexible data processing munications of the ACM,2010.[had]Hadoop./.[ORS08]Christopher Olston,Benjamin Reed,and U Srivastava.Pig latin:a not-so-foreign language for data processing....on Management ofdata,2008.[PPR09]Andrew Pavlo,Erik Paulson,and Alexander Rasin.A comparison of approaches to large-scale data analysis....on Management of data,pages165–178,2009.[SAD+10]Michael Stonebraker,Daniel Abadi,J.Dewitt DeWitt,Sam Madden, Erik Paulson,Andrew Pavlo,and Alexander Rasin.MapReduce andparallel DBMSs:friends or foes?Communications of the ACM,2010.[ver]Vertica./.13。