Active Set in GraphX Pregel

The original Pregel computation abstraction takes a vertex-centric view to express graph programs. In this abstraction, a vertex is a two-state machine which is either inactive or active. Each active vertex concurrently runs a vertex program, which can send messages to other vertices, update the value of the vertex, or deactivates the vertex (aka. votes to halt). A deactivated vertex remains inactive in subsequent execution steps unless it receives a message. A vertex activated by a message must vote to halt again to get deactivated. No computation is associated with edges.

The GraphX computation framework realizes Pregel based on the Gather-Apply-Scatter (GAS) pattern. GAS breaks down the computation of vertex program into purely edge-parallel and vertex-parallel stages with three user-defined functions, called gather, apply and scatter: the gather function is an associative and commutative operator over the inbound messages of a vertex; the apply function operates on a vertex and updates its value based on the inbound message; the scatter function computes and sends messages along each edge. Note that, unlike more standard implementations of Pregel, GraphX Pregel does not allow modifying edge attributes and graph topology during execution. Besides, it allows message exchanges only between the neighboring vertices. These restrictions are imposed so as to increase parallelism and enable more optimizations within GraphX.
GraphX Pregel does not provide operations to deactivate a vertex (i.e. to let a vertex vote to halt). Instead, it automatically avoids executing the scatter function on edges where neither endpoint has received a message. These edges will remain skipped until their endpoints receive a message again. Hence, an edge in GraphX Pregel is also a two-state machine, just like a vertex in the original Pregel: an edge will be deactivated if its endpoints do not receive any message and will be activated after its endpoints receive a message. In practice, GraphX decides whether an edge is activated or not by bookkeeping the set of vertices that received messages from the last execution step, called the active set. Hence, an edge is activated iff it is incident to some vertex in the active set. In addition, GraphX also uses the size of the active set as an indicator of workload and adjust message delivery strategy accordingly at runtime.
Adopting the bookkeeping mechanism allows GraphX Pregel to avoid unnecessary computation and adjust its message delivery strategy at runtime. The mechanism, however, can increase the complexity to reason about the behaviors of a Pregel program. Fortunately, it turns out that enabling bookkeeping does not change the observable behaviors of a Pregel program under certain conditions. More precisely, we have the following two facts:
Fact 1. The outcome of a GraphX Pregel program is not affected by the choice of message delivery strategy if the gather function is associative and commutative.
Fact 2. The outcome of a GraphX Pregel program is not affected by the bookkeeping mechanism if the scatter function only depends on the edge triplet, namely, the edge attribute, the source vertex, and the destination vertex.
The first fact is clear. To see why the second fact is true, consider two versions of GraphX frameworks, one of which executes the scatter function only on activated edges and the other executes it on all edges. Given the condition in Fact 2 and a deterministic apply function, we will show that a Pregel program produces the same output graph in both frameworks from the same initial message and input graph. The proof is done by induction on the number of execution steps taken by the program:
Step 0. Since the apply function is deterministic and the initial message is sent to all vertices without executing the scatter function, the program produces the same graph in both frameworks after the first step.
Step n+1. Suppose the program produces the same graph after the nth step. Consider an edge e in the (n+1)-th step. If e is active, then the scatter function behaves the same on it in both frameworks. If e is inactive, then two facts about its endpoints hold in the last step: i) they did not receive any message from e, and ii) they did not change values. Since the scatter function only checks the edge triplet of e, which is unchanged since the last step, the function will not send any message along e even if it is executed on e in this step. Hence, the scatter function sends the same messages to the same vertex in both frameworks. Moreover, since the gather function is associative and commutative, each vertex receives the same reduced inbound message in both frameworks. Finally, since the apply function is deterministic, the program will produce the same graph in both frameworks after the (n+1)-th step. This concludes our proof.
Note that the condition in Fact 2 holds when the scatter function is pure. Hence, we can safely assume that all edges are activated when we are simulating or reasoning about a pure Pregel program in GraphX. In the original Pregel model, however, active set is essential to decide whether vertex program should be executed on a vertex. It is the GAS decomposition and the restriction on edge modification that enable us to eliminate the use of active set in GraphX.

References

1. Pregel API in GraphX Programming Guide and its source code on GitHub.
2.Xin, Reynold S., et al. "Graphx: A resilient distributed graph system on spark." First International Workshop on Graph Data Management Experiences and Systems. 2013.
3. Xin, Reynold S., et al. "GraphX: Unifying data-parallel and graph-parallel analytics." arXiv preprint. 2014.
4. Gonzalez, Joseph E., et al. "Graphx: Graph processing in a distributed dataflow framework." OSDI. 2014.

Correctness specification for Pregel programs

In this post, we show how to prove correctness of programs written in the Pregel abstraction by specifying loop invariants and variants for the supersteps. This post is only a sketch and will be elaborated incrementally when I have time.

