A Fundamental Tradeoff between Computation and Communication in Distributed Computing

I Introduction

We consider a general distributed computing framework, motivated by prevalent structures like MapReduce and Spark , in which the overall computation is decomposed into two stages: “Map” and “Reduce”. Firstly in the Map stage, distributed computing nodes process parts of the input data locally, generating some intermediate values according to their designed Map functions. Next, they exchange the calculated intermediate values among each other (a.k.a. data shuffling), in order to calculate the final output results distributedly using their designed Reduce functions.

Within this framework, data shuffling often appears to limit the performance of distributed computing applications, including self-join , tera-sort , and machine learning algorithms . For example, in a Facebook’s Hadoop cluster, it is observed that 33% of the overall job execution time is spent on data shuffling . Also as is observed in , 70% of the overall job execution time is spent on data shuffling when running a self-join application on an Amazon EC2 cluster . As such motivated, we ask this fundamental question that if coding can help distributed computing in reducing the load of communication and speeding up the overall computation? Coding is known to be helpful in coping with the channel uncertainty in telecommunication and also in reducing the storage cost in distributed storage systems and cache networks. In this work, we extend the application of coding to distributed computing and propose a framework to substantially reduce the load of data shuffling via coding and some extra computing in the Map phase.

More specifically, we formulate and characterize a fundamental tradeoff relationship between “computation load” in the Map phase and “communication load” in the data shuffling phase, and demonstrate that the two are inversely proportional to each other. We propose an optimal coded scheme, named “Coded Distributed Computing” (CDC), which demonstrates that increasing the computation load of the Map phase by a factor of $r$ (i.e., evaluating each Map function at $r$ carefully chosen nodes) can create novel coding opportunities in the data shuffling phase that reduce the communication load by the same factor.

To illustrate our main result, consider a distributed computing framework to compute $Q$ arbitrary output functions from $N$ input files, using $K$ distributed computing nodes. As mentioned earlier, the overall computation is performed by computing a set of Map and Reduce functions distributedly across the $K$ nodes. In the Map phase, each input file is processed locally, in one of the nodes, to generate $Q$ intermediate values, each corresponding to one of the $Q$ output functions. Thus, at the end of this phase, $QN$ intermediate values are calculated, which can be split into $Q$ subsets of $N$ intermediate values and each subset is needed to calculate one of the output functions. In the Shuffle phase, for every output function to be calculated, all $N$ intermediate values corresponding to that function are transferred to one of the nodes for reduction. Of course, depending on the node that has been chosen to reduce an output function, a part of the intermediate values are already available locally, and do not need to be transferred in the Shuffle phase. This is because that the Map phase has been carried out on the same set of nodes, and the results of mapping done at a node can remain in that node to be used for the Reduce phase. This offers some saving in the load of communication. To reduce the communication load even more, we may map each input file in more than one nodes. Apparently, this increases the fraction of intermediate values that are locally available. However, as we will show, there is a better way to exploit this redundancy in computation to reduce the communication load. The main message of this paper is to show that following a particular patten in repeating Map computations along with some coding techniques, we can significantly reduce the load of communication. Perhaps surprisingly, we show that the gain of coding in reducing communication load scales with the size of the network.

To be more precise, we define the computation load $r$, $1\leq r\leq K$, as the total number of computed Map functions at the nodes, normalized by $N$. For example, $r=1$ means that none of the Map functions has been re-computed, and $r=2$ means that on average each Map function can be computed on two nodes. We also define communication load $L$, $0\leq L\leq 1$, as the total amount of information exchanged across nodes in the shuffling phase, normalized by the size of $QN$ intermediate values, in order to compute the $Q$ output functions disjointly and uniformly across the $K$ nodes. Based on this formulation, we now ask the following fundamental question:

Given a computation load $r$ in the Map phase, what is the minimum communication load $L^{*}(r)$, using any data shuffling scheme, needed to compute the final output functions?

We propose Coded Distributed Computing (CDC) that achieves a communication load of $L_{\textup{coded}}(r)=\frac{1}{r}\cdot(1-\frac{r}{K})$ for $r=1,\ldots,K$, and the lower convex envelop of these points. CDC employs a specific strategy to assign the computations of the Map and Reduce functions across the computing nodes, in order to enable novel coding opportunities for data shuffling. In particular, for a computation load $r\in\{1,\ldots,K\}$, CDC utilizes a carefully designed repetitive mapping of data blocks at $r$ distinct nodes to create coded multicast messages that deliver data simultaneously to a subset of $r\geq 1$ nodes. Hence, compared with an uncoded data shuffling scheme, which as we show later achieves a communication load $L_{\textup{uncoded}}(r)=1-\frac{r}{K}$, CDC is able to reduce the communication load by exactly a factor of the computation load $r$. Furthermore, the proposed CDC scheme applies to a more general distributed computing framework where every output function is computed by more than one, or particularly $s\in\{1,\ldots,K\}$ nodes, which provides better fault-tolerance in distributed computing.

We numerically compare the computation-communication tradeoffs of CDC and uncoded data shuffling schemes (i.e., $L_{\textup{coded}}(r)$ and $L_{\textup{uncoded}}(r)$) in Fig. 1. As it is illustrated, in the uncoded scheme that achieves a communication load $L_{\textup{uncoded}}(r)=1-\frac{r}{K}$, increasing the computation load $r$ offers only a modest reduction in communication load. In fact for any $r$, this gain vanishes for large number of nodes $K$. Consequently, it is not justified to trade computation for communication using uncoded schemes. However, for the coded scheme that achieves a communication load of $L_{\textup{coded}}(r)=\frac{1}{r}\cdot(1-\frac{r}{K})$, increasing the computation load $r$ will significantly reduce the communication load, and this gain does not vanish for large $K$. For example as illustrated in Fig. 1, when mapping each file at one extra node ($r=2$), CDC reduces the communication load by 55.6%, while the uncoded scheme only reduces it by 11.1%.

We also prove an information-theoretic lower bound on the minimum communication load $L^{*}(r)$. To prove the lower bound, we derive a lower bound on the total number of bits communicated by any subset of nodes, using induction on the size of the subset. To derive the lower bound for a particular subset of nodes, we first establish a lower bound on the number of bits needed by one of the nodes to recover the intermediate values it needs to calculate its assigned output functions, and then utilize the bound on the number of bits communicated by the rest of the nodes in that subset, which is given by the inductive argument. The derived lower bound on $L^{*}(r)$ matches the communication load achieved by the CDC scheme for any computation load $1\leq r\leq K$. As a result, we exactly characterize the optimal tradeoff between computation load and communication load in the following:

For general $1\leq r\leq K$, $L^{*}(r)$ is the lower convex envelop of the above points $\{(r,L_{\textup{coded}}(r)):r\in\{1,\ldots,K\}\}$. Note that for large $K$, $\frac{1}{r}\cdot(1-\frac{r}{K})\approx\frac{1}{r}$, hence $L^{*}(r)\approx\frac{1}{r}$. This result reveals a fundamental inversely proportional relationship between computation load and communication load in distributed computing. This also illustrates that the gain of $\frac{1}{r}$ achieved by CDC is optimal and it cannot be improved by any other scheme (since $L_{\textup{coded}}(r)$ is an information-theoretic lower bound on $L^{*}(r)$ that applies to any data shuffling scheme).

Having theoretically characterized the optimal computation-communication tradeoff achieved by the proposed CDC scheme, we also empirically demonstrate the practical impact of this tradeoff. In particular, we apply the coding techniques of CDC to a widely used Hadoop sorting benchmark TeraSort , developing a novel coded distributed sorting algorithm CodedTeraSort . We perform extensive experiments on Amazon EC2 clusters, and observe that for typical settings of interest, CodedTeraSort speeds up the overall execution of the conventional TeraSort by a factor of $1.97\times$ - $3.39\times$.

