I referenced Prof Gonzalez’s slides, which is very good.


Graphs are essential to data-mining/machine learning because a lot of human processes could be modeled as a graph, and the relationships encode many interesting information such as influence, interest etc. There is also the question of where traditional systems fail graph processing:

  • MPI: users have to solve the low level data flow problems
  • MapReduce: cannot really iterate on a graph processing
  • DAG Abstractions
    • e.g. Dryad: unaware of graph semantics. For background about what Dryad does: from Microsoft website: “A Dryad programmer writes several sequential programs and connects them using one-way channels. The computation is structured as a directed graph: programs are graph vertices, while the channels are graph edges. A Dryad job is a graph generator which can synthesize any directed acyclic graph. These graphs can even change during execution, in response to important events in the computation.”
    • Issue: also do not support iterative processes and cannot dynamically prioritize computation.

Some special behaviors with graph processing:

  • dynamic computation: some nodes converges more quickly than others and could be updated less frequently.
  • serializability for graphs are conditional on the graph structure.


“Pregel is a bulk synchronous message passing abstraction in which all vertex-programs run simultaneously in a sequence of super-steps.”

Distributed GraphLab

PowerGraph comes from the lineage of research behind GraphLab, so hre I briefly summarize the main points.

  • vertex centric shared memory programming model
    • Pregel’s vertex programs send messages whereas graphLab directly reads neighbor state
    • this is helpful because the system now could choose how to move the data around (more declarative!)
    • as a result they have “ghosts” which caches the set of vertices and edges adjacent to the partition boundary
  • asynchronous vertex scheduling (with pluggable schedules)
  • automatically enforced serializability: helpful for debugging. Below is from the most strict to least strict:
    • full consistency: customized locking
    • edge consistency: use graph coloring algorithm
    • vertex consistency: each note just need to lock for its execution


What motivated PowerGraph is that GraphLab suffers when there are changes to neighbors of high degree vertices, and this creates substantial network traffic. PowerGraph deals with the issue of Zipfian distributions in real world graphs. For instance, there are a small number of twitter handles with billions of followers, and billions of accounts with very few followers. These are challenging because:

  • sequentially process edges
  • access pattern: (pregel) sends many messages or (graphlab) touch a large fraction of graph
  • syncrhonizaation: graph lab locks (async) and pregel has stragglers (sync)
  • meta data too large for single machine

Specifically for data partitioning, “When the graph is difficult to partition, both GraphLab and Pregel resort to hashed (random) vertex placement. While fast and easy to implement, hashed vertex placement cuts most of the edge”

PowerGraph Abstraction

GAS Vertex-Programs

Gather Apply Scatter framework to distribute the work of one vertex:

  • get information from the neighbors,
  • then apply the values,
  • and finally update adjacent edge data and neighbors. This is extended from the former “update” functions. This requires an “accumulator” (standard merge function)

Delta Caching


In many cases a vertex-program will be triggered in response to a change in a few of its neighbors. The gather operation is then repeatedly invoked on all neighbors, many of which remain unchanged, thereby wasting computation cycles.

What is it?

Intuitively, ∆a acts as an additive correction on-top of the previous gather for that edge.

How it’s used?

When executing the vertex-program on v the PowerGraph engine uses the cached av if available, bypassing the gather phase.

Initiating Future Computation

The PowerGraph engine maintains a set of active vertices on which to eventually execute the vertex-program. The user initiates computation by calling Activate(v) or Activate all()

Scheduling is done by PowerGraph, and the only guarantee is that all activated vertices are eventually executed.

Distributed Graph Placement

Balanced p-way Vertex-Cut

Partition by cutting vertices instead of edges.

Why? this allows high degree vertex to be split across multiple partitions: “for any edge-cut we can directly construct a vertex-cut which requires strictly less communication and storage.”

What does it mean to partition by vertex? Powergraph evenly assigns edge to machines so that no edge information is repeated (and do not need to be communicated). Whereas Distributed GraphLab evenly assigns vertices to machines (and try to assign edges connected to the vertex to the machine).

  • An example! (a,b), (a,c), (c,d), (b,d) over 2 machines
    • for vertex cuts, we partition the edges, so machine 1: (a,b), (a,c), a, b, c and machine 2: (c,d) (b,d), b, c, d — so we need to synchronize b and c
    • for edge cuts, we partition the vertices, so machine 1: a, b, (a,b), (a,c), (b,d), and machine 2: c, d, (a,c), (b,d), (c,d) — so we duplicate (a,c) and (b,d). The issue here is that all the edge computation now need to fetch vertex data from remote and block and wait, whereas vertex cuts could just merge asynchronously and not block as much! #win!

Greedy Vertex-Cuts

We can improve upon the randomly constructed vertex-cut by de-randomizing the edge-placement process.

Random partitioning is bad! Because there will be a lot of edges that are cut (“An edge is cut if both vertices are randomly assigned to different machines. The probability that both vertices are assigned to different machines is 1−1/p.”). If a vertex has already been assigned, try to place the edges connected in the same machine. The assignment algorithm works as follows (based on heuristics), as described in the paper:

  • Case 1: If A(u) and A(v) intersect, then the edge should be assigned to a machine in the intersection.
  • Case 2: If A(u) and A(v) are not empty and do not intersect, then the edge should be assigned to one of the machines from the vertex with the most unassigned edges.
  • Case 3: If only one of the two vertices has been assigned, then choose a machine from the assigned vertex.
  • Case 4: If neither vertex has been assigned, then assign the edge to the least loaded machine.


  • Works for many algorithms
  • Performs better than hand crafted code

Comments and Questions

  • I wonder what would happen if we just disregarded the notes with few connections – I think it would depend on the program semantics but with extreme distributions there should be problems when its safe to disregard or sample.
  • Frank McSherry et al wrote a paper “Scalability! But at what COST?” that questions the need for a lot of the graph processing systems. They argue that better algorithms could run sufficiently fast on a single thread. Though the primary goal of that paper is less to question the system and more to question the baseline that the evaluations these papers use.