Symmetric and Asymmetric Aggregate Function in Massively Parallel Computing

Abstract

Applications of aggregation for information summary have great meanings in various elds. In big data era, processing aggregate function in parallel is drawing researchers' attention. The aim of our work is to propose a generic framework enabling to map an arbitrary aggregation into a generic algorithm and identify when it can be e ciently executed on modern large-scale data-processing systems. We describe our preliminary results regarding classes of symmetric and asymmetric aggregation that can be mapped, in a systematic way, into e cient MapReduce-style algorithms.

1. INTRODUCTION

The ability to summarize information is drawing increasing attention for information analysis [ 11, 6 ]. Simultaneously, under the progress of data explosive growth processing aggregate function has to experience a transition to massively distributed and parallel platforms, e.g. Hadoop MapReduce, Spark, Flink etc. Therefore aggregation function requires a decomposition approach in order to execute in parallel due to its inherent property of taking several values as input and generating a single value based on certain criteria. Decomposable aggregation function can be processed in a way that computing partial aggregation and then merging them at last to obtain nal results.

Decomposition of aggregation function is a long-standing research problem due to its bene ts in various elds. In distributed computing platforms, decomposability of aggregate function can push aggregation before shu e phase [ 17, 3 ]. This is usually called initial reduce, with which the size of data transmission on a network can be substantially reduced. For wireless sensor network, the need to reduce data transmission is more necessary because of limitation of power supply [ 15 ]. In online analytical processing (OLAP), decomposability of aggregate function enables aggregation across multi-dimensions, such that aggregate queries can be executed on pre-computation results instead of base data to accelerate query answering [ 8 ]. An important point of query optimization in relational databases is to reduce table size for join [ 10 ], and decomposable aggregation brings interests [ 4 ].

When an arbitrary aggregation function is decomposable, how to decompose it and when a decomposition is 'e cient' is a hard nut to crack. Previous works identify interesting properties for decomposing aggregation. A very relevant classi cation of aggregation functions, introduced in [ 11 ], is based on the size of sub-aggregation (i.e., partial aggregation). This classi cation distinguishes between distributive and algebraic aggregation having sub-aggregation with xed sizes, and holistic functions where there is no constant bound on the storage size of sub-aggregation. Some algebraic properties, such as associativity and commutativity, are identied as su cient conditions for decomposing aggregation [ 17, 3 ]. Compared to these works, our work provides a generic framework to identify the decomposability of any symmetric aggregation and generate generic algorithms to process it in parallel. Moreover, all but few researches in the literature consider symmetric functions. Asymmetric aggregation is inherently non-commutative functions and this makes their processing in parallel and distributed environment far from being easy. In [ 16 ], a symbolic parallel engine (SYMPLE) is proposed in order to automatically parallelize User De ned Aggregations (UDAs) that are not necessarily commutative. Although interesting, the proposed framework lacks guarantees for e ciency and accuracy in the sense that it is up to users to encode a function as SYMPLE UDA. Moreover, symbolic execution may have path explosion problem.

My research focuses on designing generic framework that enables to map symmetric and asymmetric aggregation functions into e cient massively parallel algorithms. To achieve this goal, we rstly identify a computation model, and an associated cost model to design and evaluate parallel algorithms. We consider MapReduce-style (M R) framework and use the M RC [ 12 ] cost model to de ne 'e cient' M R algorithms. We rest on the notion of well-formed aggregation [ 4 ] as a canonical form to write symmetric aggregation and provide a simple and systematic way to map well-formed aggregation function into an MR algorithm, noted by M R( ). Moreover, we provide reducible properties to identify when the generated M R( ) is e cient (when M R( ) is an M RC algorithm). Then we extend our framework to a class of asymmetric aggregation function, position-based aggregation, and propose extractable property to have generic M RC algorithms. Our main results are Theorem 1 and Theorem 2, of which proofs are provided in an extended report[ 2 ]. 2.

M RC ALGORITHM

