PaperlayerView exact revision on arXiv

Source-linked AI summary

A Fundamental Tradeoff between Computation and Communication in Distributed Computing

Songze Li, Mohammad Ali Maddah-Ali, Qian Yu, A. Salman Avestimehr

arXiv:1604.07086v2cs.ITcs.DC

TL;DR

The paper asks whether coding can reduce communication and speed distributed computing, where data shuffling can consume substantial execution time. It characterizes an inverse computation–communication tradeoff through Coded Distributed Computing (CDC), whose communication reduction is optimal under the stated framework.

  • Problem

    Data shuffling can consume substantial distributed-computing execution time, motivating whether coding can reduce communication and speed computation.

  • Method

    CDC increases Map computation by evaluating each function at r carefully chosen nodes, then uses coded multicast during shuffling to exploit the resulting redundancy.

  • Results

    CDC reduces communication load inversely with computation load, achieving an optimal 1/r gain that cannot be improved by another data-shuffling scheme.

  • Takeaways & Limitations

    Theorem 1 provides an analytical framework for trading Map-phase computation power for Shuffle-phase bandwidth to minimize execution time when applications are data-shuffling limited.

  • Takeaways & Limitations

    The redundancy parameter r is constrained by storage, and practical systems may lack computation-aware file placement; existing replicas can still enable coded multicast.

Abstract

from arXiv · show

How can we optimally trade extra computing power to reduce the communication load in distributed computing? We answer this question by characterizing a fundamental tradeoff between computation and communication in distributed computing, i.e., the two are inversely proportional to each other. More specifically, a general distributed computing framework, motivated by commonly used structures like MapReduce, is considered, where the overall computation is decomposed into computing a set of "Map" and "Reduce" functions distributedly across multiple computing nodes. A coded scheme, named "Coded Distributed Computing" (CDC), is proposed to demonstrate that increasing the computation load of the Map functions by a factor of $r$ (i.e., evaluating each function at $r$ carefully chosen nodes) can create novel coding opportunities that reduce the communication load by the same factor. An information-theoretic lower bound on the communication load is also provided, which matches the communication load achieved by the CDC scheme. As a result, the optimal computation-communication tradeoff in distributed computing is exactly characterized. Finally, the coding techniques of CDC is applied to the Hadoop TeraSort benchmark to develop a novel CodedTeraSort algorithm, which is empirically demonstrated to speed up the overall job execution by $1.97\times$ - $3.39\times$, for typical settings of interest.

I. INTRODUCTION

The paper characterizes an optimal inverse tradeoff between Map computation and Shuffle communication in a MapReduce-like framework. Its CDC scheme matches an information-theoretic lower bound and yields practical speedups on Hadoop TeraSort.

  • Framework: The framework decomposes computation into Map processing of input data and Reduce functions that use exchanged intermediate values to produce outputs.Nodes locally generate intermediate values during Map, exchange them during Shuffle, and compute final results using Reduce functions.
  • Motivation: Data shuffling can consume substantial execution time, motivating coding methods that reduce communication in distributed computing.The cited examples report 33% of job time for a Facebook Hadoop workload and 70% for an Amazon EC2 self-join.
  • CDC scheme: CDC repeats Map computations at r carefully chosen nodes to create coded multicast opportunities that reduce communication by the same factor.The scheme uses structured computation placement and coded data shuffling; the computation load r is the normalized average replication of Map computations.
  • Tradeoff: For r = 2, CDC reduces communication load by 55.6%, whereas the uncoded scheme reduces it by 11.1%.The coded gain remains significant for large K, while the uncoded gain vanishes as the number of nodes grows.
  • Optimality: The CDC communication load is optimal because it matches an information-theoretic lower bound applying to any data shuffling scheme.For general computation loads, the optimal tradeoff is given by the lower convex envelope of the achievable CDC points.

III. MAIN RESULTS

The paper characterizes the optimal computation–communication tradeoff for distributed computing and extends the result to a cascaded framework. Coded Distributed Computing achieves the tradeoff, while converse results show that no scheme can require less communication.

  • For general 1 ≤ r ≤ K, L∗(r) is the lower convex envelope of the points {(r, Lcoded(r)) : r ∈ {1, . . . , K}}.
  • CDC achieves the communication points used to define the optimal function, and a converse proves that no other scheme can achieve a smaller communication load.
  • Theorem 1 exactly characterizes the optimal tradeoff between computation load and communication load in the considered distributed computing framework.
  • Theorem 2 characterizes the computation–communication function L∗(r, s) for the cascaded distributed computing framework.
  • For general 1 ≤ r ≤ K, the cascaded function L∗(r, s) is the lower convex envelope of {(r, Lcoded(r, s)) : r ∈ {1, . . . , K}}.
  • For fixed s, the Theorem 2 communication load outperforms a linear computation–communication relationship and is superlinear with respect to r.