Finally, we discuss some future directions to extend the results of this work. In particular, we consider topics including heterogeneous networks with asymmetric tasks, straggling/failing computing nodes, multi-stage computation tasks, multi-layer networks and structured topology, joint storage and computation optimization, and coded edge/fog computing.

Related Works. The problem of characterizing the minimum communication for distributed computing has been previously considered in several settings in both computer science and information theory communities. In , a basic computing model is proposed, where two parities have $x$ and $y$ and aim to compute a boolean function $f(x,y)$ by exchanging the minimum number of bits between them. Also, the problem of minimizing the required communication for computing the modulo-two sum of distributed binary sources with symmetric joint distribution was introduced in . Following these two seminal works, a wide range of communication problems in the scope of distributed computing have been studied (see, e.g., ). The key differences distinguishing the setting in this paper from most of the prior ones are 1) We focus on the flow of communication in a general distributed computing framework, motivated by MapReduce, rather than the structures of the functions or the input distributions. 2) We do not impose any constraint on the numbers of output results, input data files and computing nodes (they can be arbitrarily large), 3) We do not assume any special property (e.g. linearity) of the computed functions.

The idea of efficiently creating and exploiting coded multicasting was initially proposed in the context of cache networks in , and extended in , where caches pre-fetch part of the content in a way to enable coding during the content delivery, minimizing the network traffic. In this paper, we propose a framework to study the tradeoff between computation and communication in distributed computing. We demonstrate that the coded multicasting opportunities exploited in the above caching problems also exist in the data shuffling of distributed computing frameworks, which can be created by a strategy of repeating the computations of the Map functions specified by the Coded Distributed Computing (CDC) scheme.

Finally, in a recent work , the authors have proposed methods for utilizing codes to speed up some specific distributed machine learning algorithms. The considered problem in this paper differs from in the following aspects. We propose a general methodology for utilizing coding in data shuffling that can be applied to any distributed computing framework with a MapReduce structure, regardless of the underlying application. In other words, any distributed computing algorithm that fits in the MapReduce framework can benefit from the proposed CDC solution. We also characterize the information-theoretic computation-communication tradeoff in such frameworks. Furthermore, the coding used in is at the application layer (i.e., applying computation on coded data), while in this paper we focus on applying codes directly on the shuffled data.

II Problem Formulation

In this section, we formulate a general distributed computing framework motivated by MapReduce, and define the function characterizing the tradeoff between computation and communication.

Motivated by MapReduce, we assume that as illustrated in Fig. 2 the computation of the output function $\phi_{q}$, $q\in\{1,\ldots,Q\}$ can be decomposed as follows:

Note that for every set of output functions $\phi_{1},\ldots,\phi_{Q}$ such a Map-Reduce decomposition exists (e.g., setting $g_{q,n}$${}^{\prime}s$ to identity functions such that $g_{q,n}(w_{n})=w_{n}$ for all $n=1,\ldots,N$, and $h_{q}$ to $\phi_{q}$ in (1)). However, such a decomposition is not unique, and in the distributed computing literature, there has been quite some work on developing appropriate decompositions of computations like join, sorting and matrix multiplication (see, e.g., ), for them to be performed efficiently in a distributed manner. Here we do not impose any constraint on how the Map and Reduce functions are chosen (for example, they can be arbitrary linear or non-linear functions). $\hfill\square$

The above computation is carried out by $K$ distributed computing nodes, labelled as Node $1$$,\ldots,\,$Node $K$. They are interconnected through a multicast network. Following the above decomposition, the computation proceeds in three phases: Map, Shuffle and Reduce.

Map Phase: Node $k$, $k\in\{1,\ldots,K\}$ computes the Map functions of a set of files $\mathcal{M}_{k}$, which are stored on Node $k$, for some design parameter $\mathcal{M}_{k}\subseteq\{w_{1},\ldots,w_{N}\}$. For each file $w_{n}$ in $\mathcal{M}_{k}$, Node $k$ computes $\vec{g}_{n}(w_{n})\!=\!(v_{1,n},\ldots,v_{Q,n})$. We assume that each file is mapped by at least one node, i.e., $\underset{k=1,\ldots,K}{\cup}\mathcal{M}_{k}=\{w_{1},\ldots,w_{N}\}$.

We define the computation load, denoted by $r$, $1\leq r\leq K$, as the total number of Map functions computed across the $K$ nodes, normalized by the number of files $N$, i.e., $r\triangleq\frac{\sum_{k=1}^{K}|\mathcal{M}_{k}|}{N}$. The computation load $r$ can be interpreted as the average number of nodes that map each file. $\hfill\Diamond$

Beyond the symmetric task assignment considered in this paper, characterizing the optimal computation-communication tradeoff allowing general asymmetric task assignments is a challenging open problem. As the first step to study this problem, in our follow-up work in which the number of output functions $Q$ is fixed and the computing resources are abundant (e.g., number of computing nodes $K\gg Q$), we have shown that asymmetric task assignments can do better than the symmetric ones, and achieve the optimum run-time performance. $\hfill\square$

Having generated the message $X_{k}$, Node $k$ multicasts it to all other nodes.

By the end of the Shuffle phase, each of the $K$ nodes receives $X_{1},\ldots,X_{K}$ free of error.

Finally, Node $k$, $k\in\{1,\ldots,K\}$, computes the Reduce function $u_{q}=h_{q}(v_{q,1}\ldots v_{q,N})$ for all $q\in\mathcal{W}_{k}$.

We define the computation-communication function of the distributed computing framework

$L^{*}(r)$ characterizes the optimal tradeoff between computation and communication in this framework. $\hfill\Diamond$

Example (Uncoded Scheme). In the Shuffle phase of a simple “uncoded” scheme, each node receives the needed intermediate values sent uncodedly by some other nodes. Since a total of $QN$ intermediate values are needed across the $K$ nodes and $rN\cdot\frac{Q}{K}=\frac{rQN}{K}$ of them are already available after the Map phase, the communication load achieved by the uncoded scheme

After the Map phase, each node knows the intermediate values of all $Q$ output functions in the files it has mapped. Therefore, for a fixed file assignment and any symmetric assignment of the Reduce functions, specified by ${\cal W}_{1},\ldots,{\cal W}_{K}$, we can satisfy the data requirements using the same data shuffling scheme up to relabelling the Reduce functions. In other words, the communication load is independent of the assignment of the Reduce functions. $\hfill\square$

III Main Results

The computation-communication function of the distributed computing framework, $L^{*}(r)$ is given by

for sufficiently large $T$. For general $1\leq r\leq K$, $L^{*}(r)$ is the lower convex envelop of the above points $\{(r,\frac{1}{r}\cdot(1-\frac{r}{K})):r\in\{1,\ldots,K\}\}$.

We prove the achievability of Theorem 1 by proposing a coded scheme, named Coded Distributed Computing, in Section V. We demonstrate that no other scheme can achieve a communication load smaller than the lower convex envelop of the points $\{(r,\frac{1}{r}\cdot(1-\frac{r}{K})):r\in\{1,\ldots,K\}\}$ by proving the converse in Section VI.

Theorem 1 exactly characterizes the optimal tradeoff between the computation load and the communication load in the considered distributed computing framework. $\hfill\square$

