Multiround algorithms are now commonly used in distributed data processing systems, yet the extent to which algorithms can benefit from running more rounds is not well understood. This paper answers this question for several rounds for the problem of computing the equijoin of n relations. Given any query Q with width w, intersection width iw, input size IN, output size OUT, and a cluster of machines with M = Ω(IN 1 ) memory available per machine, where > 1 and w ≥ 1 are constants, we show that: Intersection width is a new notion we introduce for queries and generalized hypertree decompositions (GHDs) of queries that captures how connected the adjacent components of the GHDs are. We achieve our first result by introducing a distributed and generalized version of Yannakakis's algorithm, called GYM. GYM takes as input any GHD of Q with width w and depth d, and computes Q in O(d + log(n)) rounds and O(n (IN w +OUT) 2 M ) communication cost. We achieve our second result by showing how to construct GHDs of Q with width max(w, 3iw) and depth O(log(n)). We describe another technique to construct GHDs with longer widths and shorter depths, demonstrating other tradeoffs one can make between communication and the number of rounds. network. Costs of algorithms in these systems can be broken down to: (1) local computation of machines; (2) communication between the machines; and (3) the number of new rounds of computation that are started, which can have large overheads in some systems, e.g. due to reading input from disk or waiting for resources to be available in the cluster. In this paper, we focus on communication and the number of rounds, as for many data processing tasks, the computation cost is generally subsumed by the communication cost [20, 21]. This paper studies the problem of evaluating an equijoin query Q in multiple rounds of computation in a distributed cluster. We restrict ourselves to queries that are full, i.e. do not contain projections, but queries can contain self-joins. We let n be the number of relations, IN the input size, OUT the output size of Q, and M = o(IN) the memory available per machine in the cluster. Memory sizes of the machines intuitively capture different parallelism levels: when memory sizes are smaller, we need a larger number of machines to evaluate the join, which increases parallelism. We assume throughout the paper that M = Ω(IN 1 ) for some constant > 1. For practical values of input and memory sizes, is a small constant. For instance, if IN is in terabytes, then even when M is in megabytes, ≈ 2. Our study of multiround join algorithms is motivated by two developments. First, it has been shown recently that there are prohibitively high lower bounds on the communication cost of any one-round algorithm for evaluating some join queries [2, 5]. For example, for the chain query, C n =R 1 (A 0 , A 1 ) R 2 (A 1 , A 2 ) ... R n (A n−1 , A n ), the lower bound on the communication cost of any one-round algorithm is ≥ ( IN M ) n/4 . 1 For example, if the input is one petabyte, i.e., IN=10 15 , even when we have machines with ten gigabytes of memory, i.e., M =10 10 , the communication cost of any one-round algorithm to evaluate the C 16 query is 100000 petabytes. Moreover, this lower bound holds even when the query output is known to be small, e.g., OUT = O(IN), and the input has no skew [5], implying that designing multiround algorithms is the only way to compute such joins more efficiently. Second, running a new round of computation has decreased from several minutes in the early systems (e.g., Hadoop [4]) to milliseconds in some recent systems (e.g., Spark [32]), making it practical to run algorithms that consist of a large number of rounds. Although multiround algorithms are becoming commonplace, how much algorithms can benefit from running more rounds is not well understood. In this paper, we answer this question for equijoin queries. We describe a multiround algorithm, called GYM, for Generalized Yannakakis in MapReduce (Sections 4-5), which is a distributed and generalized version of Yannakakis's algorithm for acyclic queries [31]. The performance of GYM depends on two important structural properties of the input query: depths and widths of its generalized hypertree decompositions (GHDs). We then present two algorithms, Log-GTA (Section 6) and C-GTA (Section 7), for constructing GHDs of queries with different depths and widths, exposing a spectrum of tradeoffs one can make between the number of rounds and communication using GYM. In the remainder of this section, we give an overview of our results.
Reference REF has introduced a multiround join algorithm called GYM, which takes as input a generalized hypertree decomposition (GHD) D of Q. The algorithm first computes several intermediate relations based on D in one round using Shares.
GYM: A Multiround Distributed Join Algorithm
ICDT
Mathematics Computer Science