In the following examples, we use $G=(V,E)$ to denote the base graph. Without loss of generality, we maintain a global counter to record the number of supersteps taken so far. Let $C_v(n)$ denote the attribute of vertex $v$ in the $n$-th Pregel iteration. A state of a Pregel program is a tuple $(k, G, \{C_v(k): v \in V\})$, where $k$ denotes the value of the global counter. We shall use $c(s)$ to denote the counter value in state $s$. An invariant for a Pregel program is a sequence of assertions $\{f_n\}_{n\ge0}$ over program states, such that each $f_k$ holds after the $k$-th superstep (i.e., $c(s)=k$ implies $f_k(s)$ is true). Moreover, $f_{k+1}$ should hold given that $f_{k}$ holds and that there are active vertices after the $k$-th superstep. A variant $V$ for a Pregel program is mapping from program states to a poset $(P,\succ)$, such that i) $P$ doesn't contain a infinite descending chain, and ii) $s_1\leadsto s_2 \wedge c(s_2)>c(s_1)$ implies $V(s_1)\succ V(s_2)$. Here we simply set this poset to $(\mathbb N\cup\{0\}, >)$.

Our first example is the connected components algorithm from the Spark library. This algorithm, after it terminates, assigns each vertex the smallest vertex id among the vertices in the same connected components. The vertex attribute $C_v(n)$ records the the lowest vertex id that vertex $v$ has found up to the $n$-th superstep. Let $C_v(0)=id(v)$  for all $v\in V$. It can be shown that
Invariant: $f_n$ $\equiv$ "$C_v(n) = \min\{C_u(n-1) : u\in N(v)\}$"
Variant: $V \equiv \sum_{v\in V} C_v(n)$.

The second example is the shortest landmark distances algorithm from the Spark library. Given a set of landmarks $L \subset V$, the algorithm computes for each vertex $v$ a mapping $C_v(n): L \rightarrow Int$, such that $C_v(n)(u)$ records the shortest distances from $v$ to $u$ that vertex $v$ has found up to the $n$-th superstep. For all $v,u\in V$, let $C_v(0)(u) = 0$ if $v=u \in L$, and $C_v(0)(u) = \infty$ otherwise. It turns out that
Invariant: $f_n$ $\equiv$ "For each $u \in L$, $C_v(n)(u)$ is the distance of the shortest paths from $v$ to $u$ with at most $n$ edges."
Variant: $V \equiv \sum_{v\in V}\sum_{u\in L} C_v(n)(u)$.

Sometimes, loop invariants and variants cannot be found as intuitively even for simple Pregel programs. Consider the following vertex coloring algorithm that colors the vertices using at most $\Delta+1$ colors, where $\Delta$ denotes the maximum degree of the base graph. Also, let the set of available colors be cyclically ordered with cardinality $\Delta+1$.

  Initialize all vertices to the same color.   while there exists $(v, u) \in E$ such that $v$ and $u$ share the same color do       Let $w$ denote the vertex from ${v, u}$ with a larger vertex id.       Set the color of $w$ to the minimal color larger than its current color.   done

Obviously, there exists a $(\Delta+1)$-coloring for the graph. Besides, the algorithm terminates only if it found a proper coloring. Now the question is: Is it possible that this algorithm doesn't terminate, say, given some order in which the vertices are colored? A variant cannot be found as easily as in the previous examples, because the relation $\succ$ is anti-reflexive but the number of colliding vertices doesn't decrease strictly as the algorithm iterates. However, after some reflection, we can still specify a variant as follows:

Variant: $V \equiv $ the maximal number of vertices that collide with their neighbors.

More precisely, after the $n$-th superstep finishes, there exists at least $n$ vertices in the graph, such that the coloring of these $n$ vertices is a subset of a proper $(\Delta+1)$-coloring of the graph. Observe that existence of $V$ here happens to assure both the termination and the correctness of this vertex coloring algorithm.

Writing decentralized algorithms with Pregel: A case study

In this post, we demonstrate how to implement a purely functional decentralized algorithm using the Pregel abstraction offered by the Spark GraphX library.

A vertex coloring of a graph is a way to color the vertices such that no two adjacent vertices share the same color. Formally, given an undirected graph $G=(V,E)$, a k-coloring of $G$ is a map $f: V \rightarrow \{1,...,k\}$ such that $f(v) \neq f(u)$ for any $(v,u) \in E$. A graph is called k-colorable if it has a $k$-coloring.

It takes linear time to find a $(\Delta+1)$-coloring for any graph with maximum degree $\Delta$. On the other hand, finding a 4-coloring for a 3-colorable graph is already NP-hard. There are many randomized algorithms in the literature that color a graph in exponential time and linear space. Among them, the Communication-Free Learning (CFL) algorithm [1] is a fully decentralized algorithm inspired by the problem of allocating non-interfering channels in IEEE 802.11 wireless networks.