For $r\in\{1,\ldots,K\}$, the communication load achieved in Theorem 1 is less than that of the uncoded scheme in (5) by a multiplicative factor of $r$, which equals the computation load and can grow unboundedly as the number of nodes $K$ increases if e.g. $r=\Theta(K)$. As illustrated in Fig. 1 in Section I, while the communication load of the uncoded scheme decreases linearly as the computation load increases, $L_{\textup{coded}}(r)$ achieved in Theorem 1 is inversely proportional to the computation load.$\hfill\square$

While increasing the computation load $r$ causes a longer Map phase, the coded achievable scheme of Theorem 1 maximizes the reduction of the communication load using the extra computations. Therefore, Theorem 1 provides an analytical framework to optimally trading the computation power in the Map phase for more bandwidth in the Shuffle phase, which helps to minimize the overall execution time of applications whose performances are limited by data shuffling. $\hfill\square$

The computation-communication function of the cascaded distributed computing framework, $L^{*}(r,s)$, for $r\in\{1,\ldots,K\}$, is characterized by

for some $s\in\{1,\ldots,K\}$ and sufficiently large $T$. For general $1\leq r\leq K$, $L^{*}(r,s)$ is the lower convex envelop of the above points $\{(r,L_{\textup{coded}}(r,s)):r\in\{1,\ldots,K\}\}$.

We present the Coded Distributed Computing scheme that achieves the computation-communication function in Theorem 2 in Section V, and the converse of Theorem 2 in Section VII.

A preliminary part of this result, in particular the achievability for the special case of $s\!=\!1$, or the achievable scheme of Theorem 1 was presented in . We note that when $s=1$, Theorem 2 provides the same result as in Theorem 1, i.e., $L^{*}(r,1)=\frac{1}{r}\cdot(1-\frac{r}{K})$, for $r\in\{1,\ldots,K\}$. $\hfill\square$

For any fixed $s\in\{1,\ldots,K\}$ (number of nodes that compute each Reduce function), as illustrated in Fig. 3, the communication load achieved in Theorem 2 outperforms the linear relationship between computation and communication, i.e., it is superlinear with respect to the computation load $r$. $\hfill\square$

Before we proceed to describe the general achievability scheme for the cascaded distributed computing framework (also the distributed computing framework as a special case of $s=1$), we first illustrate the key ideas of the proposed Coded Distributed Computing scheme by presenting two examples in the next section, for the cases of $s=1$ and $s>1$ respectively.

IV Illustrative Examples: Coded Distributed Computing

In this section, we present two illustrative examples of the proposed achievable scheme for Theorem 1 and Theorem 2, which we call Coded Distributed Computing (CDC), for the cases of $s=1$ (Theorem 1) and $s>1$ (Theorem 2) respectively.

We consider a MapReduce-type problem in Fig. 4 for distributed computing of $Q=3$ output functions, represented by red/circle, green/square, and blue/triangle respectively, from $N=6$ input files, using $K=3$ computing nodes. Nodes $1$, $2$, and $3$ are respectively responsible for final reduction of red/circle, green/square, and blue/triangle output functions. Let us first consider the case where no redundancy is imposed on the computations, i.e., each file is mapped once and computation load $r=1$. As shown in Fig. 4(a), Node $k$ maps File $2k-1$ and File $2k$ for $k=1,2,3$. In this case, each node maps $2$ input files locally, computing all three intermediate values needed for the three output functions from each mapped file. In Fig. 4, we represent, for example, the intermediate value of the red/circle function in File $n$ using a red circle labelled by $n$, for all $n=1,\ldots,6$. Similar representations follow for the green/square and the blue/triangle functions. After the Map phase, each node obtains $2$ out of $6$ required intermediate values to reduce the output function it is responsible for (e.g., Node 1 knows the red circles in File 1 and File 2). Hence, each node needs $4$ intermediate values from the other nodes, yielding a communication load of $\frac{4\times 3}{3\times 6}=\frac{2}{3}$.

Now, we demonstrate how the proposed CDC scheme trades the computation load to slash the communication load via in-network coding. As shown in Fig. 4(b), we double the computation load such that each file is now mapped on two nodes ($r=2$). It is apparent that since more local computations are performed, each node now only requires $2$ other intermediate values, and an uncoded shuffling scheme would achieve a communication load of $\frac{2\times 3}{3\times 6}=\frac{1}{3}$. However, we can do much better with coding. As shown in Fig. 4(b), instead of unicasting individual intermediate values, every node multicasts a bit-wise XOR, denoted by $\oplus$, of $2$ locally computed intermediate values to the other two nodes, simultaneously satisfying their data demands. For example, knowing the blue/triangle in File $3$, Node $2$ can cancel it from the coded packet sent by Node $1$, recovering the needed green/square in File $1$. Therefore, this coding incurs a communication load of $\frac{3}{3\times 6}=\frac{1}{6}$, achieving a $2\times$ gain from the uncoded shuffling. $\square$

From the above example, we see that for the case of $s=1$, i.e., each of the $Q$ output functions is computed on one node and the computations of the Reduce functions are symmetrically distributed across nodes, the proposed CDC scheme only requires performing bit-wise XOR as the encoding and decoding operations. However, for the case of $s>1$, as we will show in the following example, the proposed CDC scheme requires computing linear combinations of the intermediate values during the encoding process.

In this example, we consider a job of computing $Q=6$ output functions from $N=6$ input files, using $K=4$ nodes. We focus on the case where the computation load $r=2$, and each Reduce function is computed by $s=2$ nodes. In the Map phase, each file is mapped by $r=2$ nodes. As shown in Fig. 5, the sets of the files mapped by the $4$ nodes are $\mathcal{M}_{1}=\{w_{1},w_{2},w_{3}\}$, $\mathcal{M}_{2}=\{w_{1},w_{4},w_{5}\}$, $\mathcal{M}_{3}=\{w_{2},w_{4},w_{6}\}$, and $\mathcal{M}_{4}=\{w_{3},w_{5},w_{6}\}$. After the Map phase, Node $k$, $k\in\{1,2,3,4\}$, knows the intermediate values of all $Q=6$ output functions in the files in $\mathcal{M}_{k}$, i.e., $\{v_{q,n}:q\in\{1,\ldots,6\},w_{n}\in\mathcal{M}_{k}\}$. In the Reduce phase, we assign the computations of the Reduce functions in a symmetric manner such that every subset of $s\!=\!2$ nodes compute a common Reduce function. More specifically as shown in Fig. 5, the sets of indices of the Reduce functions computed by the $4$ nodes are $\mathcal{W}_{1}\!=\!\{1,2,3\}$, $\mathcal{W}_{2}\!=\!\{1,4,5\}$, $\mathcal{W}_{3}\!=\!\{2,4,6\}$, and $\mathcal{W}_{4}\!=\!\{3,5,6\}$. Therefore, for example, Node 1 still needs the intermediate values $\{v_{q,n}\!:\!q\in\{1,2,3\},n\in\{4,5,6\}\}$ through data shuffling to compute its assigned Reduce functions $h_{1}$, $h_{2}$, $h_{3}$.

The data shuffling process consists of two rounds of communication over the multicast network. In the first round, intermediate values are communicated within each subset of $3$ nodes. In the second round, intermediate values are communicated within the set of all $4$ nodes. In what follows, we describe these two rounds of communication respectively.

Round 1: Subsets of $3$ nodes. We first consider the subset $\{1,2,3\}$. During the data shuffling, each node whose index is in $\{1,2,3\}$ multicasts a bit-wise XOR of two locally computed intermediate values to the other two nodes:

Node 1 multicasts $v_{1,2}\oplus v_{2,1}$ to Node $2$ and Node $3$,

Node 2 multicasts $v_{4,1}\oplus v_{1,4}$ to Node $1$ and Node $3$,