IV. ILLUSTRATIVE EXAMPLES: CODED DISTRIBUTED COMPUTING

The CDC examples show how coded shuffling converts additional Map computation into lower communication through XORs and linear combinations. For s = 1, XOR multicasting achieves a 2× communication gain; for s > 1, segmented intermediate values enable linear-combination-based decoding.

  • Example 1 (CDC for s = 1): CDC increases Map redundancy so nodes need fewer missing intermediate values during shuffling.With r = 2, each file is mapped on two nodes, reducing the remaining demands at each node compared with r = 1.
  • Example 1 (CDC for s = 1): 2× gain: XOR multicasting simultaneously satisfies both receiving nodes’ intermediate-value demands in the r = 2 example.A node cancels the locally known intermediate value from the XOR and recovers the needed value.
  • Example 1 (CDC for s = 1): For s = 1, CDC uses bit-wise XORs, with each output function computed at one node.The example assigns the three Reduce functions to three nodes and uses one-round coded shuffling.
  • Example 2 (CDC for s > 1): For s > 1, CDC splits intermediate values and multicasts linear combinations to support decoding at multiple nodes.In the K = 4, r = 2, s = 2 example, shuffling uses two rounds: XORs to pairs, then linear combinations of two segments to the other nodes.
  • Example 2 (CDC for s > 1): Distinct coefficients allow nodes to subtract locally known segments and recover the required unknown segments.The decoding succeeds when α1, α2, and α3 are distinct, requiring a field size of at least 3.
  • Example 2 (CDC for s > 1): Each linear combination sent in the second round is simultaneously useful to the other three nodes.This multicast structure lets one transmission serve all remaining nodes in the four-node example.

V. GENERAL ACHIEVABLE SCHEME: CODED DISTRIBUTED COMPUTING

The general CDC construction assigns files to r-node subsets, so each Map function is evaluated at r nodes with balanced computation across the system. The construction begins with sufficiently many files and partitions them symmetrically among node subsets.

  • General construction: The general CDC scheme is analyzed for s ≥1, with the s = 1 scheme obtained as a special case.The paper presents and analyzes the more general achievable construction before specializing it.
  • General construction: When r = K, every node maps all input files, computes all output functions locally, and communication is zero.The resulting communication load is L*(K, s) = 0 for every s ∈ {1, . . . , K}.
  • Map-phase assignment: The construction assumes a sufficiently large number of input files and adds empty files to obtain the required total.The padded file count is asymptotically equivalent to the original count.
  • Map-phase assignment: Files are partitioned into disjoint batches indexed by r-node subsets, and each batch is mapped by exactly the nodes in its indexing subset.This creates a symmetric assignment of files to Map nodes.
  • Map-phase assignment: Each node maps rN̄/K files and stores intermediate values for all Q output functions on those files.The balanced assignment gives every node the same Map workload.

B. Coded Data Shuffling

CDC organizes Reduce functions across s-node subsets and groups intermediate values by which nodes need them and which nodes uniquely know them. It then segments these values so coded multicast messages can serve multiple nodes.

  • Reduce assignment: The construction uses a symmetric Reduce assignment across the K nodes.Each node participates in the same number of Reduce-function assignments under the symmetric design.
  • Reduce assignment: Reduce functions are evenly partitioned into batches indexed by unique subsets of s nodes.Every node in the indexing subset computes the corresponding Reduce-function batch.
  • Intermediate-value grouping: V_S\S1 contains intermediate values needed by nodes in S\S1, known by S1, and unavailable outside S.The set is defined for S1 ⊂ S with |S1| = r.
  • Intermediate-value grouping: For each relevant node subset, CDC concatenates the grouped intermediate values and evenly splits them into r segments.Each segment is associated with one node in S1 and can be known by one subset of nodes while needed by its complement.
  • Intermediate-value grouping: The segment assignment makes each segment known to the nodes in one r-subset and needed by the nodes outside that subset.This placement creates the side-information pattern used by the coded shuffling scheme.

1) Encoding:

CDC encodes the segments associated with each node subset using common linear transformations and multicasts the resulting symbols within that subset. Decoding removes locally known segments and solves for the desired ones under a coefficient condition.

  • Encoding: CDC runs shuffling rounds over node subsets whose sizes range from max{r + 1, s} to min{r + s, K}.Each subset size defines a coded multicast stage in the general construction.
  • Encoding: Each node computes message symbols from its associated segments using a shared linear-encoding matrix.The same matrix is used by every node in the subset, with coefficients selected from a finite field.
  • Encoding: Each node multicasts its encoded symbols to the other nodes in the same subset.The transmission is designed for simultaneous use by the receiving nodes.
  • Encoding: When s = 1, the general construction reduces to one round over subsets of size r + 1 using bit-wise XORs.Linear combinations can be replaced by XORs of associated segments in this special case.
  • Decoding: A receiving node removes locally known segments before decoding the remaining desired segments from the received symbols.The relevant coefficient submatrix must be invertible for this step to succeed.
  • Decoding: Successful decoding requires distinct coefficients in the finite field, together with sufficiently large segment length.The paper states that the coded shuffling scheme can deliver all required intermediate values under this condition.

C. Correctness of CDC

CDC’s shuffle scheme lets nodes decode all intermediate values needed for their assigned Reduce functions while using coded multicast messages created by repetitive Map assignments. For non-integer computation loads, convex combinations of integer-load CDC schemes remain achievable.

  • Correctness: CDC’s shuffle scheme enables each node to decode every intermediate value required for its assigned Reduce functions.The correctness argument verifies this for Node 1 and states that the same reasoning applies to all nodes and Reduce functions.
  • Coded shuffling: For each relevant node subset S, CDC sends coded messages that simultaneously deliver intermediate values to multiple nodes.Each node in S communicates n2 message symbols, and the symbols carry r bits.
  • Non-integer computation load: For non-integer computation load r, partitioning files between floor(r) and ceiling(r) loads achieves the corresponding convex combination of CDC communication loads.The resulting points on connecting line segments are achievable, yielding the lower convex envelope for general r.
  • Coded shuffling: CDC creates these multicast opportunities by mapping data blocks repetitively at r distinct nodes.The repeated Map computations provide side information that supports coding during data shuffling.
  • Coding perspective: CDC’s shuffle can be viewed as an index-coding instance because Map computation determines the side information available for coded transmissions.The paper uses this perspective to motivate tight communication lower bounds.

VI. CONVERSE OF THEOREM 1

The converse proof lower-bounds communication for every file assignment with computation load r, then optimizes over assignments to obtain the lower convex envelope of the integer-load tradeoff points. The argument uses subset communication bounds proved by induction.

  • Converse setup: The converse lower-bounds the minimum communication load by optimizing over all file assignments that realize computation load r.For a fixed assignment M, the proof defines its minimum communication load and then minimizes this quantity over admissible assignments.
  • Converse setup: The proof tracks a_j, the number of files mapped at exactly j nodes, under the total computation-load constraint.These assignment counts are used to express and constrain the communication lower bound.
  • Non-integer loads: For non-integer r, the converse uses the line joining the points at floor(r) and ceiling(r) to lower-bound the tradeoff.Convexity of the relevant function yields the lower convex envelope of the integer computation-load points.
  • Scope of converse: The resulting converse remains valid even when shuffling uses multiple rounds and messages may depend on one another.This extension preserves the same lower bound on L*(r).
  • Inductive lower bound: A subset communication lower bound is established by induction on subset size, combining a cut-set bound with the contribution from the remaining nodes.The induction begins with singleton subsets and extends to larger subsets.

VII. CONVERSE OF THEOREM 2

Theorem 2’s converse extends the lower-bound argument to the case where each Reduce function is computed by multiple nodes. It proves that the lower convex envelope of CDC’s achievable points lower-bounds the optimal communication load.

  • Scope: Theorem 2 considers s > 1, where each Reduce function is calculated by two or more nodes.The proof generalizes the converse result for the case s = 1.
  • Fixed-assignment bound: For a fixed file assignment, the proof defines the minimum communication load L*M(s) and lower-bounds it using subset-based inequalities.The assignment must satisfy the same computation-load constraints used in the single-Reduce-node case.
  • Non-integer loads: For non-integer computation load r, convexity of Lcoded(j, s) gives the line-segment lower bound between the floor(r) and ceiling(r) points.This establishes the lower convex envelope of the CDC points for general r.
  • Conclusion: The converse shows that L*(r, s) is lower-bounded by the lower convex envelope of {(r, Lcoded(r, s))}.This completes the converse part of Theorem 2.
  • Inductive lower bound: The lower-bound proof for s > 1 uses induction over node subsets and accounts for output functions computed exclusively by nodes in a subset.The argument relies on independence of intermediate values and the distribution of output functions.