The CFL algorithm works as follows. For each vertex $v \in V$, let $C_n(v) \in \{1,...,k\}$ denote the color of $v$ in the $n$-th iteration. Also, let $P_n(v,c)$ denote the probability that vertex $v$ has color $c$ in the $n$-th iteration. That is, $\Pr\{C_n(v) = c\} = P_n(v,c)$. At $n = 0$, the color of each vertex $v$ is chosen uniformly at random, i.e., $P_0(v, c) = 1/k$ for all $v \in V$ and $c = 1,...,k$. For $n \ge 1$, the color of $v$ is chosen randomly according to the following rule: If no adjacent vertex to $v$ shares the same color with $v$, then we let $P_n(v, c) = 1$ for $c = C_{n-1}(v)$ and let $P_n(v, c) = 0$ otherwise. Note that this implies $C_n(v) = C_{n-1}(v)$. If there is a collision of color between $v$ and its neighbors, then we choose $C_n(v)$ according to probability distribution $P_n(v, \cdot)$, such that for $c = 1,...,k$,
$$ P_{n}(v,c)=\begin{cases} (1-\beta)\cdot P_{n-1}(v,c), & c=C_{n-1}(v)\\ (1-\beta)\cdot P_{n-1}(v,c)+\beta\cdot(k-1)^{-1}, & \mbox{otherwise,} \end{cases} $$ where $\beta \in (0,1)$ is a pre-determined parameter. Observe that the vertex colors stabilize if and only if they form a $k$-coloring.
The CFL algorithm can be executed via the following decentralized procedure:

1. Set up $C_0(v)$ and $P_0(v,\cdot)$ for each vertex $v$ and let all vertices be active.
2. A) An active vertex $v$ is deactivated if no vertices in $N(v)$ share the same color with $v$. An inactive vertex $v$ is activated if there is a vertex in $N(v)$ sharing the same color with $v$.
   B) Each vertex $v$ computes $C_n(v)$ and $P_n(v,\cdot)$ upon the $n$-th change of the vertex state (from active to inactive or vice versa).
3. Repeat Step 2 until all vertices are simultaneously inactive.

Thus we implement the algorithm by carrying out the above procedure with Pregel. A vertex $v$ is represented as a 4-tuple

type Palette = (Color, List[Double], Boolean, Random)

composed of the vertex color $C_n(v)$, the color distribution $P_n(v,\cdot)$, the vertex state (active/inactive), and a random number generator in turn. Given a graph RDD graph, we first construct its base graph and initialize the attributes of the vertices:

val seed = Random.nextInt
val baseGraph = graph.mapVertices((id, _) => {
  val rng = new Random(seed + id)
  (rnd.nextInt(k + 1), initColorDist, true, rng)
})

Note that we choose to maintain an RNG at each vertex. One may feel tempted to declare a global RNG and share it among the vertices. However, since Spark serializes the closure as it dispatches jobs, doing so will lead each vertex to use a copy of the global RNG instead of sharing it. Hence, all vertices will obtain the same sequence of random numbers upon their queries, which is definitely an undesired situation.
We use functions sendMsg and mergeMsg to carry out Part A of the second step in the procedure:

def sendMsg (edge: EdgeTriplet[Palette, _]): Iterator[(VertexId, Boolean)] = {
  if (edge.srcAttr._1 == edge.dstAttr._1)
    return Iterator((edge.srcId, true))
  if (edge.srcAttr._3)
    return Iterator((edge.srcId, false))
  Iterator.empty
}
def mergeMsg (msg1: Boolean, msg2: Boolean) = msg1 || msg2

For each edge $(v, u)$, the sendMsg function can activate or deactivate the source vertex $v$ by sending it a Boolean message. The messages are aggregated by mergeMsg via disjunction, such that a vertex is activated if any of its neighbors wants to activate it, and is deactivated if all of its neighbors want to deactivate it.
Part B of the second step is carried out by the vprog function:

def vprog (id: VertexId, attr: Palette, active: Boolean): Palette = {
  val color = attr._1
  val dist  = attr._2
  val rng   = attr._4
  val newDist = dist.foldLeft((1, List[Double]())) {
    case ((i, list), weight) => (i + 1,
      if (active)
        list :+ (weight * (1 - beta) + (if (color == i) 0.0 else beta / (maxNumColors - 1)))
      else
        list :+ (if (color == i) 1.0 else 0.0))
  }._2
  val newColor = if (active) sampleColor(newDist, rng.nextDouble) else color
  (newColor, newDist, active, rng)
}

The vprog function uses a helper function colorSampler to color a vertex randomly according to its current color distribution:

def sampleColor (dist: List[Double], rnd: Double): Int = {
  dist.foldLeft((1, 0.0)) {
    case ((color, mass), weight) => {
      val m = mass + weight
      (if (m < rnd) color + 1 else color, m)
    }}._1
}

Finally, we invoke Pregel with the above functions to compute a $k$-coloring for the base graph baseG:

Pregel(baseGraph, true)(vprog, sendMsg, mergeMsg).mapVertices((_, attr) => attr._1)

It is shown in [2] that the CFL algorithm colors a graph in exponential time with high probability. More precisely, given any $\epsilon \in (0,1)$, the algorithm can find a $k$-coloring for a $k$-colorable graph with probability $1-\epsilon$ in $O(N\exp^{cN\lg 1/\epsilon})$ iterations, where $N$ is the number of vertices and $c$ is a constant depending on $\beta$.

References

1. D. J. Leith, P. Clifford, "Convergence of distributed learning algorithms for optimal wireless channel allocation." IEEE CDC, 2006.
2. Duffy, Ken R., et al. "Complexity analysis of a decentralised graph colouring algorithm." Information Processing Letters, 107.2, 2008.

Tips for writing Spark programs

Writing serializable closures

When you use a closure, Spark will serialize it and send it around the cluster. This means that any captured variables must be serializable. But sometimes it can't be. Consider this example:

class Foo {
  val factor = 3.14159
  val log = new Log(...) // not serializable
  def multiply(rdd: RDD[Int]) = {
    rdd.map(x => x * factor).reduce(...)
  }
}

The JVM must pass the closure to map to capture variable factor. However, since the closure contains an unserializable variable log, a NotSerializableException will be thrown at runtime. A work around here is to assign factor to a local variable:

class Foo {
  val factor = 3.14159
  val log = new Log(...) // not serializable
  def multiply(rdd: RDD[Int]) = {
    // Only factor2 will be serialized.
    val factor2 = factor 
    rdd.map(x => x * factor2).reduce(...)
  }
}

Serialization is a general issue for distributed programs written on the JVM. A future version of Scala may introduce a "serialization-safe" mechanism for defining closures for this purpose.

Preventing the driver program from OOM

1. Operations on RDDs includes transformations and actions. When the dataset is distributed over the workers, the outputs of transformations are still distributed. In contrast, actions will transfer the outputs to driver program, which can lead to OOM for large outputs. A rule of thumb here is to avoid using actions that may return unbounded outputs, such as collect, countByKey, collectAsMap, etc. If you need to probe a large RDD, try to use actions such as sample, count and take to obtain bounded results. Also, you can write outputs directly from workers to the HDFS using saveAsTextFile and then load them later.
2. Sometimes, you get an OOM Error not because your RDDs don't fit in memory, but because the working set of one of your tasks was too large. Spark’s shuffle operations (sortByKey, groupByKey, reduceByKey, join, etc) build a hash table within each task to perform the grouping, which can often be large. The simplest fix here is to increase the number of tasks, so that each task’s input set is smaller. Spark reuses one executor across many tasks with low launching costs, so you can safely increase the number of tasks to more than the number of cores in your clusters.

Avoiding imbalant join operations

1. Suppose that you want to inner join two RDDs in Spark, one is very small with $n$ partitions and the other is very large with $m\gg n$ partitions. This operation will shuffle the data in the large RDD into $n$ partitions, carried out by at most $n$ executors regardless of the number of work nodes. A better way to do inner join in this case is to first broadcasts the small RDD to the nodes on which the large RDD is distributed, and then performs the join operation at each executor. In this way, you not only avoid the need of shuffling, but also maintain the parallelism of the larger RDD (with $m$ output partitions instead of $n$). The benefits are obtained at the cost of replicating and broadcasting the smaller RDD around the cluster.
2. If you want to left join a very small RDD with a very large RDD, it may be helpful to first filter the the large RDD for only those entries that sharing keys with the small RDD. This filtering step reduces the data needed to be shuffled over the network and thus may speed up the overall execution.

Note that this kind of balancing tricks are not needed for reduce-like operations such as groupByKey and reduceByKey, where the largest parent RDD’s number of partitions will be used.

References and resources

1. The official guidelines for turning Spark
2. How-to: Tune Your Apache Spark Jobs at Cloudera
3. CME 323: Distributed Algorithms and Optimization at Stanford Univ.

Designing algorithms with Spark aggregate – Part II

In the previous post, I proposed a framework for Spark aggregate and listed several examples derived from the framework. One may observe that the $\mathrm{comb}$ functions are associative and commutative in all but the last example. One may also observe that the results of aggregate calls are deterministic in all but the last example. Now a question rises: is it necessary to use an associative and commutative $\mathrm{comb}$ function in an aggregate call, so that the result can be deterministic? The answer is (somewhat surprisingly) no. In this post, I shall give a simple guideline to find deterministic aggregate examples with non-commutative $\mathrm{comb}$ functions.