Node 3 mulicasts $v_{4,2}\oplus v_{2,4}$ to Node $1$ and Node $2$,

Since Node 2 knows $v_{2,1}$ and Node 3 knows $v_{1,2}$ locally, they can respectively decode $v_{1,2}$ and $v_{2,1}$ from the coded message $v_{1,2}\oplus v_{2,1}$.

We employ the similar coded shuffling scheme on the other 3 subsets of 3 nodes. After the first round of shuffling,

Node 1 recovers $(v_{1,4},v_{1,5})$, $(v_{2,4},v_{2,6})$ and $(v_{3,5},v_{3,6})$,

Node 2 recovers $(v_{1,2},v_{1,3})$, $(v_{4,2},v_{4,6})$ and $(v_{5,3},v_{5,6})$,

Node 3 recovers $(v_{2,1},v_{2,3})$, $(v_{4,1},v_{4,5})$ and $(v_{6,3},v_{6,5})$,

Node 4 recovers $(v_{3,1},v_{3,2})$, $(v_{5,1},v_{5,4})$ and $(v_{6,2},v_{6,4})$.

Similarly, as shown in Fig. 5, each of Node $2$, Node $3$, and Node $4$ multicasts two linear combinations of three locally computed segments to the other three nodes, using the same coefficients $\alpha_{1}$, $\alpha_{2}$, and $\alpha_{3}$.

Having received the above two linear combinations, each of Node $2$, Node $3$, and Node $4$ first subtracts out one segment available locally from the combinations, or more specifically, $v_{6,1}^{(1)}$ for Node $2$, $v_{5,2}^{(1)}$ for Node $3$, and $v_{4,3}^{(1)}$ for Node $4$. After the subtraction, each of these three nodes recovers the required segments from the two linear combinations. More specifically, Node 2 recovers $v_{4,3}^{(1)}$ and $v_{5,2}^{(1)}$, Node 3 recovers $v_{4,3}^{(1)}$ and $v_{6,1}^{(1)}$, and Node 4 recovers $v_{5,2}^{(1)}$ and $v_{6,1}^{(1)}$. It is not difficult to see that the above decoding process is guaranteed to be successful if $\alpha_{1}$, $\alpha_{2}$, and $\alpha_{3}$ are all distinct from each other, which requires the field size $2^{\frac{T}{2}}\geq 3$ (e.g., $T=4$). Following the similar procedure, each node recovers the required segments from the linear combinations multicast by the other three nodes. More specifically, after the second round of data shuffling,

Node 1 recovers $v_{1,6}$, $v_{2,5}$ and $v_{3,4}$,

Node 2 recovers $v_{1,6}$, $v_{4,3}$ and $v_{5,2}$,

Node 3 recovers $v_{2,5}$, $v_{4,3}$ and $v_{6,1}$,

Node 4 recovers $v_{3,4}$, $v_{5,2}$ and $v_{6,1}$.

We finally note that in the second round of data shuffling, each linear combination multicast by a node is simultaneously useful for the rest of the three nodes. $\hfill\square$

V General Achievable Scheme: Coded Distributed Computing

In this section, we formally prove the upper bounds in Theorem 1 and 2 by presenting and analyzing the Coded Distributed Computing (CDC) scheme. We focus on the more general case considered in Theorem 2 with $s\geq 1$, and the scheme for Theorem 1 simply follows by setting $s=1$.

We first consider the integer-valued computation load $r\in\{1,\ldots,K\}$, and then generalize the CDC scheme for any $1\leq r\leq K$. When $r=K$, every node can map all the input files and compute all the output functions locally, thus no communication is needed and $L^{*}(K,s)=0$ for all $s\in\{1,\ldots,K\}$. In what follows, we focus on the case where $r<K$.

In the Map phase the $\bar{N}$ input files are evenly partitioned into ${K\choose r}$ disjoint batches of size $\eta_{1}$, each corresponding to a subset $\mathcal{T}\subset\{1,\ldots,K\}$ of size $r$, i.e.,

where $\mathcal{B}_{\cal T}$ denotes the batch of $\eta_{1}$ files corresponding to the subset $\mathcal{T}$.

Given this partition, Node $k$, $k\in\{1,\ldots,K\}$, computes the Map functions of the files in $\mathcal{B}_{\cal T}$ if $k\in\mathcal{T}$. Or equivalently, $\mathcal{B}_{\cal T}\subseteq\mathcal{M}_{k}$ if $k\in\mathcal{T}$. Since each node is in ${K-1\choose r-1}$ subsets of size $r$, each node computes ${K-1\choose r-1}\eta_{1}=\frac{r\bar{N}}{K}$ Map functions, i.e., $|\mathcal{M}_{k}|=\frac{r\bar{N}}{K}$ for all $k\in\{1,\ldots,K\}$. After the Map phase, Node $k$, $k\in\{1,\ldots,K\}$, knows the intermediate values of all $Q$ output functions in the files in $\mathcal{M}_{k}$, i.e., $\{v_{q,n}:q\in\{1,\ldots,Q\},w_{n}\in\mathcal{M}_{k}\}$.

V-B Coded Data Shuffling

where $\mathcal{D}_{\cal P}$ denotes the indices of the batch of $\eta_{2}$ Reduce functions corresponding to the subset $\mathcal{P}$.

Given this partition, Node $k$, $k\in\{1,\ldots,K\}$, computes the Reduce functions whose indices are in $\mathcal{D}_{\cal P}$ if $k\in\mathcal{P}$. Or equivalently, $\mathcal{D}_{\cal P}\subseteq\mathcal{W}_{k}$ if $k\in\mathcal{P}$. As a result, each node computes ${K-1\choose s-1}\eta_{2}=\frac{sQ}{K}$ Reduce functions, i.e., $|\mathcal{W}_{k}|=\frac{sQ}{K}$ for all $k\in\{1,\ldots,K\}$.

For a subset ${\cal S}$ of $\{1,\ldots,K\}$ and ${\cal S}_{1}\subset\mathcal{S}$ with $|{\cal S}_{1}|=r$, we denote the set of intermediate values needed by all nodes in ${\cal S}\backslash\mathcal{S}_{1}$, no node outside $\mathcal{S}$, and known exclusively by nodes in $\mathcal{S}_{1}$ as $\mathcal{V}_{\mathcal{S}_{1}}^{\mathcal{S}\backslash\mathcal{S}_{1}}$. More formally:

We observe that the set $\mathcal{V}_{\mathcal{S}_{1}}^{{\cal S}\backslash\mathcal{S}_{1}}$ defined above contains intermediate values of ${r\choose|{\cal S}|-s}\eta_{2}$ output functions. This is because that the output functions whose intermediate values are included in $\mathcal{V}_{\mathcal{S}_{1}}^{{\cal S}\backslash\mathcal{S}_{1}}$ should be computed exclusively by the nodes in ${\cal S}\backslash\mathcal{S}_{1}$ and a subset of $s-(|{\cal S}|-r)$ nodes in ${\cal S}_{1}$. Therefore, $\mathcal{V}_{\mathcal{S}_{1}}^{{\cal S}\backslash\mathcal{S}_{1}}$ contains the intermediate values of a total of ${r\choose s-(|{\cal S}|-r)}\eta_{2}={r\choose|{\cal S}|-s}\eta_{2}$ output functions. Since every subset of $r$ nodes map a unique batch of $\eta_{1}$ files, $\mathcal{V}_{\mathcal{S}_{1}}^{{\cal S}\backslash\mathcal{S}_{1}}$ contains $|\mathcal{V}_{\mathcal{S}_{1}}^{{\cal S}\backslash\mathcal{S}_{1}}|={r\choose|{\cal S}|-s}\eta_{1}\eta_{2}$ intermediate values.