Several research works concentrate on the complexity of parallel algorithms. M U D[ 7 ] algorithm was proposed to transform symmetric streaming algorithms to parallel algorithms with nice bounds in terms of communication and space complexity, but without any bound on time complexity. This disquali es M U D as a possible candidate cost model to be used in our context. M RC[ 12 ] is another popular model that has been used to evaluate whether a MapeReduce algorithm is e cient. The constraints enforced by M RC w.r.t. total input data size can be summarized as following: sublinear number of total computing nodes, sublinear space for any mapper or reducer, polynomial time for any mapper or reducer, and logarithm round number. We illustrate these constraints besides round number in a simpli ed MapReduce owchart in gure 1 where > 0.

Hence, the M RC model considers necessary parameters for parallel computing, communication time, computation space and computing time, and makes more realistic assumptions. A MapReduce algorithm satisfying these constraints is considered as an e cient parallel algorithm and will be called hereafter an M RC algorithm.

SYMMETRIC AGGREGATION WITH M RC

Let I be a doamin, an n-ary aggregation is a function[ 9 ]: In ! I. is symmetric or commutative[ 9 ] if (X) = ( (X)) for any X 2 I and any permutation , where (X) = (x (1); :::; x (n)). Symmetric aggregation result does not depend on the order of input data, therefore input is considered as a multiset. In this section, we de ne a generic framework to map symmetric aggregation into an M RC algorithm. 3.1

A Generic Form for Symmetric Aggregation