Given domain $A$ and range $B$, the idea is to find $B' \subset B$, $\mathrm{seq}: B \rightarrow A \rightarrow B'$ and $\mathrm{comb}: B \rightarrow B \rightarrow B'$, such that $\mathrm{aggregate(zero,\ seq,\ comb,}\ as)$ is deterministic w.r.t. the provided $\mathrm{zero} \in B$ and all $as \in A^+$. Moreover, $\mathrm{comb}$ is non-commutative over $B$, but the restriction of $\mathrm{comb}$ to $B'$ is associative and commutative.

The above idea can be instantiated using an abelian subgroup $(B', ⊗)$ of a non-abelian group $(B, ⊗)$. For example, we can let $B'$ be a cyclic subgroup of a symmetric group $B$. Here ⊗ is function composition and $A$ is the set of integers:

$\quad \mathrm{zero}$: an arbitrary element from $B$.
$\quad \mathrm{seq}\ (b, n) = b^n$.
$\quad \mathrm{comb}\ (b_1, b_2) = b_1 ⊗ b_2$.

Observer that the range of $\mathrm{comb}$ is confined to $B'$ due to the closure property of group operators. For another example, we can let $B$ be the group comprised of all non-singular 2-by-2 square matrices, and $B'$ be the subgroup formed by the rotation matrices in $B$. Here we set $A = B$ and let ⊗ be matrix multiplication:

$\quad \mathrm{zero}$: the identity matrix.
$\quad \mathrm{seq}\ (b, a) = b ⊗ [a]$, where $[x] = x \cdot |\det(x)|^{-1}$, the normalized matrix of $x$.
$\quad \mathrm{comb}\ (b_1, b_2) = b_1 ⊗ b_2$.

We can also construct examples using an abelian base group. We just have to "twist" the $\mathrm{comb}$ function such that it is commutative only over $B'$. For example, let $B$ be the set of n-by-n square matrices, and B' be the subset of $B'$ formed by all matrices whose last row is zero. Moreover, set $A = B$ and use matrix addition $+$ as a commutative group operator. Then we have the following example:

$\quad \mathrm{zero}$: the zero matrix.
$\quad \mathrm{seq}\ (b, a) = b + \{a\}$, where $\{x\}$ is obtained from setting the last row of ${x}$ to zero.
$\quad \mathrm{comb}\ (b_1, b_2) = b_1 + \langle b_2\rangle$, where $\langle x\rangle$ is obtained from doubling the last row of $x$.

Observe that $\mathrm{seq}$ and $\mathrm{comb}$ behave just like ordinary matrix additions over $B'$. They however become non-commutative operators when they are applied to elements outside $B'$. It turns out that we can use a textbook of elementary abstract algebra as a free source of interesting aggregate examples, at least in theory.

Designing algorithms with Spark aggregate – Part I

In this post, I shall give a simple framework to design algorithms in terms of Spark's aggregate function. This framework is just a proof of concept and should not be applied blindly in practice. However, it does induce efficient algorithms in some special cases, as will be shown in the examples.

In general, our goal is to compute a property of any given instance from a domain. Let $G$ denote the domain of all interested instances, and $C$ the domain of the desired property. For example, if $G$ denotes graphs and the property to compute is MST, then $C$ will be the set of trees. Furthermore, we assume that there exists an associative and commutative "composition" operator $\oplus : G \rightarrow G \rightarrow G$, such that an instance of interests can be decomposed into (and composed back from) sub-instances in $G$. We shall denote the desired property of instance $g$ as $f(g)$. For example, if $g$ is a graph, then $f(g)$ can be some graph property such as diameter, MST and vertex cover. Note that the the mapping $f : G \rightarrow C$ can be non-deterministic if the property is not unique w.r.t. the given instance.

To use the aggregate function, we specify zero, seg and comb as follows:
$\mathrm{zero} \in (G, C)$ is a pair comprised of the zero values from $G$ and $C$. If there are no such zero values, we can set $\mathrm{zero} = (g, f(g))$ for an arbitrary sub-instance of $g$.
$\mathrm{seq} : (G, C) \rightarrow G \rightarrow (G, C)$ takes a partially aggregated result $(g_1, f(g_1))$ and a sub-instance $g_2$. It outputs $(g_3, f(g_1 \oplus g_2))$ such that $f(g_3 \oplus g) = f(g_1 \oplus g_2 \oplus g)$ for all $g \in G$.
$\mathrm{comb} : (G, C) \rightarrow (G, C) \rightarrow (G, C)$ takes two partially aggregated results $(g_1, f(g_1))$ and $(g_2, f(g_2))$. It outputs $(g_3, f(g_1 \oplus g_2))$ such that $f(g_3 \oplus g) = f(g_1 \oplus g_2 \oplus g)$ for all $g \in G$.
The rest of the steps are standard: we first decompose the input instance into a sequence of sub-instances. The sequence is then converted to an RDD through parallelize. Finally, we invoke RDD.aggregate(zero)(seq,comb) to compute the desired property of the instance.

Remarks

1. While functions seq and comb look very similar in spec, they can use different methods to compute $f(g_3)$. Due to the partial results provided, seq is suitable to use incremental methods and comb can use (the merge part of) divide-and-conquer methods to compute the desired property. See the convex hull example below.
2. The assumption that operator $\oplus$ is associative and commutative over $G$ can be relaxed to be so only with respect to the given property $f$. For instance, suppose that $G$ is a set of finite strings and $\oplus$ is the string concatenation operator. While $\oplus$ is not commutative per se, it is commutative w.r.t. the string length property. That is, $\mathrm{strlen}(g_1 \oplus g_2) = \mathrm{strlen}(g_2 \oplus g_1)$ for all $g_1, g_2 \in G$. The framework is thus applicable to compute the length of string. See the majority algorithm below for a more non-trivial example.
3. The framework is practical only if $|g_3|$ can be significantly less than $|g_1 \oplus g_2|$, for otherwise we will have to pass almost the whole instance from node to node. In the case that $f(g_1 \oplus g_2)$ can be computed from $f(g_1)$ and $f(g_2)$, there is no need to output $g_3$, i.e., $|g_3|=0$. This special case allows us to operate on simpler domains (e.g., the type of seq becomes $C \rightarrow G \rightarrow C$) and to obtain more efficient aggregate algorithms due to less communication overheads.

Examples

The above naive framework can yield efficient algorithms if $f(g_1 \oplus g_2)$ can be computed solely based on $f(g_1)$ and $f(g_2)$. Below we give three examples that satisfy this condition. The first is a variant of the reverse-delete algorithm that finds an MST of a graph. The second is an outline of a convex hull algorithm. The last is Boyer and Moore's algorithm that finds the majority element in a list.

Finding MST. Let $G$ be the set of all weighted undirected graphs, and $C$ be the set of trees. Given a graph $g$ from $G$, we assign a unique index to each of its edges. Let $w(e)$ and $idx(e)$ denote the weight and index of edge $e$, respectively. Given two edges $e_1$ and $e_2$ of $g$, we say $e_1 < e_2$ iff either $w(e_1) < w(e_2)$, or $w(e_1) = w(e_2)$ and $idx(e_1)$ < $idx(e_2)$. Now we define an operator msf over $G$, such that msf computes a minimum spanning forest of $g_1 \cup g_2$ from $g_1$ and $g_2$ as follows:

def msf (g1, g2) = {
  var g = g1 ∪ g2
  while (g contains a cycle c)
    remove the largest edge from c
  return g
}

Let RDD(g) denote the RDD obtained from decomposing and parallelizing graph $g$. Then RDD(g).aggregate(Nil)(msf,msf) computes a minimum spanning forest of $g$, which would be a MST if $g$ is connected. Note that the result is deterministic if the assignment of indices is deterministic.

Finding convex hull. Here $G$ is the collection of finite sets in Euclidean plane and $C$ is the set of convex polygons. The convex hull of a set of points is the smallest convex polygon containing all the points. If the points are given as a sequence, we can use aggregate to find their convex hull by computing smaller convex hulls incrementally in seq and then merging them in comb to obtain the final convex hull. Both of seq and comb can be implemented to take linear time (see the links for pseudo codes).

Finding majority. Let $C$ be a set of comparable elements. Fix some element $\bot \in C$ and let $G = (C - \{\bot\})^*$ be a set of finite sequences. The majority problem is to determine in a given sequence from $G$ whether there is an element with more occurrences than all the others, and if so, to determine this element. Click here to see a linear-time algorithm that solves the problem using aggregate. Note that the aggregate invocation only finds a candidate for majority; we need a further check to assert that the candidate is indeed the majority element.

// Input:  a list of elements and an element 
// Output: a boolean value indicating whether elem is the majority of list
def check_majority (list: G, elem: C): Boolean = {
  def zero = (elem, 0)
  def op ((a: C, c: Int), b: Int) = {
    if(a != b)
      return (a, c - 1)
    else
      return (a, c + 1)
  }
  val res = list.fold(zero)(op)
  return res._2 > 0
}
// Input:  a list of comparable elements
// Output: the majority element of the list, or ⊥ if there is no such element
def find_majority (list: G): C = {
  def zero = (⊥, 0)  
  def seq ((a: C, c: Int), b: Int) = { 
    if(c == 0) 
      return (b, 1)
    if(a == b)
      return (a, c + 1)
    else
      return (a, c - 1)
  }  
  def comb ((a1: C, c1: Int), (a2: C, c2: Int)) = {
    if(a1 == a2) 
      return (a1, c1 + c2)
    if(c1 > c2)
      return (a1, c1 - c2)
    else
      return (a2, c2 - c1)
  }
  val res = list.aggregate(zero)(seq, comb)

  if(check_majority(list, res._1))
    return res._1
  else
    return ⊥
}

Deterministic aggregate semantics in Scala and Spark

The aggregate function in Scala has method signature as follows:

def aggregate[A](z: => A)(seqOp: (A, B) => A, combOp: (A, A) => A): A

The semantics of aggregate depend on the collection upon which it is invoked. When aggregate is invoked on a sequential traversable collection such as a List, it fallbacks to foldLeft and ignores the provided compOp. In this case, aggregate behaves just like the ordinary fold in typical functional programming languages such as Lisp and ML. When aggregate is invoked on a parallel traversable collection, however, it performs a two-stage aggregation over the data elements in parallel. In Scala, a parallel collection can be splitted into partitions and distributed over several cores or nodes. Invoking aggregate on a parallel collection reduces a value for the collection by first folding each partition separately using seqOp, and then merging the results one by one using combOp.

Note that the Scala API document doesn't specify the way in which the collection is partitioned, as well as the order in which the intermediate results are merged. Hence, if a programmer hopes to obtain the same output whenever aggregate is invoked on the same parallel collection, he has to make sure himself that seqOp is associative and combOp is both associative and commutative.

Aggregate v.s fold

The fold function in Scala has method signature as follows:

def fold[A1 >: A](z: A1)(combOp: (A1, A1) => A1): A1

One can observe two differences between the signatures of fold and aggregate: 1) fold only needs one operator, while aggregate needs two; 2) fold requires the result type to be a supertype of the element type, while aggregate doesn't. Note that both methods fallback to foldLeft when they are invoked on a sequential collection. One may notice that Scala's aggregate is in effect the ordinary FP fold