For each $k\in{\cal S}$, there are a total of ${|{\cal S}|-1\choose r-1}$ subsets of ${\cal S}$ with size $r$ that contain the element $k$. We index these subsets as ${\cal S}_{(k)},{\cal S}_{(k)}\ldots,{\cal S}_{(k)}[{|{\cal S}|-1\choose r-1}]$. Within a subset ${\cal S}_{(k)}[i]$, the segment associated with Node $k$ is $U_{\mathcal{S}_{(k)}[i],k}^{{\cal S}\backslash\mathcal{S}_{(k)}[i]}$, for all $i=1,\ldots,{|{\cal S}|-1\choose r-1}$. We note that each segment $U_{\mathcal{S}_{(k)}[i],k}^{{\cal S}\backslash\mathcal{S}_{(k)}[i]}$, $i=1,\ldots,{|{\cal S}|-1\choose r-1}$, is known by all nodes whose indices are in $\mathcal{S}_{(k)}[i]$, and needed by all nodes whose indices are in ${\cal S}\backslash\mathcal{S}_{(k)}[i]$.

We note that the above encoding process is the same at all nodes whose indices are in ${\cal S}$, i.e., each of them multiplies the same matrix ${\bf A}^{\cal S}$ in (16) with the segments associated with it.

Having generated the above message symbols, Node $k$ multicasts them to the other nodes whose indices are in ${\cal S}$.

When $s=1$, i.e., every output function is computed by one node, the above shuffling scheme only takes one round for all subsets ${\cal S}$ of size $|{\cal S}|=r+1$. Instead of multicasting linear combinations, every node in ${\cal S}$ can simply multicast the bit-wise XOR of its associated segments to the other $r$ nodes in ${\cal S}$. $\hfill\square$

V-B2 Decoding

For $j\in{\cal S}$ and $j\neq k$, there are a total of ${|{\cal S}|-2\choose r-2}$ subsets of ${\cal S}$ that have size $r$ and simultaneously contain $j$ and $k$. Hence, among all $n_{1}$ segments $U_{\mathcal{S}_{(k)},k}^{{\cal S}\backslash\mathcal{S}_{(k)}},U_{\mathcal{S}_{(k)},k}^{{\cal S}\backslash\mathcal{S}_{(k)}},\ldots,U_{\mathcal{S}_{(k)}[n_{1}],k}^{{\cal S}\backslash\mathcal{S}_{(k)}[n_{1}]}$ associated with Node $k$, ${|{\cal S}|-2\choose r-2}$ of them are already known at Node $j$, and the rest of $n_{1}-{|{\cal S}|-2\choose r-2}={|{\cal S}|-1\choose r-1}-{|{\cal S}|-2\choose r-2}={|{\cal S}|-2\choose r-1}=n_{2}$ segments are needed by Node $j$. We denote the indices of the subsets that contain the element $k$ but not the element $j$ as $b_{jk}^{1},b_{jk}^{2},\ldots,b_{jk}^{n_{2}}$, such that $1\leq b_{jk}^{1}<b_{jk}^{2}<\cdots<b_{jk}^{n_{2}}\leq n_{1}$, and $j\notin{\cal S}_{(k)}[b_{jk}^{i}]$ for all $i=1,2,\ldots,n_{2}$.

After receiving the symbols $X_{k}^{\cal S},X_{k}^{\cal S},\ldots,X_{k}^{\cal S}[n_{2}]$ from Node $k$, Node $j$ first removes the locally known segments from the linear combinations to generate $n_{2}$ symbols $Y_{jk}^{\cal S},Y_{jk}^{\cal S},\ldots,Y_{jk}^{\cal S}[n_{2}]$, such that

V-C Correctness of CDC

We demonstrate the correctness of the above shuffling scheme by showing that after the Shuffle phase, each node can decode all of the required intermediate values to compute its assigned Reduce functions. We use Node 1 as an example, and similar arguments apply to all other nodes. WLOG we assume that the Reduce function $h_{1}$ is to be computed by Node 1. Node 1 will need a total of ${K-1\choose r}\eta_{1}$ distinct intermediate values of $h_{1}$ from other nodes (it already knows $\frac{r\bar{N}}{K}=\bar{N}-{K-1\choose r}\eta_{1}$ intermediate values of $h_{1}$ by mapping the files in ${\cal M}_{1}$). By the assignment of the Reduce functions, there exits a subset $\mathcal{S}_{2}$ of size $s$ containing Node 1 such that all nodes in $\mathcal{S}_{2}$ need to compute $h_{1}$. Then, during the data shuffling process within each subset $\mathcal{S}$ containing $\mathcal{S}_{2}$ (note that by the definition of ${\cal V}_{{\cal S}_{1}}^{{\cal S}\backslash{\cal S}_{1}}$ in (13), the intermediate values of $h_{1}$ will not be communicated to Node 1 if ${\cal S}_{2}\nsubseteq{\cal S}$, and this is because that some node outside ${\cal S}$ also wants to compute $h_{1}$), there are ${s-1\choose|{\cal S}|-r-1}$ subsets ${\cal S}_{1}$ of ${\cal S}$ with size $|{\cal S}_{1}|=r$ such that $1\notin{\cal S}_{1}$ and ${\cal S}\backslash{\cal S}_{1}\subseteq{\cal S}_{2}$, and thus Node 1 decodes ${s-1\choose|\mathcal{S}|-r-1}\eta_{1}$ distinct intermediate values of $h_{1}$. Therefore, the total number of distinct intermediate values of $h_{1}$ Node 1 decodes over the entire Shuffle phase is

which matches the required number of intermediate values for $h_{1}$. This is also true for all the other Reduce functions assigned to Node 1.

V-D Communication Load

In the above shuffling scheme, for each subset $\mathcal{S}\subseteq\{1,\ldots,K\}$ of size $\max\{r+1,s\}\leq|\mathcal{S}|\leq\min\{r+s,K\}$, each Node $k\in{\cal S}$ communicates $n_{2}={|{\cal S}|-2\choose r-1}$ message symbols. Each of these symbols contains ${r\choose|{\cal S}|-s}\frac{\eta_{1}\eta_{2}T}{r}$ bits. Hence, all nodes whose indices are in ${\cal S}$ communicate a total of $|{\cal S}|{|{\cal S}|-2\choose r-1}{r\choose|{\cal S}|-s}\frac{\eta_{1}\eta_{2}T}{r}$ bits. The overall communication load achieved by the proposed CDC scheme is

V-E Non-Integer Valued Computation Load

For non-integer valued computation load $r\geq 1$, we generalize the CDC scheme as follows. We first expand the computation load $r=\alpha r_{1}+(1-\alpha)r_{2}$ as a convex combination of $r_{1}\triangleq\lfloor r\rfloor$ and $r_{2}\triangleq\lceil r\rceil$, for some $0\leq\alpha\leq 1$. Then we partition the set of $\bar{N}$ input files $\{w_{1},\ldots,w_{\bar{N}}\}$ into two disjoint subsets $\mathcal{I}_{1}$ and $\mathcal{I}_{2}$ of sizes $|\mathcal{I}_{1}|=\alpha\bar{N}$ and $|\mathcal{I}_{2}|=(1-\alpha)\bar{N}$. We next apply the CDC scheme described above respectively to the files in $\mathcal{I}_{1}$ with a computation load $r_{1}$ and the files in $\mathcal{I}_{2}$ with a computation load $r_{2}$, to compute each of the $Q$ output functions at the same set of $s$ nodes. This results in a communication load of

