MapReduce: Simplified Data Processing on Large Clusters PA154 Jazykové modelování (9.2) Pavel Rychlý pary@fi.muni.cz April 27, 2021 Source: Jeff Dean, Sanjay Ghemawat Google, Inc. December, 2004 https://research. goog le/pu bs/pu b62/ PA154 Jazykové modelování (9.2) MapReduce 1/32 Motivation: Large Scale Data Processing Many tasks: Process lots of data to produce other data Want to use hundreds or thousands of CPUs ■ ... but this needs to be easy MapReduce provides: ■ Automatic parallelization and distribution ■ Fault-tolerance I/O scheduling Status and monitoring PA154 Jazykové modelování (9.2) MapReduce 2/32 Programming model Input & Output: each a set of key/value pairs Programmer specifies two functions: map (in_key, in_value) -> Iist (out-key , intermediate_value Processes input key/value pair Produces set of intermediate pairs reduce (out-key, I ist (interme .value)) -> list (out_value) ■ Combines all intermediate values for a particular key ■ Produces a set of merged output values (usually just one) Inspired by similar primitives in LISP and other languages PA154 Jazykové modelování (9.2) MapReduce 3/32 Example: Count word occurrences map(String input-key, String input-value): // input-key: document name // input_value: document contents for each word w in input-value: Emitlntermediate (w, "1"); reduce (String output-key , Iterator inter mediate .values ) // output-key: a word // output-values: a list of counts int result = 0; for each v in intermediate-values: result += Parselnt(v); Emit (As St ring (resu It )); Pseudocode: See appendix in paper for real code PA154 Jazykové modelování (9.2) MapReduce Model is Widely Applicable MapReduce Programs In Google Source Tree Example uses: distributed grep distributed sort web link-graph reversal term-vector per host web access log stats inverted index construction document clustering machine learning statistical machine translation PA154 Jazykové modelování (9.2) MapReduce 5/32 Implementation Overview Typical cluster: ■ 100s/1000s of 2-CPU x86 machines, 2-4 GB of memory ■ Limited bisection bandwidth ■ Storage is on local IDE disks ■ GFS: distributed file system manages data (SOSP'03) ■ Job scheduling system: jobs made up of tasks, scheduler assigns tasks to machines Implementation is a C++ library linked into user programs PA154 Jazykové modelování (9.2) MapReduce 6/32 Execution Input Intermediate Group by Key Grouped k 1 :v,v,v3v k2:v k3:v,v k4:v,v,v k5:v ».©© ©>© Output kl :v kl:v k2:v kl:v k3:v k4:v k4:v k5:v k4:v kl :v k3:v PA154 Jazykové modelování (9.2) MapReduce 7/32 Parallel Execution Sort and Group k2;v k4 v v v k5 ;v Reduce Task 1 Map Task 1 t r Map Task 2 | r Map Task 3 1 i 1 i <£) (i) ; ; (i) ; ! kl ;v kl ;v k2 :v kl:v i i lc3;v k4:v k4:v k5;v i i k4:v ki:v k3:v Partitioning Function ■ i Partition in I Function i Partitioning Function Sort and Group kl ;V|V|V,v k3:v.v Reduce Task 2 PA154 Jazykové modelování (9.2) MapReduce 8/32 Task Granularity And Pipelining Fine granularity tasks: many more map tasks than machines ■ Minimizes time for fault recovery ■ Can pipeline shuffling with map execution ■ Better dynamic load balancing Often use 200,000 map/5000 reduce tasks/ 2000 machines Process Time---------- - > User Program MapRediice() ... wait... Master Assign tasks to worker machines... Worker 1 Map 1 Map 3 Worker 2 Map 2 Worker 3 Read 1.1 Read 1.3 Read 1.2 Reduce 1 Worker 4 Read 2.1 Read 2.2 Read 2.3 Reduce 2 PA154 Jazykové modelování (9.2) MapReduce 9/32 MapReduce status: MR_Indexer-beta6-large-2003_10_28_00_03 Started: Fri Nov 7 09:51:07 2003 -- up 0 hr 00 min 18 sec 323 workers; 0 deaths Counters Type Shards Done Active Input(MB) Done (ME) Output(ME) Map 13853 0 323 878934.6 1314.4 717.0 Shuffle 500 0 323 717.0 0.0 0.0 Reduce 500 0 0 0.0 0.0 0.0 100 - 90 30 U 70 *J 9> Q. 60 E O O 50 -p c o 40 u [_ O 0- 30 20 10 4 Reduce Shard Variable Mimite Mapped (MB/s) 72.5 Shuffle (MB/s) 0.0 Output (MB/s) 0.0 doc-index-hits 145825686 docs-índexed 506631 dups-in-index-merge 0 mr-operator-calis 508192 mr-operator- 506631 PA154 Jazykové modelování (9.2) MapReduce 10/32 MapReduce status: MR Indexer-beta6-large-2003_10_28 00 03 Started: FnNov7 09:51:07 2003 - up 0 hr05 min 07 sec 1707 workers, 1 deaths Counters Type Shards Done Active Input(MB) Done(M£) Output(MB) Map 13S53 1857 1707 878934.6 191995.8 113936.6 Shuffle 500 0 500 113936.6 571137 571137 Reduce 500 0 0 57113.7 0.0 0.0 100 90 BO ■D O 70 -U 9) -H Q. &< Ľ O U 50 ■P C ií 40 o t- u Q- 30 10 0 Reduce 5hard Variable Minute Mapped CMB/s) 699.1 Shuffle (MB/s) 349.5 Output (MB/s) 0.0 doc-index-hits 5004411944 docs-indexed 17290135 dups-in-ind ex-merge 0 inr-operator-i: alls 17331371 mr-operator-OUtDUtS 17290135 PA154 Jazykové modelování (9.2) MapReduce 11/32 MapReduce status: MRIndexer-betaó-large-2003 10 28 00 03 Started: Fťi Nôv 7 09:51:07 2003 - up 0 hr 10 min 18 sec 1707 workers; 1 deaths Type Shards Done Active Input(MB) Done(MB) Output(MB) Map 13853 5354 1707 873934 6 406020.1 241058.2 Shuffle 500 0 500 241058.2 196362.5 196362.5 Reduce 500 0 0 196362.5 0,0 0.0 1PP 90 30 "S 70 c z — = ľ u L V 60 o- o : j o Tľ Reduce Shard Counters ■z IT. Variable Minute Mapped (MEJs) 704.4 Shuffle (MB^s) 371.9 Output 0.0 (MB/s) doc-índex-hits 5000364228 docs-índexed 17300709 dups-in- 0 operator- 17342493 mr- operator- 17300709 outputs PA154 Jazykové modelování (9.2) MapReduce 12/32 84^45454828^82^7528080 MapReduce status: MR_Indexer-beta6-large-2003_10_2S 00 03 Started: Fii Nov 7 09:51:07 2003 - up 0 hr 15 mín 31 sec 1707 workers; 1 deaths Counters Type Shards Done Active Input (MB) D one (MB) Output(MB) Map 13353 3841 1707 873934.6 621603.5 369459.8 Shuffle 500 0 500 369459.8 326986 8 326986.8 Reduce 500 0 0 326986,8 0.0 0.0 (M Reduce Shard Variable Minute Mapped CMB/s) 706.5 Shuffle (MB/s) 419.2 Output (MB/s) 0.0 doc-index-hits 4982370667 docs-índexed 17229926 dups-in-index-merge 0 I mr-operator-calls 17272056 mr-operator-outouts 17229926 PA154 Jazykové modelování (9.2) MapReduce 13/32 MapReduce status: MRIndexer-betaó-large-2003 10 28 00 03 up 0 hr 29 min 45 sec Started: FnNov7 09:51:07 2003 1707 workers: 1 deaths Type Shards Done Active Input D o so ■p v 40 £ 30 Zu 10 o ■z vi o -z-■z- <: j en Reduce Shard o c -í- Counters o o L. Variable Minute Mapped (MB/s) 0.3 Shuffle (MB/s) 0.5 Output (MB/s) 45.7 doc-index-hits 2313178 docs-indexed 7936 dups-in-índex-rnerge 0 rnr-rnerge-calls 1954105 rm-rnerge-outputs 1954105 PA154 Jazykové modelování (9.2) MapReduce 14/32 MapReduce status: MRIiidexer-betaó-large-2003 10 28 00 03 Started; FriNov7 09:51:07 2003 -- up 0 hr 31 min 34 sec 1707 workers: 1 deaths Type Shards Done Active Input(ME) Done(MB) OuTput(MB) Map 13853 13853 0 878934.6 878934.6 523499.2 Shuffle 500 500 0 523499.2 523499.5 523499.5 Re due e 500 0 500 523499.5 133837.8 136929.6 ICO 90 30 70 *j a. sc ľ 0 u 50 *j z Ľ ■10 ü L U a_ j" 2C 10 0 ■c-oj Keduce Shard ■z- o T Counters Variable Mapped (MB/s) Shuffle (MB/s) Output (MB/s) doc-index-hits docs-indexed dups-in- index- merge mr- merge- calls c mr-^ merge -outputs Minute 0.0 0.1 1238.3 0 51738599 51738599 lí PA154 Jazykové modelování (9.2) MapReduce 15/32 44555 MapReduce status: MR_Indexer-betaó-large-2003_10_28 00 03 Started: FríNov7 09:51:07 2003 -- up 0 hr 33 min 22 sec 1707 workers: 1 deaths Type Shards Done Active Input(MB) D imp (ME) Output(ME) Map 13853 13853 0 878934.6 878934.6 523499.2 Shuffle 500 500 0 523499.2 5234995 523499.5 Reduce 500 0 500 523499.5 263283.3 269351.2 (M n Reduce Shard Counters Variable Mapped (MB/s) Shuffle (MB/s) Output (MB/s) doc- index-hits docs-indexed dups-in- index- merge rnr- rnerge- calls o mr li- merge-outputs Minute 0.0 0,0 1225.1 0 51842100 51S42100 K PA154 Jazykové modelování (9.2) MapReduce 16/32 MapReduce status: MR_Indexer-betaó-large-2003 10 28 00 03 Started: Fri Nov 7 09:51:07 2003 - up 0 hr 35 nun 08 sec 1707 workers; 1 deaths Type Shards Done Active Input(MB) D one (MB) Output(MB) Map 13S53 13853 0 878934.6 878934.6 523499.2 Shuffle 500 500 0 523499.2 523499.5 523499.5 Reduce 500 0 500 523499.5 390447.6 399457.2 Reduce Shard Counters Variable Mapped (MB/s) Shuffle (MB/s) Output (MB/s) doc- index-hits docs-indexed dups-in- index- rnerge mr-merge-r all í I merge-outputs Minute 0.0 0.0 1222.0 51640600 51640600 PA154 Jazykové modelování (9.2) MapReduce 17/32 MapReduce status: MR_Indexer-beta6-large-2003_10_28 00 03 Started: FnNov7 09:51:07 2003 - up 0 hr 37 min 01 sec 1707 workers; 1 deaths Type Shards Done Active Input(MB) D Dne (MB) Output(ME) Map 13853 13853 0 873934.6 878934 6 523499.2 Shuffle 500 500 0 523499.2 520463.6 520463.6 Reduce 500 406 94 520468.6 512265.2 514373,3 Reduce Shard Counters Variable Mapped (MB/s) Shuffle (MB/s) Output (MB/s) doc- index-hits docs-indexed dups-un- index- merge mr- merge- calls mr- merge- outouts IVIinute 0.0 0.0 849.5 350S3350 35033350 It PA154 Jazykové modelování (9.2) MapReduce 18/32 MapReduce status: MRIndexer-betaó-large-2003 10 28_00_03 Started: Fri Nov 7 09:51:07 2003 - up Ohr 33min 56 sec 1707 workers; 1 deaths Counters Type Shards Done Active Input(M£) Donei MR) rhitput(ME) Variable Minute Mat) 13S53 13853 0 878934.6 878 934.6 523499.2 Mapped 0.0 Shuffle 500 500 0 523499.2 519 781.8 519781.8 (MB/;) Reduce 500 498 2 519781 3 519 3947 5194407 Shuffle (MB/s) 0.0 Output (MB/s) i<"'' ■■■■■■■■■■■■■■■■■■■■■■■■■■■■ 90 9.4 60 "S 70 i ^ 60 c □ doc-index-hits 0 105( docs-indexed 0 * ■p £ 40 o dups-m-index- 0 £ 30 merge 20 10 mr- merge-calls 394792 ■ 0. miimmiimiimiiimii o o Reduce Sh O O G O O C Tŕ IT ard rnr-rnerge-outputs 394792 * PA154 Jazykové modelování (9.2) MapReduce 19/32 MapReduce status: MR_Indexer-beta6-large-2003 10 28 00 03 Started: Fn Nov 7 09:51:07 2003 - up 0 hr 40 mm 43 sec 1707 workers; 1 deaths Counters Type Shards Done Active lnput(M_B) D one (MB) Output(MB) Map 13S53 13S53 0 878934.6 878934.6 523499.2 Shuffle 500 500 0 523499.2 519774.3 519774.3 Reduce 500 499 1 519774.3 519735.2 519764.0 Reduce Shard Variable Minute Mapped (MB/s) 0.0 Shuffle (MB/s) 0.0 Output (MB/s) 1.9 doc-index-hits 0 105« docs-indexed 0 dups-in-index-merge 0 mr-merge-calls 73442 - mr-merge-OUtDUtS 73442 - PA154 Jazykové modelování (9.2) MapReduce 20/32 Fault tolerance: Handled via re-execution ■ On worker failure: ► Detect failure via periodic heartbeats ► Re-execute completed and in-progress map tasks ► Re-execute in progress reduce tasks ► Task completion committed through master ■ Master failure: ► Could handle, but don't yet (master failure unlikely) Robust: lost 1600 of 1800 machines once, but finished fine Semantics in presence of failures: see paper PA154 Jazykové modelování (9.2) MapReduce 21/32 Refinement: Redundant Execution Slow workers significantly lengthen completion time ■ Other jobs consuming resources on machine ■ Bad disks with soft errors transfer data very slowly ■ Weird things: processor caches disabled (!!) Solution: Near end of phase, spawn backup copies of tasks ■ Whichever one finishes first "wins" Effect: Dramatically shortens job completion time PA154 Jazykové modelování (9.2) MapReduce 22/32 Refinement: Locality Optimization Master scheduling policy: ■ Asks GFS for locations of replicas of input file blocks ■ Map tasks typically split into 64MB (== GFS block size) ■ Map tasks scheduled so GFS input block replica are on same machine or same rack Effect: Thousands of machines read input at local disk speed ■ Without this, rack switches limit read rate PA154 Jazykové modelování (9.2) MapReduce 23/32 Refinement: Skipping Bad Records Map/Reduce functions sometimes fail for particular inputs ■ Best solution is to debug & fix, but not always possible ■ On seg fault: ► Send UDP packet to master from signal handler ► Include sequence number of record being processed ■ If master sees two failures for same record: ► Next worker is told to skip the record Effect: Can work around bugs in third-party libraries PA154 Jazykové modelování (9.2) MapReduce 24/32 Other Refinements (see paper) Sorting guarantees within each reduce partition Compression of intermediate data Combiner: useful for saving network bandwidth Local execution for debugging/testing User-defined counters PA154 Jazykové modelování (9.2) MapReduce 25/32 Performance Tests run on cluster of 1800 machines: ■ 4 GB of memory ■ Dual-processor 2 GHz Xeons with Hyperthreading ■ Dual 160 GB IDE disks ■ Gigabit Ethernet per machine ■ Bisection bandwidth approximately 100 Gbps Two benchmarks: MFLGrep Scan 1010 100-byte records to extract records matching a rare pattern (92K matching records) MFLSort Sort 1010 100-byte records (modeled after TeraSort benchmark) PA154 Jazykové modelování (9.2) MapReduce 26/32 MR.Grep Locality optimization helps: ■ 1800 machines read 1 TB of data at peak of « 31 GB/s ■ Without this, rack switches would limit to 10 GB/s Startup overhead is significant for short jobs PA154 Jazykové modelování (9.2) MapReduce 27/32 MR.Sort Backup tasks reduce job completion time significantly System deals well with failures Normal No backup tasks 200 processes killed 20000 -i 10000 +-> CL c Done: 839 s n—i—i—r—i—r 0 200 400 600 60 > 10001200 ~ 20000 -, 10000 - <4- 3 "v. LJŮ i r 0 200 400 600 80 20000 10000 - 3 CL ~i r ) 10001200 i-í-1-r—i-r 0 200 400 600 800 10001200 Seconds 20000 -, 10000 Done: 1235 s i—i—i—i—i—r 0 200 400 600 800 1000 12(0 20000 -, 10000 - 20000 —i-r 0 200 400 600 800 100012< 10000 - ^rv^-1 , i 0 200 400 600 800 10001200 Seconds 20000 -i 10000 0 Done: 886 s 4 tjy~\—i—r 0 200 400 600 800 20000 -, 10000 - i i r 0 200 400 600 800 20000 10000 - -1-r 10001200 ~~i r 10001200 i i i i i r 0 200 400 600 800 10001200 Seconds PA154 Jazykové modelování (9.2) MapReduce 28/32 Experience: Rewrite of Production Indexing System Rewrote Google's production indexing system using MapReduce ■ Set of +9, 44, 47, 24, 24 MapReduce operations ■ New code is simpler, easier to understand ■ MapReduce takes care of failures, slow machines ■ Easy to make indexing faster by adding more machines PA154 Jazykové modelování (9.2) MapReduce 29/32 Usage: MapReduce jobs run in August 2004 Number of jobs 29,423 Average job completion time 634 sees Machine days used 79,186 days Input data read 3,288 TB Intermediate data produced 758 TB Output data written 193 TB Average worker machines per job 157 Average worker deaths per job 1.2 Average map tasks per job 3,351 Average reduce tasks per job 55 Unique map implementations 395 Unique reduce implementations 269 Unique map/reduce combinations 426 MapReduce Related Work ■ Programming model inspired by functional language primitives ■ Partitioning/shuffling similar to many large-scale sorting systems ► NOW-Sort ['97] ■ Re-execution for fault tolerance ► BAD-FS ['04] and TACC ['97] ■ Locality optimization has parallels with Active Disks/Diamond work ► Active Disks ['01], Diamond ['04] ■ Backup tasks similar to Eager Scheduling in Charlotte system ► Charlotte ['96] ■ Dynamic load balancing solves similar problem as River's distributed queues ► River ['99] PA154 Jazykové modelování (9.2) MapReduce 31/32 Conclusions ■ MapReduce has proven to be a useful abstraction ■ Greatly simplifies large-scale computations at Google ■ Fun to use: focus on problem, let library deal w/ messy details Thanks to Josh Levenberg, who has made many significant improvements and to everyone else at Google who has used and helped to improve MapReduce. PA154 Jazykové modelování (9.2) MapReduce 32/32