DEUTSCH-FRANZSISCHE SOMMERUNIVERSITTFR NACHWUCHSWISSENSCHAFTLER 2011UNIVERSIT DT FRANCO-ALLEMANDEPOUR JEUNES CHERCHEURS 2011CLOUD COMPUTING :DFIS ET OPPORTUNITSCLOUD COMPUTING :HERAUSFORDERUNGEN UND MGLICHKEITEN17.7. 22.7. 2011Big Data Analytics beyond Map/ReduceProf. Dr. Volker MarklTUBerlin7/25/2011 DIMA TU Berlin 2Shift Happens! Our Digital World! Video courtesy of Michael Brodie, Chief Scientist, VerizonOriginal "Shift Happens" video by K. Fisch and S. McLeodOriginal focuses on shift in society, aimed at teachers educationMichael Brodie focuses on shift in/because of the digital world7/25/2011 DIMA TU Berlin 3Data Growth and Value About data growth: $600 to buy a disk drive that can store all of the worlds music 5 billion mobile phones in use in 2010 30 billion pieces of content shared on Facebook every month 40% projected growth in global data per year About the value of captured data: 250 billion Euro potential value to Europes public sector administration 60% potential increase in retailers operating margins possible withbig data 140,000-190,000 more deep analytical talent positions neededSource: Big Data: The next frontier for innovation, competition and productivity (McKinsey)7/25/2011 DIMA TU Berlin 4Big Data Data have swept into every industry and business function important factor of production exabytes of data stored by companies every year much of modern economic activity could not take place without that Big Data creates value in several ways provides transparency enables experimentation brings about customization and tailored products supports human decisions triggers new business models Use of Big Data will become a key basis of competition and growth companies failing to develop their analysis capabilities will fall behindSource: Big Data: The next frontier for innovation, competition and productivity (McKinsey)7/25/2011 DIMA TU Berlin 5Big Data Analytics Data volume keeps growing Data Warehouse sizes of about 1PB are not uncommon! Some businesses produce >1TB of new data per day! Scientific scenarios are even larger (e.g. LHC experiment results in ~15PB / yr) Some systems are required to support extreme throughput in transaction processing Especially financial institutes Analysis Queries become more and more complex Discovering statistical patterns is compute intensive May require multiple passes over the data Performance of single computing cores or single machines is not increasing substantially enough to cope with this development7/25/2011 DIMA TU Berlin 6Trends Massive parallelization Virtualization Service-based computing Web-scale datamanagement Analytics / BI Operational Multi-tenancyClaremont Report Re-architecting DBMS Parallelization Continuous optimization Tight integration Service-based everything Programming Model Combining structured andunstructured data Media ConvergenceTrends and Challenges7/25/2011 DIMA TU Berlin 7Overview Introduction Big Data Analytics Map/Reduce/Merge Introducing the Cloud Stratosphere (PACT and Nephele) Demo(Thomas Bodner, Matthias Ringwald) Mahout and Scalable Data Mining(Sebastian Schelter)7/25/2011 DIMA TU Berlin 8BIG DATA ANALYTICSMap/Reduce Revisited7/25/2011 DIMA TU Berlin 9Data Partitioning (I) Partitioning the data means creating a set of disjunct sub-sets Example: Sales data, every year gets its own partition For shared-nothing, data must be partitioned across nodes If it were replicated, it would effectively become a shared-disk with the local disks acting like a cache (must be kept coherent) Partitioning with certain characteristics has more advantages Some queries can be limited to operate on certain sets only, if it is provable that all relevant data (passing the predicates) is in that partition Partitions can be simply dropped as a whole (data is rolled out) when it is no longer needed (e.g. discard old sales)7/25/2011 DIMA TU Berlin 10Data Partitioning (II) How to partition the data into disjoint sets? Round robin: Each set gets a tuple in a round, all sets have guaranteed equal amount of tuples, no apparent relationship between tuples in one set. Hash Partitioned: Define a set of partitioning columns. Generate a hash value over those columns to decide the target set. All tuples with equal values in the partitioning columns are in the same set. Range Partitioned: Define a set of partitioning columns and split the domain of those columns into ranges. The range determines the target set. All tuples on one set are in the same range.7/25/2011 DIMA TU Berlin 11 The data model key/value pairs e.g. (int, string) Functional programming model with 2ndorder functions map: input key-value pairs: output key-value pairs: reduce: input keyand a list of values output key and a single value The framework accepts a list outputs result pairsMap/Reduce Revisited7/25/2011 DIMA TU Berlin 12Data Flow in Map/Reduce(Km,Vm)*(Km,Vm) (Km,Vm) (Km,Vm)MAP(Km,Vm) MAP(Km,Vm) MAP(Km,Vm)(Kr ,Vr)* (Kr ,Vr)* (Kr ,Vr)*REDUCE(Kr ,Vr*) REDUCE(Kr ,Vr*) REDUCE(Kr ,Vr*)(Kr ,Vr*) (Kr ,Vr*) (Kr ,Vr*)(Kr ,Vr) (Kr ,Vr) (Kr ,Vr)(Kr ,Vr)*FrameworkFrameworkFramework7/25/2011 DIMA TU Berlin 13 Problem: Counting words in a parallel fashion How many times different words appear in a set of files juliet.txt: Romeo, Romeo, wherefore art thou Romeo? benvolio.txt: What, art thou hurt? Expected output: Romeo (3), art (2), thou (2), art (2), hurt (1), wherefore (1), what (1) Solution: Map-Reduce Jobmap(filename, line) {foreach (word in line)emit(word, 1);}reduce(word, numbers) {int sum = 0;foreach (value in numbers) {sum += value;}emit(word, sum);}Map Reduce Illustrated (1)7/25/2011 DIMA TU Berlin 14Map Reduce Illustrated (2)7/25/2011 DIMA TU Berlin 15Data Analytics: Relational Algebra Base Operators selection (o) projection (t) set/bag union () set/bag difference (\ or -) Cartesian product () Derived Operators join () set/bag intersection () division (/) Further Operators de-duplication generalized projection(grouping and aggregation) outer-joins und semi-joins Sort7/25/2011 DIMA TU Berlin 16 Selection / projection / aggregation SQL Query:SELECT year, SUM(price)FROM salesWHERE area_code = USGROUP BY year Map/Reduce job:map(key, tuple) {int year = YEAR(tuple.date);if (tuple.area_code = US)emit(year, {year => year, price => tuple.price });}reduce(key, tuples) {double sum_price = 0;foreach (tuple in tuples) {sum_price += tuple.price;}emit(key, sum_price);}Relational Operators as Map/Reduce jobs7/25/2011 DIMA TU Berlin 17 Sorting SQL Query:SELECT * FROM sales ORDER BY year Map/Reduce job:map(key, tuple) {emit(YEAR(tuple.date) DIV 10, tuple);}reduce(key, tuples) {emit(key, sort(tuples));}Relational Operators as Map/Reduce jobs7/25/2011 DIMA TU Berlin 18 UNION SQL Query:SELECT phone_number FROM employees UNIONSELECT phone_number FROM bosses Map/Reduce job needs two different mappers:map(key, employees_phonebook_entry) {emit(employees_phonebook_entry.number, ``);}map(key, bosses_phonebook_entry) {emit(bosses_phonebook_entry.number, ``);}reduce(phone_number, tuples) {emit(phone_number, ``);}Relational Operators as Map/Reduce jobs7/25/2011 DIMA TU Berlin 19 INTERSECT SQL Query:SELECT first_name FROM employees INTERSECTSELECT first_name FROM bosses Map/Reduce job needs two different mappers:map(key, employee_listing_entry) {emit(employee_listing_entry.first_name, `E`);}map(key, boss_listing_entry) {emit(bosses_listing_entry.first_name, `B`);}reduce(first_name, markers) {if (`E` in markers and `B` in markers) {emit(first_name, ``);}}Relational Operators as Map/Reduce jobs7/25/2011 DIMA TU Berlin 20 Benchmark to test the performance of distributed systems Goal: Sort one Petabyte of 100 byte numbers Implementation in Hadoop: Range-Partitioner that splits the data in equal ranges (one for each participating node) Sort is basically "Range partitioning sort" as described earlierThe Petabyte Sort Benchmark7/25/2011 DIMA TU Berlin 21Petabyte sorting benchmark Per node: 2 quad core Xeons @ 2.5ghz, 4 SATA disks,8G RAM (upgraded to 16GB before petabyte sort), 1 Gigabit Ethernet. Per Rack: 40 nodes, 8 gigabit Ethernet uplinks.7/25/2011 DIMA TU Berlin 22Cluster Utilization during Sort7/25/2011 DIMA TU Berlin 23JOINS IN MAP/REDUCEMap/Reduce Revisited7/25/2011 DIMA TU Berlin 24Symmetric Fragment-and-Replicate Join (II)Nodes in the Cluster7/25/2011 DIMA TU Berlin 25 We can do better, if relation S is much smaller than R. Idea: Reuse the existing partitioning of R and replicate the whole relation S to each node. Cost: p * B(S) transport??? local joinAsymmetric Fragment-and-replicate Join is a special case of the Symmetric Algorithm with m=p and n=1. The Asymmetric Fragment-and-replicate join is also called Broadcast JoinAsymmetric Fragment-and-Replicate Join7/25/2011 DIMA TU Berlin 26 Equi-Join: L(A,X)R(X,C) assumption: |L|