where $L_{\textup{coded}}(r,s)$ is the communication load achieved by CDC in (20) for integer-valued $r,s\in\{1,\ldots,K\}$.

Using this generalized CDC scheme, for any two integer-valued computation loads $r_{1}$ and $r_{2}$, the points on the line segment connecting $(r_{1},L_{\textup{coded}}(r_{1},s))$ and $(r_{2},L_{\textup{coded}}(r_{2},s))$ are achievable. Therefore, for general $1\leq r\leq K$, the lower convex envelop of the achievable points $\{(r,L_{\textup{coded}}(r,s)):r\in\{1,\ldots,K\}\}$ is achievable. This proves the upper bound on the computation-communication function in Theorem 2 (also the achievability part of Theorem 1 by setting $s=1$).

The ideas of efficiently creating and exploiting coded multicasting opportunities have been introduced in caching problems . In this section, we illustrated how coding opportunities can be utilized in distributed computing to slash the load of communicating intermediate values, by designing a particular assignment of extra computations across distributed computing nodes. We note that the calculated intermediate values in the Map phase mimics the locally stored cache contents in caching problems, providing the “side information” to enable coding in the following Shuffle phase (or content delivery).

For the case of $s=1$ where no two nodes are interested in computing a common Reduce function, the coded data shuffling of CDC is similar to a coded transmission strategy in wireless D2D networks proposed in , where the side information enabling coded multicasting are pre-fetched in a specific repetitive manner in the caches of wireless nodes (in CDC such information is obtained by computing the Map functions locally). When $s$ is larger than $1$, i.e., every Reduce function needs to be computed at multiple nodes, our CDC scheme creates novel coding opportunities that exploit both the redundancy of the Map computations and the commonality of the data requests for Reduce functions across nodes, further reducing the communication load. $\hfill\square$

Generally speaking, we can view the Shuffle phase of the considered distributed computing framework as an instance of the index coding problem , in which a central server aims to design a broadcast message (code) with minimum length to simultaneously satisfy the requests of all the clients, given the clients’ side information stored in their local caches. Note that while a randomized linear network coding approach (see e.g., ) is sufficient to implement any multicast communication where messages are intended by all receivers, it is generally sub-optimal for index coding problems where every client requests different messages. Although the index coding problem is still open in general, for the considered distributed computing scenario where we are given the flexibility of designing Map computation (thus the flexibility of designing side information), we prove in the next two sections tight lower bounds on the minimum communication loads for the cases $s=1$ and $s>1$ respectively, demonstrating the optimality of the proposed CDC scheme. $\hfill\square$

VI Converse of Theorem 1

In this section, we prove the lower bound on $L^{*}(r)$ in Theorem 1.

For $k\in\{1,\ldots,K\}$, we denote the set of indices of the files mapped by Node $k$ as $\mathcal{M}_{k}$, and the set of indices of the Reduce functions computed by Node $k$ as $\mathcal{W}_{k}$. As the first step, we consider the communication load for a given file assignment $\mathcal{M}\triangleq({\cal M}_{1},{\cal M}_{2}\ldots,{\cal M}_{K})$ in the Map phase. We denote the minimum communication load under the file assignment $\mathcal{M}$ by $L^{*}_{\mathcal{M}}$.

We denote the number of files that are mapped at $j$ nodes under a file assignment ${\cal M}$, as $a^{j}_{{\cal M}}$, for all $j\in\{1,\ldots,K\}$:

For example, for the particular file assignment in Fig. 6, i.e., ${\cal M}=(\{1,3,5,6\},\{4,5,6\},\{2,3,4,6\})$, $a^{1}_{\cal M}=2$ since File 1 and File 2 are mapped on a single node (i.e., Node 1 and Node 3 respectively). Similarly, we have $a^{2}_{\cal M}=3$ (Files 3, 4, and 5), and $a^{3}_{\cal M}=1$ (File 6).

For a particular file assignment ${\cal M}$, we present a lower bound on $L^{*}_{\mathcal{M}}$ in the following lemma.

$L^{*}_{\mathcal{M}}\geq\sum\limits_{j=1}^{K}\frac{a^{j}_{\cal M}}{N}\cdot\frac{K-j}{Kj}$.

Next, we first demonstrate the converse of Theorem 1 using Lemma 1, and then give the proof of Lemma 1.

Converse Proof of Theorem 1. It is clear that the minimum communication load $L^{*}(r)$ is lower bounded by the minimum value of $L^{*}_{\mathcal{M}}$ over all possible file assignments which admit a computation load of $r$:

For every file assignment ${\cal M}$ such that $|{\cal M}_{1}|+\cdots+|{\cal M}_{K}|=rN$, $\{a^{j}_{\cal M}\}_{j=1}^{K}$ satisfy

Then since the function $\frac{K-j}{Kj}$ in (24) is convex in $j$, and by (26) $\sum\limits_{j=1}^{K}\frac{a^{j}_{\cal M}}{N}=1$, (24) becomes

where (a) is due to the requirement imposed by the computation load in (27).

Then by the convexity of the function $\frac{K-j}{Kj}$ in $j$, we have for integer-valued $j=1,\ldots,K$,

where (b) is due to the constraints on $\{a^{j}_{\cal M}\}_{j=1}^{K}$ in (26) and (27).

Therefore, $L^{*}(r)$ is lower bounded by the lower convex envelop of the points $\{(r,\frac{K-r}{Kr}):r\in\{1,...,K\}\}$. This completes the proof of the converse part of Theorem 1. $\hfill\blacksquare$

Although the model proposed in this paper only allows each node sending messages independently, we can show that even if the data shuffling process can be carried out in multiple rounds and dependency between messages are allowed, the lower bound on $L^{*}(r)$ remains the same. $\hfill\square$

We devote the rest of this section to the proof of Lemma 1. To prove Lemma 1, we develop a lower bound on the number of bits communicated by any subset of nodes, by induction on the size of the subset. In particular, for a subset of computing nodes, we first characterize a lower bound on the minimum number of bits required by a particular node in the subset, which is given by a cut-set bound separating this node and all the other nodes in the subset. Then, we combine this bound with the lower bound on the number of bits communicated by the rest of the nodes in the subset, which is given by the inductive argument.

Since each message $X_{k}$ is generated as a function of the intermediate values that are computed at Node $k$, the following equation holds for all $k\in\{1,...,K\}$.

where we use “:” to denote the set of all possible indices.

The validity of the shuffling scheme requires that for all $k\in\{1,...,K\}$, the following equation holds :

For a subset $\mathcal{S}\subseteq\{1,...,K\}$, we define

which contains all the intermediate values required by the nodes in ${\cal S}$ and all the intermediate values known locally by the nodes in ${\cal S}$ after the Map phase.

For any subset ${\cal S}\subseteq\{1,\ldots,K\}$ and a file assignment ${\cal M}$, we denote the number of files that are exclusively mapped by $j$ nodes in $\mathcal{S}$ as $a^{j,\mathcal{S}}_{\cal M}$:

and the message symbols communicated by the nodes whose indices are in ${\cal S}$ as

For any subset $\mathcal{S}\subseteq\{1,...,K\}$, we have

where ${\cal S}^{c}\triangleq\{1,\ldots,K\}\backslash{\cal S}$ denotes the complement of ${\cal S}$. $\hfill\square$

a. If $\mathcal{S}=\{k\}$ for any $k\in\{1,\ldots,K\}$, obviously

b. Suppose the statement is true for all subsets of size $S_{0}$.

For any $\mathcal{S}\subseteq\{1,...,K\}$ of size $|\mathcal{S}|=S_{0}+1$ and any $k\in\mathcal{S}$, we have

For each $k\in{\cal S}$, we have the following subset version of (36) and (37).

