Automating Distributed Partial Aggregation
Addressing challenges in distributed computing systems by automating partial aggregation processes, optimizing performance, and reducing network latency through techniques such as MapReduce and user-defined reducers.
Download Presentation
Please find below an Image/Link to download the presentation.
The content on the website is provided AS IS for your information and personal use only. It may not be sold, licensed, or shared on other websites without obtaining consent from the author.If you encounter any issues during the download, it is possible that the publisher has removed the file from their server.
You are allowed to download the files provided on this website for personal or commercial use, subject to the condition that they are used lawfully. All files are the property of their respective owners.
The content on the website is provided AS IS for your information and personal use only. It may not be sold, licensed, or shared on other websites without obtaining consent from the author.
E N D
Presentation Transcript
Automating Distributed Partial Aggregation Chang Liu1, Jiaxing Zhang2, Hucheng Zhou2, Sean McDirmid2, Zhenyu Guo2, Thomas Moscibroda2 1University of Maryland, College Park 2Microsoft Research
MapReduce Framework Mapper emits key-value pairs Mapper Mapper Mapper Key-value pairs will be shuffled over the network, and the pairs with the same key are transmitted to the same reducer Network IO is one of the bottlenecks dominating the overall latency Reducer takes all pairs with the same key as input, and compute the output Reducer Reducer Reducer
MapReduce Partial Aggregation Some traditional reducer to apply partial aggregation Mapper Mapper Mapper SUM MAX COUNT AVERAGE By aggregating partial results produced by an initial reducer, with a combiner, network IO can be reduced significantly Initial Reducer Initial Reducer Initial Reducer Combiner In distributed computation scenario, most reducers are provided as user defined functions Final Reducer
Partial Aggregation state-of-the-art Most MapReduce systems support Partial Aggregation Hadoop Combiner SCOPE Recursive reducer Etc. Partial Aggregation is effective Our study shows a up to 99.98% shuffling IO reduction, and up to 62% overall latency reduction But most require programmers to manually rewrite the reducer Programmers may write bugs when programs become sophisticated We study 183 real SCOPE programs, and found 28 of them (15.3%) are problematic
An example of User Defined Reducer User Defined Reducer Row reduce(RowSet inputs) { int sum = 0, count = 0; bool isFirst = true; Row output; for(row in inputs.Rows) { if(isFirst) { output[0] = row[0]; sum = 10 + row[1]; count = 2; isFirst = false; } else sum+=row[1], count++; } output[1] = sum/count; return output; } Inputs Reducer Key Value output[0]=Key output[1]= 10+ ? 1 Inputs[i].Value 1+ Inputs 1 2 1 5 1 3 How to apply partial aggregation to such a reducer?
Automating Partial Aggregation Reduce the partial aggregation application problem to the combiner synthesis problem
Automating Partial Aggregation Reduce the partial aggregation application problem to the combiner synthesis problem A necessary and sufficient condition as the mathematical foundation
Automating Partial Aggregation Reduce the partial aggregation application problem to the combiner synthesis problem A necessary and sufficient condition as the mathematical foundation Programming language techniques to solve the following problems: Decomposable verification Combiner synthesis
How to apply partial aggregation? Row reduce(RowSet inputs) { int sum = 0, count = 0; bool isFirst = true; Mapper Mapper Row output; ? for(row in inputs.Rows) { Initial Reducer Initial Reducer if(isFirst) { output[0] = row[0]; sum = 10 + row[1]; count = 2; Combiner isFirst = false; } else sum+=row[1], count++; Final Reducer } output[1] = sum/count; return output; }
How to apply partial aggregation? It depends on the aggregator in the main loop body! Row reduce(RowSet inputs) { int sum = 0, count = 0; InitialReduce(inputs) int sum = 0, count = 0; // the same as the original reducer foreach (row in inputs.Rows) { } emit(output[0], sum, count, isFirst); bool isFirst = true; Row output; for(row in inputs.Rows) { if(isFirst) { output[0] = row[0]; sum = 10 + row[1]; count = 2; What is the Combiner? isFirst = false; } else sum+=row[1], count++; } output[1] = sum/count; FinalReduce(partialResults) emit(partialResults[0], sum/count) return output; }
What reducers have a combiner A mathematical foundation combiner? ? is decomposableif and only if there exists a combiner?such that: (commutativity of ?)? ?0,? ? = ? ?0,? ?1 (correctness of ?) ? ?0,? ? = ? ? ?0,? ,? ?0,? (commutativity of ?)? ?1,?2 = ? ?2,?1 (associativity of ?)?(?(?1,?2),?3) = ?(?1,?(?2,?3)) Yu et al. Distributed aggregation for data-parallel computing: interfaces and implementations. In SOSP 2009. Notation: ?0is the initial results ?(?,?) is the aggregator ?(?,?) is the combiner 1 is the operation concatenating two sequenses
Main theorem What is this? Theorem: ? is decomposable if and only if ? ?,?? = ?(?,??) holds for all ?,?,?. The combiner ? is uniquely determined by ?, and takes the form ? ?1,?2 = ?(?1,? ?2) where ? is any general inverse function of ? Commutativity of ? Existence of ? Commutativity of ? Associativity of ? Correctness of ?
General Inverse Function Initial Reducer ? ? ?(?) is a general inverse function of ? if and only if ? ?0,? ? = ?. H(s) ? ? ? may have many general inverse functions ? = ?(?2) ? = ? ?1 Initial Reducer Initial Reducer ?(?1) ?(?2) Question: will different general inverse functions result in different combiners? ?1 ?2 Initial Reducer We prove that for any two different general inverse functions ?,? , ?? ?? Combiner ?? ? ?
Decomposability Verification Decomposability Verification and Combiner Synthesis Programming Language Techniques Combiner Synthesis Theorem: ? is decomposable if and only if ? ?,?? = ?(?,??) holds for all ?,?,?. The combiner ? is uniquely determined by ?, and takes the form ? ?1,?2 = ?(?1,? ?2) where ? is any general inverse function of ?
Decomposability Verification Decomposability Verification and Combiner Synthesis Programming Language Techniques Combiner Synthesis Theorem: ? is decomposable if and only if ? ?,?? = ?(?,??) holds for all ?,?,?. The combiner ? is uniquely determined by ?, and takes the form ? ?1,?2 = ?(?1,? ?2) where ? is any general inverse function of ? Decomposability verification: checking ? ?,?? = ?(?,??) holds true for all ?,?,?
Decomposability Verification Decomposability Verification and Combiner Synthesis Programming Language Techniques Combiner Synthesis Theorem: ? is decomposable if and only if ? ?,?? = ?(?,??) holds for all ?,?,?. The combiner ? is uniquely determined by ?, and takes the form ? ??,?? = ?(??,? ??) where ? is any general inverse function of ? Decomposability verification: checking ? ?,?? = ?(?,??) holds true for all ?,?,? Combiner Synthesis: given ?, find an arbitrary general inverse function.
Decomposability Verification SMT solver-based method Accumulator? Path formula for ? (isFirsti= 1 keyo= x 0 sumo= 10 + x 1 counto= 2 isFirsto= 0) (isFirsti= 0 keyo= keyi sumo= sumi+ x 1 counto= counti+ 1 isFirsto= isFirsti if(isFirst) { Symbolic Execution key = x[0]; sum = 10 + x[1]; count = 2; isFirst = false; } else sum+=x[1], count++; Formula for ? ?0,?? ?(?0,??) Unsatisfiable! ? is Decomposable!
Combiner Synthesis Hard Finding ? is a nondeterministic program inversion problem We find that many eligible reducers belong to the following categories: Counting Single Input State Machine
Combiner Synthesis COUNTING Example Accumulator ? if(isFirst) { key = x[0]; sum = 10 + x[1]; count = 2; isFirst = false; } else { sum+=x[1]; count++; } Observation: the results only depends on the size of the input sequence, instead of their values. Strategy: Mimic the input size to get one partial result, then aggregate these input over the other partial result. Heuristics to accelerate: if the aggregator is a linear function of COUNT, then compute the function.
Combiner Synthesis Single Input Example Accumulator ? if(isFirst) { key = x[0]; sum = 10 + x[1]; count = 2; isFirst = false; } else { sum+=x[1]; count++; } Observation: there exists a general inverse function whose output is a sequence of length one. Strategy: Given the partial results, find one input value to compute this result, and apply that value over the other partial result. Illustrating Example: given sum, then sum = 10 + x 1 x 1 = sum 10 So Combine sumx,sumy = sumx+ (sumy 10) More details can be found in our paper.
Evaluation We study 4,429 SCOPE jobs 183 have already used partial aggregation 28 of them (15.3%) are buggy Xiao et al. Nondeterminism in MapReduce Considered Harmful? An Empirical Study on Non-commutative Aggregators in MapReduce Programs, ICSE 2014 We identify 261 more jobs that are eligible for partial aggregation 22 unique reducers (reducers are heavily reused in different jobs)
Evaluation decomposability verification We verify ? ?,?? = ?(?,??) over the 22 identified reducers with two quantifier: (1) ?,?,? and (2) ?,?.? = ?01. 1The reasons why to use two quantifiers can be found in Section 6.1
Evaluation combiner synthesis We apply our techniques to synthesize the combiner for these 22 identified reducers. C for COUNTING, SM for State Machine, SI for Single Input We failed on synthesizing the combiner for reducer 21 and 22.
Evaluation Performance Gain Real job with tens of TBs input data End-to-end latency reduction from 165 seconds to 64 seconds (61.6%) Data volume over network decreases from 7.99 GB to 1.22 MB (99.98%) Synthesized jobs (SUM, COUNT, and MAX) An average of 62.4% reduction in latency An average of 76% reduction in network I/O
Conclusion and Future Directions A necessary and sufficient condition for decomposability SMT solver-based methods for the decomposability verification problem Several techniques to solve the combiner synthesis problem Evaluation on real dataset shows our techniques are effective Future directions More techniques handling the combiner synthesis problem