fold :  (A → B → A) → A → ([B] → A)

and Scala's fold is just a special form of its aggregate. The idea behind this design is to allow for parallel FP fold. In Scala, fold and aggregate are both done in two stages, namely folding and combining. For fold, the same operator is used for folding and combining, and this requires the supertype relationship between the element and the result type. Without the supertype relationship, it will take two operators to fold and combine, and this is why Scala provides aggregate that takes two operators as arguments.

Aggregate in Spark

Spark implements its own aggregate operations for RDD. One can observe from the source code of Spark that the intermediate results of aggregate are always iterated and merged in a fixed order (see the Java, Python and Scala implementations for details). Hence, we don't have to use commutative combOp to obtain deterministic results from aggregate, given that at least one of the following conditions holds:
(i) there doesn't exist an empty partition in any partitioning of the collection, or
(ii) z is an identity of combOp, i.e., for all a of type A, combOp(z,a) = combOp(a,z) = a.
Note that this observation is not assured by the official Spark API document and is thus not guaranteed to hold in the future versions of Spark. However, it is still useful if we want to model the behaviours of a Spark program formally.

Below, we shall try to formalize Spark's implementation of aggregate. While Scala is an impure functional language, we only consider pure functions here for simplicity. Let $R(L,m)$ denote an RDD object obtained from partitioning a list $L \in \mathbb{D}^*$ into $m \ge 1$ sub-lists. Suppose that we have fixed an element $zero \in \mathbb{E}$, as well as operators $seq:{\mathbb{E} \times \mathbb{D} \to \mathbb{E}}$ and $comb:{\mathbb{E} \times \mathbb{E} \to \mathbb{E}}$. An invocation of aggregate is said to be deterministic with respect to $zero$, $seq$ and $comb$ if the output of $$R(L,\cdot).\textrm{aggregate}(zero)(seq,\ comb)$$ only depends on $L$. Define a function $\textrm{foldl}:({\mathbb{E} \times \mathbb{D} \to \mathbb{E}}) \times \mathbb{E} \times \mathbb{D}^* \to \mathbb{E}$ by \begin{eqnarray*} \textrm{foldl}(f,\ a,\ Nil) & = & a \\ \textrm{foldl}(f,\ a,\ b \!::\! bs) & = & \textrm{foldl}(f,\ f(a,b),\ bs), \end{eqnarray*} where $f\in {\mathbb{E} \times \mathbb{D} \to \mathbb{E}}$, $a \in \mathbb{E}$, $b \in \mathbb{D}$ and $bs \in \mathbb{D}^*$. Suppose that $1\le m \le |L|$ and $R(L,m)$ partitions $L$ into $\{L_1, \dots, L_m\}$ such that $L=L_1 \cdots L_m$. According to the source code of Spark, we can define the aggregate operation in Spark as \begin{eqnarray*} R(L,m).\textrm{aggregate}(zero)(seq,\ comb) = \textrm{foldl}(comb,\ zero,\ L'), \end{eqnarray*} where $L' \in \mathbb{E}^*$ is a list of length $m$ and $L'[i] = \textrm{foldl}(seq,\ zero,\ L_i)$ for $i=1,\dots,m$.