The first term on the RHS of (52) can be lower bounded as follows.

where (a) is due to the independence of intermediate values and the fact that ${\cal W}_{k}\cap{\cal W}_{{\cal S}^{c}}=\varnothing$ (different nodes calculate different output functions), (b) and (c) are due to the independence of intermediate values, and (d) is due to the independence of the intermediate values and the fact that $|{\cal W}_{k}|=\frac{Q}{K}$.

The second term on the RHS of (52) can be lower bounded by the induction assumption:

Thus by (48), (52), (57) and (59), we have

By the definition of $a^{j,\mathcal{S}}_{\cal M}$, we have the following equations.

c. Thus for all subsets $\mathcal{S}\subseteq\{1,...,K\}$, the following equation holds:

Then by Claim 1, let $\mathcal{S}=\{1,...,K\}$ be the set of all $K$ nodes,

This completes the proof of Lemma 1. $\hfill\blacksquare$

VII Converse of Theorem 2

In this section, we prove the lower bound on $L^{*}(r,s)$ in Theorem 2, which generalizes the converse result of Theorem 1 for the case $s>1$. Since the lower bound on $L^{*}(r,1)$ in Theorem 2 exactly matches the lower bound on $L^{*}(r)$ in Theorem 1, we focus on the case $s>1$ (i.e., each Reduce function is calculated by 2 or more nodes) throughout this section.

We denote the minimum communication load under a particular file assignment $\mathcal{M}$ as $L^{*}_{\mathcal{M}}(s)$, and we present a lower bound on $L^{*}_{\mathcal{M}}(s)$ in the following lemma.

Converse Proof of Theorem 2. The minimum communication load $L^{*}(r,s)$ is lower bounded by the minimum value of $L^{*}_{\mathcal{M}}(s)$ over all possible file assignments having a computation load of $r$:

For every file assignment ${\cal M}$ such that $|{\cal M}_{1}|+\cdots+|{\cal M}_{K}|=rN$, $\{a^{j}_{\cal M}\}_{j=1}^{K}$ satisfy the same conditions as the case of $s=1$ in (25), (26) and (27).

Then by the convexity of the function $L_{\textup{coded}}(j,s)$ in $j$, we have for integer-valued $j=1,\ldots,K$,

Next, we first apply Lemma 2 to (73), then by (76), we have

where (a) is due to the constraints on $\{a^{j}_{\cal M}\}_{j=1}^{K}$ in (26) and (27).

Therefore, $L^{*}(r,s)$ is lower bounded by the lower convex envelop of the points $\{(r,L_{\textup{coded}}(r,s)):r\in\{1,...,K\}\}$. This completes the proof of the converse part of Theorem 2. $\hfill\blacksquare$

The proof of lemma 2 follows the same steps of the proof of Lemma 1, where a lower bound on the number of bits communicated by any subset of nodes, for the case of $s>1$, is established by induction.

Proof of Lemma 2. We first prove the following claim.

For any subset $\mathcal{S}\subseteq\{1,...,K\}$, we have

where $a^{j,{\cal S}}_{\cal M}$ is defined in (39). $\hfill\square$

a. If $\mathcal{S}=\{k\}$ for any $k\in\{1,\ldots,K\}$, obviously

b. Suppose the statement is true for all subsets of size $S_{0}$.

For any $\mathcal{S}\subseteq\{1,...,K\}$ of size $|\mathcal{S}|=S_{0}+1$, and all $k\in\mathcal{S}$, we have as derived in (61):

where $Y_{\mathcal{S}^{c}}=(V_{\mathcal{W}_{\mathcal{S}^{c}},:},V_{:,\mathcal{M}_{\mathcal{S}^{c}}})$.

The first term on the RHS of (82) is lower bounded by the induction assumption:

The second term on the RHS of (82) can be calculated based on the independence of intermediate values:

where (a) and (b) are due to the independence of the intermediate values, and (c) is due to the uniform distribution of the output functions such that each node in ${\cal S}$ calculates $\frac{Q}{{K\choose s}}\cdot{|{\cal S}|-1\choose s-1}$ output functions computed exclusively by $s$ nodes in ${\cal S}$.

For each $j\in\{1,\ldots,S_{0}+1\}$ in (93), we have

Since (99) holds for all subsets ${\cal S}$ of size $|{\cal S}|=S_{0}+1$, we have proven Claim 2.

Then by Claim 2, let $\mathcal{S}=\{1,...,K\}$ be the set of all $K$ nodes,

This completes the proof of Lemma 2. $\hfill\blacksquare$

VIII Implementation and Empirical Evaluation of Coded Distributed Computing

In this section, we demonstrate the impact of the proposed Coded Distributed Computing (CDC) scheme on balancing the time spent on task execution and the time spent on data movement, in order to speed up practical distributed computing applications. In particular, let us consider a MapReduce-type application for which the total execution time is roughly composed of the time spent executing the Map tasks, denoted by $T_{\textup{map}}$, the time spent shuffling intermediate values, denoted by $T_{\textup{shuffle}}$, and the time spent executing the Reduce tasks, denoted by $T_{\textup{reduce}}$, i.e.,

Using CDC, we can leverage $r\times$ more computations in the Map phase, in order to reduce the communication load by the same multiplicative factor. Hence, ignoring the coding overheads, CDC promises an approximate total execution time of

To minimize the above execution time, one would choose $r^{*}=\left\lfloor\sqrt{\tfrac{T_{\textup{shuffle}}}{T_{\textup{map}}}}\right\rfloor\textup{ or }\left\lceil\sqrt{\tfrac{T_{\textup{shuffle}}}{T_{\textup{map}}}}\right\rceil$, resulting in the minimum execution time of

For example, in an application that $T_{\textup{shuffle}}$ is $10\times$ - $100\times$ larger than $T_{\textup{map}}+T_{\textup{reduce}}$, by comparing from (101) and (103), we note that CDC can reduce the execution time by approximately $1.5\times$ - $5\times$.

In the rest of this section, we empirically demonstrate the performance gain of applying CDC to TeraSort , which is a commonly used Hadoop benchmark for distributed sorting terabytes of data . In particular, we first incorporate the coding ideas in CDC into TeraSort to develop a novel coded distributed sorting algorithm, named CodedTeraSort, which imposes structured redundancy in the input data, in order to enable in-network coding opportunities that overcome the data shuffling bottleneck of TeraSort. Then, we evaluate the performance of CodedTeraSort on Amazon EC2 clusters, and observe a $1.97\times$ - $3.39\times$ speedup, compared with TeraSort, for typical settings of interest.

TeraSort is a conventional algorithm for distributed sorting of a large amount of data. The input data that is to be sorted is in the format of key-value (KV) pairs, meaning that each input KV pair consists of a key and a value. For example, the domain of the keys can be 10-byte integers, and the domain of the values can be arbitrary strings. TeraSort sorts the input data according to their keys, e.g., sorting integers.

Let us consider implementing TeraSort over $K$ distributed computing nodes, which consists of 5 stages: File Placement, Key Domain Partitioning, Map Phase, Shuffle Phase, and Reduce Phase. In File Placement, all input KV pairs are split into $K$ disjoint files, and each file is placed on one of the $K$ nodes. In Key Domain Partitioning, the domain of the keys is split into $K$ partitions, and each node will be responsible for sorting the KV pairs whose keys fall into one of the partitions. In Map Phase, each node hashes each KV pair in its locally stored file into one of the $K$ partitions, according to its key. In Shuffle Phase, the KV pairs in the same partition are transferred to the node that is responsible for sorting that partition. In Reduce Stage, each node locally sorts KV pairs belonging to its assigned partition. We illustrate the TeraSort algorithm using a simple example shown in Fig. 7.