VIII. IMPLEMENTATION AND EMPIRICAL EVALUATION OF CODED DISTRIBUTED COMPUTING

CDC is evaluated as a way to balance Map computation against data-shuffling time in MapReduce applications. Applied to Hadoop TeraSort, it produces CodedTeraSort, which reports substantial speedups in typical settings.

  • Execution-time tradeoff: CDC trades r× more Map computation for an r× reduction in communication load, ignoring coding overheads.The resulting approximate execution-time expression includes the reduced shuffle term and the unchanged Reduce time.
  • Execution-time tradeoff: When shuffling takes 10×–100× more time than Map plus Reduce execution, CDC is estimated to reduce execution time by approximately 1.5×–5×.This estimate compares the baseline and CDC execution-time expressions.
  • CodedTeraSort: CodedTeraSort applies CDC to Hadoop TeraSort by adding structured redundancy to input data and enabling in-network coding during shuffling.The algorithm targets TeraSort’s data-shuffling bottleneck.
  • Empirical evaluation: 1.97×–3.39× speedup is observed for CodedTeraSort compared with TeraSort on Amazon EC2 clusters in typical settings of interest.The reported evaluation concerns the overall job execution time.

A. TeraSort

TeraSort is a distributed algorithm for sorting large datasets of key-value pairs according to their keys.

  • TeraSort sorts large amounts of key-value pairs using their keys as the ordering criterion.

1) Algorithm Description:

TeraSort partitions key-value data across worker nodes, shuffles partition-specific records to designated reducers, and sorts each partition locally. CodedTeraSort adds structured redundancy and coded multicast shuffling to reduce communication, while retaining the MapReduce-style workflow.

  • TeraSort: TeraSort divides execution into File Placement, Key Domain Partitioning, Map, Shuffle, and Reduce stages.Input key-value pairs are split into disjoint files, and key domains are divided into K partitions.
  • TeraSort: In the Shuffle phase, records for each key partition computed across nodes are fetched to its corresponding node for local sorting.With K = 4, the key domain is partitioned into [0, 25), [25, 50), [50, 75), and [75, 100].
  • Performance Evaluation: The baseline experiment sorts 12GB on 16 EC2 instances, and conventional TeraSort spends 98.4% of total execution time in data shuffling.The shuffling time is reported as 508.5× the Map-phase time.
  • CodedTeraSort: CodedTeraSort repeatedly places each small input file on r nodes, increasing Map computation to create coded multicast opportunities during shuffling.Its stages include structured redundant placement, Map, coded-packet generation, multicast shuffling, decoding, and Reduce.
  • Implementation: CodedTeraSort uses Open MPI for communication, with serial unicast in TeraSort replaced by serial multicast of coded packets.Both implementations are written in C++ and use Open MPI between EC2 instances.

2) Experiment Results:

Experiments show that CodedTeraSort substantially accelerates TeraSort by trading increased Map computation for reduced shuffling time. The benefit depends on computation load and worker count, with CodeGen overhead eventually limiting gains.

  • Experiment Results: 1.97×–3.39×: CodedTeraSort speeds up sorting 12 GB compared with TeraSort using 16 or 20 workers at 100 Mbps.Experiments were repeated five times and report average execution times.
  • Impact of computation load r: 3.2× and 5.8×: CodedTeraSort Map-time ratios relative to TeraSort are reported for the evaluated settings.These ratios are computed from the Map times in Tables II and III.
  • Impact of computation load r: As computation load r increases, shuffling time decreases by approximately r times, while Map time increases linearly with r.For K = 16 and r = 3, the measured Shuffle-stage speedup is approximately 2.3× rather than 3×.
  • Impact of worker number K: As K increases, speedup decreases because the number of multicast groups grows and fixed-r operation raises the communication load.
  • Scope and limitations: The redundancy parameter r is bounded by total worker-node storage because each input piece must be stored on r nodes.
  • Scope and limitations: The framework assumes a single-layer network with uniform multicast cost, whereas heterogeneous nodes, asymmetric tasks, and structured topologies require extensions.The paper also identifies multi-stage computation and practical data placement as follow-up settings.
Loading 1604.07086v2…