Note that the partitioning of $L$ is unspecified and may be non-deterministic. Hence, we have to use an associative $comb$ operator if we want to ensure that the result of aggregation is deterministic. Suppose that $comb$ is associative and at least one of conditions (i) and (ii) is satisfied. It turns out that the output of aggregate is deterministic with respect to $zero$, $seq$ and $comb$ if and only if for all lists $L_1$ and $L_2$, $$comb(zero,\ \textrm{foldl}(seq,\ zero,\ L_1L_2)) = comb(\textrm{foldl}(seq,\ zero,\ L_1),\ \textrm{foldl}(seq,\ zero,\ L_2)).$$ While the above condition is undecidable in general, it can be effectively checked for some classes of programs such as FSMs over finite alphabets.

Concluding remarks

Aggregate in Spark is the parallelized version of the conventional fold, which was already shown to be extremely useful in functional programming. It is therefore of practical interests to study the properties and behaviours of the aggregate function. In particular, it is interesting to establish and verify the conditions under which aggregate is deterministic with respect to the provided $zero$, $seq$ and $comb$. We may further investigate this problem in the coming posts.

Emulating Hadoop in Spark

The map and reduce functions in MapReduce have the following general form:

Map: (K₁, V₁) → list(K₂, V₂)
Reduce: (K₂, list(V₂)) → list(K₃, V₃)