To de ne our generic aggregation framework, we rest on the notion of well-formed aggregation [ 4 ]. A symmetric aggregation de ned on a multiset X = fd1; : : : ; dng can be written in well-formed aggregation as following: (X) = T (F (d1) : : :

F (dn)); where F is translating function(tuple at a time), is a commutative and associative binary operation, and T is terminating function. For instance, average can be easily transformed into well-formed aggregation: F (d) = (d; 1); (d; k) d (d0; k0) = (d + d0; k + k0) and T ((d; n)) = . In fact, any n symmetric aggregation can be rewritten into well-formed aggregation with a exible choice of , e.g = [.

Well-formed aggregation provides a generic plan for processing aggregate function in distributed architecture based on the associative and commutative property of : processing F and at mapper, and T at reducer. Table 1 depicts the corresponding generic MapReduce(MR) algorithm(the case of one key and trivially extending to any number of keys), noted by M R( ), where mapper input is a submultiset Xi of X and mapper output is oi, and P is the concatenation of .

However, the obtained M R( ) are not necessarily an e cient MapReduce algorithm. We identify when M R( ) is a M RC algorithm using reducibility property.

De nition 1. A symmetric aggregation function de ned on domain I is reducible if the well-formed aggregation (F; ; T ) of satis es 8di; dj 2 I : jF (di)

F (dj)j= O(1):

With this reducible property, we provide a theorem identifying when M R( ) of a symmetric aggregation is a M RC algorithm.

Theorem 1. Let be a symmetric well-formed aggregation and M R( ) be the generic algorithm for , then M R( ) is an MRC algorithm if and only if is reducible. 3.2

Deriving MRC Algorithm from Algebraic

Properties

In this section, we investigate several symmetric aggregation properties satisfying Theorem 1. If an aggregation is in one of the following classes, then has an M RC( ) algorithm illustrated in table 1.

An aggregate function is associative [ 9 ] if for multiset X = X1 [ X2, (X) = ( (X1); (X2)) : Associative and symmetric aggregation function can be transformed in well-formed aggregation (F; ; T ) as following, F = ; ;

= ; T = id where id denotes identity function. is reducible because it is an aggregation. Therefore M R( ) of associative and symmetric aggregation is an M RC algorithm.

An aggregation is distributive [ 11 ] if there exists a combining function C such that (X; Y ) = C( (X); (Y )). Distributive and symmetric aggregation can be rewritten in well-formed aggregation (F; ; T ) as following,

F = ;

= C; T = id: Similarly, is reducible and corresponding M R( ) is an M RC algorithm.

Another kind of aggregate function having the same behavior as symmetric and distributive aggregation is commutative semigroup aggregate function [ 5 ]. An aggregation is in this class if there exists a commutative semigroup (H; ), such that (X) = Nxi2X (xi). The corresponding well-formed aggregation (F; ; T ) is illustrated as following,

F = ; = ; T = id: (1) (2) (3) It is clearly that is reducible and M R( ) is an M RC algorithm.

A more general property than commutative semi-group aggregation is symmetric and preassociative aggregate function. An aggregation is preassociative [ 13 ] if it satises (Y ) = (Y 0) =) (XY Z) = (XY 0Z): According to [ 13 ], some symmetric and preassociative(unarily quasi-range-idempotent and continuous) aggregation functions can be constructed as (X) = Pin=1 '(xi) ; n 1; where and ' are continuous and strictly monotonic function. For instance, (X) = Pin=1 2 xi, where = id and '(xi) = 2 xi. The well-formed aggregation (F; ; T ) for this kind of preassociative aggregation is illustrated as following F = '; = +; T = : (4) The corresponding M R( ) is also an M RC algorithm.

An aggregate function is barycentrically associative [ 14 ] if it satis es (XY Z) = (X (Y )jY jZ), where jY j denotes the number of elements contained in multiset Y and (Y )jY j denotes jY j occurrences of (Y ). A well-known class of symmetric and barycentrically associative aggregation is quasiarithmetic mean : (X) = f 1 n1 Pin=1 f (xi) ; n 1; where f is an unary function and f 1 is a quasi-inverse of f . With di erent choices of f , can be di erent kinds of mean functions, e.g arithmetic mean, quadratic mean, harmonic mean etc. It is trivial to rewrite this kind of aggregation into well-formed aggregation (F; ; T ) and the M R( ) is also an M RC algorithm,

F = (f; 1); = (+; +); T = f 1(

Pn i=1 f (xi) ): n (5)

ASYMMETRIC AGGREGATION

Many commonly used aggregation function is symmetric(commutative) such that the order of input data can be ignored, while asymmetric aggregation considers the order. Two common asymmetric cases could be weighted aggregation and cumulative aggregation, where aggregated result will be changed if data order is changed, e.g. WMA(weighted moving average) and EMA(exponential moving average)[ 1 ], which are used to highlight trends. 4.1

A Generic Form for Asymmetric Aggregation

In contrast to symmetric aggregation, asymmetric function is impossible to rewrite into well-formed aggregation, because translating function F is a tuple at a time function and is commutative and hence both of them are insensitive to the order. For this reason, we propose an extended form based on well-formed aggregation which is more suitable for asymmetric aggregation.

De nition 2. An asymmetric aggregation de ned on an ordered sequence X is an asymmetric well-formed aggregation if can be rewritten as following, (X) = T (F o(X; x1) :::

F o(X; xn)); (6) where F o is order-in uenced translating function, is a commutative and associative binary operation, and T is terminating function.

For instance, (X) = P z)i 1xi[ 14 ] with a constant z can be rewritten asxiF2Xo((X1 ; xi) = (1 z)i 1xi; = +; T = id; where i is the position of xi in the sequence X.

Asymmetric well-formed aggregation can rewrite any asymmetric aggregation , and with the associative property of , also has a generic MR algorithm M R( ): processing F o and at mapper, and T at reducer. Similar to the behavior of symmetric well-formed aggregation, reducible property is needed to ensure M RC constraints. The reducible property for asymmetric well-formed aggregation is 8xi; xi+1 2 X : jF o(X; xi)

F o(X; xi+1)j= O(1): However, in order to have a correct generic M RC algorithm for asymmetric aggregation, reducible property is not enough, because asymmetric function considers data order such that operations for combining mapper outputs are more than . We illustrate this problem and identify properties to have correct MRC algorithm for a class of asymmetric well-formed aggregation in the following.

We deal with a kind of asymmetric aggregation called position-based aggregation, for which F o is F o(X; xi) = h(i) f (xi), where h() and f () are unary functions, and is a binary operation. The corresponding asymmetric well-formed framework is (X) = T (P ;xi2X h(i) f (xi)), where P is the concatenation of .

Let X be an ordered sequence X = S1 ::: Sm, where Sl is a subsequence of X, l 2 f1; :::; mg and is the concatenation of subsequence, and i be the holistic position of xi in X and j be the relative position of xj in subsequence Sl. Then P F o(X; xi) of on any subsequence Sl is

X ;xi2Sl

X ;xj2Sl F o(X; xi) = h(j + k) f (xi); where j + k (j + k = i) is the holistic position of the jth element xj in Sl. In order to process in parallel on these subsequences, the rst requirement is to have l, which means in distributed and parallel computing data set is split into ordered chunks and chunk indexes can be stored. It can be trivially implemented in Hadoop[ 16 ]. Secondly, k is needed, the number of elements before Sl. Sequential distributing subsequence count values then starting aggregation is costly due to too many times of data transferring on network. If k can be extracted out of P ;xj2Sl h(j +k) f (xi), then can be processed without distributing counts because operations relating to count can be pushed to reducer. We identify conditions to extract k which we call extractable property.

Lemma 1. Given an ordered sequence X, a position-based asymmetric well-formed aggregation de ned in (F o; ; T ) and F o(X; xi) = h(i) f (xi) for any xi 2 X, where h() and f () are unary functions, is extractable if there exists a binary operation making h() satisfy h(i + k) = h(i) h(k + c) with a constant c, and ; and satisfy one of the following conditions, and

are same, and

are same and they are distributive over , is distributive over

which is same as .

The behavior of h() is similar to group homomorphism however they are not exactly same, and our intention is to extract k instead of preserving exact operations.

Theorem 2. Let be a position-based well-formed aggregation and M R( ) be the generic algorithm for , then M R( ) is an MRC algorithm if is reducible and extractable.

Extractable property of position-based aggregation allows previous subsequences count value 'k' to be extracted out of mapper operation, then can be correctly processed by P F o or (P f (xi); P h(i)) at mapper phase. To combine mapper outputs, more than and T are needed and speci c combining operation depends on the three different extractable conditions (provided in our extended report[ 2 ]).

For instance, given an input sequence X = (x1; :::; xn), then EM A(X) = PPin=in1=(11(1 a)ai)i1 1xi ; where a is a constant between 0 and 1. We give below the asymmetric well-formed aggregation of EM A, where h(i) = (1 a)i 1,

F o : F o(X; xi) = h(i) xi; h(i) ; : h(i) xi; h(i)

h(i + 1) xi+1; h(i + 1) = h(i) xi + h(i + 1) xi+1; h(i) + h(i + 1) ;

n n T : T (X h(i) xi; X h(i)) = i=1 i=1

Pn i=1 h(i) xi Pn i=1 h(i) :

It is clearly that EMA is a position-based aggregation, and EMA is reducible because is a pair of addition. Moreover h() satis es h(i + k) = h(i) h(k + 1), and the corresponding three binary operations = , = , = + satisfy the second extractable condition. Therefore EMA has a MRC algorithm(the generic MRC algorithm for the second extractable condition) illustrated as following, where we assume input sequence X = S1 ::: Sm and mapper input is Sl; l 2 f1; :::; mg, and count(S0) = 0; mapper: OMl0 = Pxj2Sl h(j) xj, OMl00 = Pxj2Sl h(j), OMl000 = count(Sl) , reducer:

Pm

l=1 OMl0 (1 Pm l=1 OM 00 (1 l a)Plj=10 OMj000 a)Plj=10 OMj000 .

CONCLUSION AND FUTURE WORK

In this work, we studied how to map aggregation functions, in a systematic way, into generic M RC algorithms and we identi ed properties that enable to e ciently execute symmetric and asymmetric aggregations using MapReducestyle platforms. For symmetric aggregation, we proposed the reducible property within well-formed aggregation framework to satisfy space and time complexity of M RC. Several algebraic properties of symmetric aggregation leading to a generic M RC algorithm have been identi ed. Moreover, we extended the notion of well-formed aggregation to asymmetric aggregation and showed how it can be exploited to deal with position-based asymmetric aggregation. Through identifying the problem for parallelizing it, we proposed extractable property and merged it with the reducible property of asymmetric well-formed aggregation to have M RC algorithms.

Our future work will be devoted to the implementation and experimentation. We will study the extension of our framework to mainstream parallel computing platforms (e.g. Apache Spark). Moreover, we also plan to extend our framework to cover additional classes of asymmetric aggregations. Finally, we plan to investigate how to generalize our approach to nested aggregation functions (i.e., functions dened as a complex composition of aggregation functions).

References

[1] Moving average . https://en.wikipedia.org/wiki/Moving_average.

[2] Symmetric and asymmetric aggregate function in massively parallel computing(extened version) . https://hal-clermont-univ.archives-ouvertes. fr/hal-01533675.

[3] C. Liu , J. Zhang , H. Zhou , Z. S. McDirmid , and T. Moscibroda . Automating distributed partial aggregation . In SOCC'14 , pages 1 { 12 , 2014 .

[4] S. Cohen . User-de ned aggregate functions: bridging theory and practice . In SIGMOD'06 , pages 49 { 60 , 2006 .

[5] S. COHEN , W.NUTT , and Y. SAGIV . Rewriting queries with arbitrary aggregation functions using views . ACM TODS , 31 ( 2 ): 672 { 715 , June 2006 .

[6] A. Cuzzocrea . Aggregation and multidimensional analysis of big data for large-scale scienti c applications: models, issues, analytics, and beyond . In SSDBM'15 , 2015 .

[7] J. Feldman , S.muthukrishnan, A. Sidiropoulos, C. Stein , and Z. Svitkina . On distributing symmetric streaming computations . ACM TALG , 6 ( 4 ), August 2010 .

[8] M. Franklin . An overview of data warehousing and olap technology . ACM SIGMOD Record , 26 ( 1 ): 65 { 74 , March 1997 .

[9] M. Grabisch , J.-L. Marichal , R. Mesiar , and E. Pap . Aggregation function: Means. Information Sciences , 181 ( 1 ):1{ 22 , January 2011 .

[10] H. Garcia-Molina , J.D. Ullman , and J. Widom . Database System Implementation. Prentice-Hall, New Jersey, 2000 .

[11] J. Gray , A. Bosworth , A. Layman , and H. Pirahesh . Data cube: A relational aggregation operator generalizing group-by, cross-tab, and sub-totals . Data Mining and Knowledge Discovery , 1 ( 1 ): 29 { 53 , Janaury 1997 .

[12] H. Karlo , S. Suri , and S. Vassilvitskii . A model of computation for mapreduce . In SODA'10 , pages 938 { 948 , 2010 .

[13] M. Jean-Luc and T. Bruno . Preassociative aggregation functions . Fuzzy Sets and Systems , 268 : 15 { 26 , June 2015 .

[14] M. Jean-Luc and T. Bruno . Strongly barycentrically associative and preassociative functions . Fuzzy Sets and Systems , 437 ( 1 ): 181 { 193 , May 2016 .

[15] S. Madden , M.J. Franklin , J.M. Hellerstein , and W. Hong . Tag: a tiny aggregation service for ad-hoc sensor networks . In OSDI'02 , pages 131 { 146 , 2002 .

[16] V. Raychev , M. Musuvathi , and T. Mytkowicz . Parallelizing user-de ned aggregaions using symbolic execution . In SOSP'15 , pages 153 { 167 , 2015 .

[17] Y. Yu , M. Isard , and P. Gunda . Distributed aggregation for data-parallel computing: Interfaces and implementations . In SOSP'09 , pages 247 { 260 , 2009 .