VIII-A2 Performance Evaluation

To understand the performance of TeraSort, we performed an experiment on Amazon EC2 to sort 12GB of data by running TeraSort on 16 instances.We note that EC2 uses virtual machines, and each instance may not be hosted by a dedicated physical machine. The breakdown of the total execution time is shown in Table I.

We observe from Table I that for a conventional TeraSort execution, 98.4% of the total execution time was spent in data shuffling, which is $508.5\times$ of the time spent in the Map phase. Given the fact that data shuffling dominates the job execution time, the principle of optimally trading computation for communication of the proposed CDC scheme can be applied to significantly improve the performance of TeraSort. For example, when executing the same sorting job using a coded version of TeraSort with a computation load of $r=10$, according to (102), we could theoretically save the total execution time by approximately $8\times$. This motivates us to develop a novel coded distributed sorting algorithm, named CodedTeraSort, which is briefly described in the next sub-section.

VIII-B Coded TeraSort

We develop the CodedTeraSort algorithm by applying the proposed CDC scheme for the case of $s=1$ (see Example 1 in Section IV for an illustration) to the above described TeraSort algorithm. CodedTeraSort exploits redundant computations on the input files in the Map phase, creating in-network coding opportunities to significantly slash the load of data shuffling. In particular, the execution of CodedTeraSort consists of following $6$ stages of operations. Here we give high-lever descriptions of these operations, and we refer the interested readers to for more detailed descriptions.

Structured Redundant File Placement. The entire input KV pairs are split into many small files, each of which is repeatedly placed on $1\leq r\leq K$ nodes (i.e., a computation load of $r$), according to the particular pattern specified by the CDC scheme.

Map. Each node applies the hashing operation as in TeraSort on each of its assigned files.

Encoding to Create Coded Packets. Each node generates coded multicast packets from local results computed in Map phase, according to the encoding process of the CDC scheme.

Multicast Shuffling. Each node multicasts each of its generated coded packet to a specific set of $r$ other nodes.

Decoding. Each node locally decodes the required KV pairs from the received coded packets.

Reduce. Each node locally sorts the KV pairs within its assigned partition as in the Reduce phase of TeraSort.

VIII-C Empirical Evaluations

We imperially demonstrate the performance gain of CodedTeraSort through experiments on Amazon EC2 clusters. In this sub-section, we first present some choices we have made for the implementation. Then, we discuss the experiment results.

We first describe the following common implementation choices that we have made for both TeraSort and CodedTeraSort algorithms.

Data Format: All input KV pairs are generated from TeraGen in the standard Hadoop package. Each input KV pair consists of a $10$-byte key and a $90$-byte value. A key is a $10$-byte unsigned integer, and the value is an arbitrary string of $90$ bytes. The KV pairs are sorted based on their keys, using the standard integer ordering.

Library: We implement both TeraSort and CodedTeraSort algorithms in C++, and use Open MPI library for communications between EC2 instances.

In the TeraSort implementation, each node sequentially steps through Map, Pack, Shuffle, Unpack, and Reduce stages. The Pack stage serializes each intermediate value to a continuous memory array to ensure that a single TCP flow is created for each intermediate value (which may contain multiple KV pairs) when MPI_Send is calledCreating a TCP flow per KV pair leads to inefficiency from overhead and convergence issue.. The Unpack stage deserializes the received data to a list of KV pairs. In the Shuffle stage, intermediate values are unicast serially, meaning that there is only one sender node and one receiver node at any time instance. Specifically, as illustrated in Fig. 8(a), Node $1$ starts to unicast to Nodes 2, 3, and 4 back-to-back. After Node $1$ finishes, Node $2$ unicasts back-to-back to Nodes 1, 3, and 4. This continues until Node 4 finishes.

In the CodedTeraSort implementation, each node sequentially steps through CodeGen, Map, Encode, Multicast Shuffling, Decode, and Reduce stages. In the CodeGen (or code generation) stage, firstly, each node generates all file indices, as subsets of $r$ nodes. Then each node uses MPI_Comm_split to initialize $\binom{K}{r+1}$ multicast groups each containing $r+1$ nodes on Open MPI, such that multicast communications will be performed within each of these groups. The serialization and deserialization are implemented respectively in the Encode and the Decode stages. In Multicast Shuffling, MPI_Bcast is called to multicast a coded packet in a serial manner, so only one node multicasts one of its encoded packets at any time instance. Specifically, as illustrated in Fig. 8(b), Node 1 multicasts to the other 2 nodes in each multicast group Node 1 is in. For example, Node 1 first multicasts to Node 2 and 3 in the multicast group $\{1,2,3\}$. After Node $1$ finishes, Node $2$ starts multicasting in the same manner. This process continues until Node $4$ finishes.

VIII-C2 Experiment Results

We evaluate the run-time performance of TeraSort and CodedTeraSort, for different combinations of the number of workers $K$ and the computation load $1\leq r\leq K$. All experiments are repeated $5$ times, and the average values are recorded.

In Table II and Table III, we list the breakdowns of the average execution times to sort 12 GB of input data using $K=16$ workers and $K=20$ workers respectively. Here we limit the incoming and outgoing traffic rates of each instance to $100$ Mbps. This is to alleviate the effects of the bursty behaviors of the transmission rates in the beginning of some TCP sessions, given the particular size of the data to be sorted. We observe an overall $1.97\times$ - $3.39\times$ speedup of CodedTeraSort as compared with TeraSort. From the experiment results we make the following observations:

For CodedTeraSort, the time spent in the CodeGen stage is proportional to $\binom{K}{r+1}$, which is the number of multicast groups.

The Map time of CodedTeraSort is approximately $r$ times higher than that of TeraSort. This is because that each node hashes $r$ times more KV pairs than that in TeraSort. Specifically, the ratios of the CodedTeraSort’s Map time to the TeraSort’s Map time from Table II are $6.03/1.86\approx 3.2$ and $10.84/1.86\approx 5.8$, and from Table III are $4.68/1.47\approx 3.2$ and $8.59/1.47\approx 5.8$.

While CodedTeraSort theoretically promises a factor of more than $r\times$ reduction in shuffling time, the actual gains observed in the experiments are slightly less than $r$. For example, for the experiment with $K=16$ nodes and $r=3$, as shown in Table II, the speedup of the Shuffle stage is $945.72/412.22\approx 2.3<3$. This phenomenon is caused by the following two factors. 1) Open MPI’s multicast API (MPI_Bcast) has an inherent overhead per a multicast group, for instance, a multicast tree is constructed before multicasting to a set of nodes. 2) Using the MPI_Bcast API, the time of multicasting a packet to $r$ nodes is higher than that of unicasting the same packet to a single node. In fact, as measured in , the multicasting time increases logarithmically with $r$.

Further, we observe the following trends from both tables:

The impact of computation load $r$: As $r$ increases, the shuffling time reduces by approximately $r$ times. However, the Map execution time increases linearly with $r$, and more importantly the CodeGen time increases exponentially with $r$ as $\binom{K}{r+1}$. Hence, for small values of $r$ ($r<6$) we observe overall reduction in execution time, and the speedup increases. However, as we further increase $r$, the CodeGen time will dominate the execution time, and the speedup decreases. Hence, in our evaluations, we have limited $r$ to be at most $5$.The redundancy parameter $r$ is also limited by the total storage available at the nodes. Since for a choice of redundancy parameter $r$, each piece of input KV pairs should be stored at $r$ nodes, we can not increase $r$ beyond $\frac{\text{total available storage at the worker nodes}}{\text{input size}}.$