When a MapReduce program is executed, the Map function is applied in parallel to every pair in the input dataset. Each invocation of map produces a list of pairs. The execution framework then collects all pairs with the same key from all lists and groups them together, creating one group for each key. After that, the Reduce function is applied in parallel to each group, which in turn produces a collection of key-value pairs. In Hadoop's implementation of MapReduce, the values in the same group are sorted before they are sent to a reducer.

Despite the suggestive naming, map and reduce operations in Spark do not directly correspond to the functions of the same name in MapReduce. For example, Map and Reduce functions in MapReduce can return multiple key-value pairs. This feature can be implemented by the flatMap operation in Spark (and typical FP languages in general):

List(1, 2, 3).map(a => List(a))     // List(List(1), List(2), List(3))
List(1, 2, 3).flatMap(a => List(a)) // List(1, 2, 3)

One naive attempt to emulate Hadoop MapReduce (without combiners) in Spark is to use two flatMap operations for map and reduce, separated by a groupByKey and a sortByKey to perform shuffle and sort:

val input: RDD[(K1, V1)] = ...
val mapOutput: RDD[(K2, V2)] = input.flatMap(mapFn)
val shuffled: RDD[(K2, Iterable[V2])] = mapOutput.groupByKey().sortByKey()
val output: RDD[(K3, V3)] = shuffled.flatMap(reduceFn)

Here the key type K2 needs to inherit from Scala’s Ordering type to satisfy the signature of sortByKey. The above code is useful to demonstrate the relationship between Hadoop and Spark, but it should not be applied blindly in practice. For one thing, its semantics are slightly different from Hadoop's MapReduce, since sortByKey performs a total sort over all key-value pairs while Hadoop sorts within groups. This issue can be avoided by using repartitionAndSortWithinPartitions in Spark 1.2 to perform a partial sort:

val shuffled: RDD[(K2, Iterable[V2])] = mapOutput
    .groupByKey()
    .repartitionAndSortWithinPartitions(Partitioner.defaultPartitioner(mapOutput))

The improvement is unfortunately still not as efficient, since Spark uses two shuffles (one for the groupByKey and one for the sort) while MapReduce only uses one. It turns out one cannot precisely and efficiently capture the semantics of Hadoop's MapReduce in the current version of Spark.

Rather than trying to reproduce MapReduce in Spark, it is better to use only the operations that you actually need. For example, if you don’t need keys to be sorted, you can omit the sortByKey call to avoid sorting (which is impossible in regular Hadoop MapReduce). Similarly, groupByKey is too general in most cases. If shuffling is needed only for aggregating values, you should use reduceByKey, foldByKey, or aggregateByKey, which are more efficient than groupByKey since they exploit combiners in the map task. Finally, flatMap may not always be needed either. Instead, you can use map if there is always one return value, or use filter if there is zero or one.

Consult these two articles on Cloudera for details of translating Hadoop programs to efficient Spark programs: How-to: Translate from MapReduce to Apache Spark, part 1 and part 2.

Further readings

1. Jimmy Lin, Chris Dyer. Data-Intensive Text Processing with MapReduce.
2. Tom White. Hadoop: The Definitive Guide, 4th Edition.
3. Srinivasa, K.G., Muppalla, Anil K. Guide to High Performance Distributed Computing.