Graph processing with GraphX

Part 5Hortonworks · Spark crash course: GraphXSpark GraphX Tutorialedureka! · GraphX tutorialSpark GraphX OperationsSkillspeed · GraphX operatorsGoogle’s PageRank Algorithm Explained with ExamplesPolo Club of Data Science @ Georgia Tech · PageRank with examples

Why graphsSearch on YouTubeBookmarkSaved

Many datasets fit a table of rows and columns. Some datasets fit a graph better. The Internet is a graph of computers, routers and switches. The Web is a graph of pages joined by hyperlinks. Social networks, airports and train terminals are graphs too.

A graph gives an intuitive model for linked data. Special graph algorithms process such data fast. Spark GraphX is the Spark library for large-scale graph data.

Spark GraphXSearch on YouTubeBookmarkSaved

GraphX is the Spark API for graphs and graph-parallel computation. It extends the Spark RDD with a Resilient Distributed Property Graph. A computation sees the same data as a graph and as a collection.

The property graph is a directed multigraph. Every vertex and every edge carries user-defined properties. Parallel edges allow several relationships between the same two vertices.

GraphX featuresSearch on YouTubeBookmarkSaved

• Flexibility. One engine does exploratory analysis, ETL and iterative graph computation. You can view data as graphs and as collections. You can transform and join graphs with RDDs. The Pregel API supports custom iterative algorithms.
• Speed. GraphX is fast against other graph systems. It keeps Spark fault tolerance and ease of use. It also includes graph builders to simplify graph analytics tasks.
• Algorithm library. PageRank, connected components, strongly connected components and triangle count ship with the library. The library also covers centralities (degrees, collectNeighborIds).

Property graphsSearch on YouTubeBookmarkSaved

A property graph stores data on the vertices and on the edges. Each vertex has properties (attributes). Each edge has a label and properties. The slides use a Twitter social network as the example: users are vertices and follows edges link a follower to the followed user.

Formally, it is a directed multigraph with user-defined objects on every vertex and edge. It allows parallel edges between the same source and destination vertex. Each vertex key is a 64-bit VertexId.

Like an RDD, a property graph is immutable, distributed and fault-tolerant. GraphX spreads the graph over the cluster with vertex-partitioning heuristics.

Collaborator graphSearch on YouTubeBookmarkSaved

The slides build a graph of GraphX collaborators. The vertex property holds username and occupation. The edge property holds the relationship string.

val users: RDD[(VertexId, (String, String))] = sc.parallelize(Seq(
  (3L, ("rxin", "student")),
  (7L, ("jgonzal", "postdoc")),
  (5L, ("franklin", "prof")),
  (2L, ("istoica", "prof"))))
val relationships: RDD[Edge[String]] = sc.parallelize(Seq(
  Edge(3L, 7L, "collab"),
  Edge(5L, 3L, "advisor"),
  Edge(2L, 5L, "colleague"),
  Edge(5L, 7L, "pi")))
val defaultUser = ("John Doe", "Missing")
val graph = Graph(users, relationships, defaultUser)

defaultUser covers vertices that appear in an edge but not in the vertex RDD. Here every edge endpoint exists, so the default stays unused.

GraphX software stackSearch on YouTubeBookmarkSaved

The slide stacks the layers from storage to user code. The property graph sits on RDDs of vertices and edges. Those sit on an HDFS-compatible store. Operators sit on the graph. Pregel and the algorithm library sit on the operators.

GraphX APISearch on YouTubeBookmarkSaved

Import Spark, GraphX and the RDD package first:

import org.apache.spark._
import org.apache.spark.graphx._
import org.apache.spark.rdd.RDD

The API mirrors RDDs. RDDs have map, filter and reduceByKey. Graphs have operators that take user functions and return new graphs. It also implements the Pregel API from Google. Because GraphX lives inside Spark, user code can mix GraphX, RDD and DataFrame calls.

The slide computes the in-degree of each vertex with one call:

val inDegrees = graph.inDegrees

Data typesSearch on YouTubeBookmarkSaved

Edge tripletsSearch on YouTubeBookmarkSaved

EdgeTriplet extends Edge with srcAttr and dstAttr. These hold the source and destination vertex properties. One triplet exists per edge.

The triplet view joins edge and vertex data in one record.

Building the city graphSearch on YouTubeBookmarkSaved

The running example uses six US cities as vertices. Each vertex holds city name and population. Each edge holds the distance in kilometers.

Launch the shell and import the library:

/path/to/spark/bin/spark-shell

import org.apache.spark.graphx._
import org.apache.spark.graphx.{Edge, Graph}

Create the vertex array:

val verArray = Array(
  (1L, ("Philadelphia", 1580863)),
  (2L, ("Baltimore", 620961)),
  (3L, ("Harrisburg", 49528)),
  (4L, ("Wilmington", 70851)),
  (5L, ("New York", 8175133)),
  (6L, ("Scranton", 76089)))

The output echoes the array with its type:

verArray: Array[(Long, (String, Int))] = Array((1,(Philadelphia,1580863)),
  (2,(Baltimore,620961)), (3,(Harrisburg,49528)), (4,(Wilmington,70851)),
  (5,(New York,8175133)), (6,(Scranton,76089)))

Create the edge array. The first two arguments are source and destination ids. The third argument is the distance:

val edgeArray = Array(
  Edge(2L, 3L, 113),
  Edge(2L, 4L, 106),
  Edge(3L, 4L, 128),
  Edge(3L, 5L, 248),
  Edge(3L, 6L, 162),
  Edge(4L, 1L, 39),
  Edge(1L, 6L, 168),
  Edge(1L, 5L, 130),
  Edge(5L, 6L, 159))

Parallelize both arrays and build the graph:

val verRDD = sc.parallelize(verArray)
val edgeRDD = sc.parallelize(edgeArray)
val graph = Graph(verRDD, edgeRDD)

Vertex filtersSearch on YouTubeBookmarkSaved

graph.vertices.filter keeps the vertices that pass a predicate. The task: keep cities with population above 50000.

graph.vertices.filter {
  case (id, (city, population)) => population > 50000
}.collect.foreach {
  case (id, (city, population)) =>
    println(s"The population of $city is $population")
}

The population of Philadelphia is 1580863
The population of Baltimore is 620961
The population of Wilmington is 70851
The population of New York is 8175133
The population of Scranton is 76089

Triplet queriesSearch on YouTubeBookmarkSaved

graph.triplets.collect returns one triplet per edge. The slides print the distance stored on each edge.

The slides use it to print one sentence per relationship:

for (triplet <- graph.triplets.collect) {
  println(s"The distance between ${triplet.srcAttr._1} and " +
    s"${triplet.dstAttr._1} is ${triplet.attr} kilometers")
}

triplet.srcAttr._1 is the source city name. triplet.dstAttr._1 is the destination city name. triplet.attr is the edge property (distance).

The distance between Baltimore and Harrisburg is 113 kilometers
The distance between Baltimore and Wilmington is 106 kilometers
The distance between Harrisburg and Wilmington is 128 kilometers
The distance between Harrisburg and New York is 248 kilometers
The distance between Harrisburg and Scranton is 162 kilometers
The distance between Wilmington and Philadelphia is 39 kilometers
The distance between Philadelphia and Scranton is 168 kilometers
The distance between Philadelphia and New York is 130 kilometers
The distance between New York and Scranton is 159 kilometers

Edge filtersSearch on YouTubeBookmarkSaved

graph.edges.filter keeps the edges that pass a predicate. The task: keep city pairs with distance below 150 kilometers.

graph.edges.filter {
  case Edge(city1, city2, distance) => distance < 150
}.collect.foreach {
  case Edge(city1, city2, distance) =>
    println(s"The distance between $city1 and $city2 is $distance")
}

The distance between 2 and 3 is 113
The distance between 2 and 4 is 106
The distance between 3 and 4 is 128
The distance between 4 and 1 is 39
The distance between 1 and 5 is 130

Neighbour aggregationSearch on YouTubeBookmarkSaved

aggregateMessages sends a message along each edge and merges the messages at each vertex. The task: total population of the neighbouring cities of each vertex.

GraphX graphs are directed. To count neighbours in both directions, first add the reversed edges:

val undirectedEdgeRDD = graph.reverse.edges.union(graph.edges)
val ugraph = Graph(verRDD, undirectedEdgeRDD)
val neighbors = ugraph.aggregateMessages[Int](
  ectx => ectx.sendToSrc(ectx.dstAttr._2), _ + _)

sendToSrc sends the destination population to the source vertex. _ + _ adds the received values. Each vertex total is the sum of the populations of its neighbours.

(4,2251352)
(1,8322073)
(5,1706480)
(2,120379)
(6,9805524)
(3,8943034)

Graph operatorsSearch on YouTubeBookmarkSaved

Call an operator on the graph object. The operators fall in four groups.

Basic operatorsSearch on YouTubeBookmarkSaved

val numEdges = socialGraph.numEdges
val numVertices = socialGraph.numVertices
val inDegrees = socialGraph.inDegrees
inDegrees.collect
val outDegrees = socialGraph.outDegrees
outDegrees.collect
val degrees = socialGraph.degrees
degrees.collect

graph.vertices, graph.edges and graph.triplets give collection views of each part. triplets.take(3) shows the first three triplets.

Property operatorsSearch on YouTubeBookmarkSaved

Social graph degreesSearch on YouTubeBookmarkSaved

The slides build a second graph of ten users and fifteen follows edges. Vertex 11 has no user record, so it takes the default:

val users = List((1L, User("Alex", 26)), (2L, User("Bill", 42)),
  (3L, User("Carol", 18)), (4L, User("Dave", 16)), (5L, User("Eve", 45)),
  (6L, User("Farell", 30)), (7L, User("Garry", 32)),
  (8L, User("Harry", 36)), (9L, User("Ivan", 28)), (10L, User("Jill", 48)))
val usersRDD = sc.parallelize(users)
val follows = List(Edge(1L, 2L, 1), Edge(2L, 3L, 1), Edge(3L, 1L, 1),
  Edge(3L, 4L, 1), Edge(3L, 5L, 1), Edge(4L, 5L, 1), Edge(6L, 5L, 1),
  Edge(7L, 6L, 1), Edge(6L, 8L, 1), Edge(7L, 8L, 1), Edge(7L, 9L, 1),
  Edge(9L, 8L, 1), Edge(8L, 10L, 1), Edge(10L, 9L, 1), Edge(1L, 11L, 1))
val followsRDD = sc.parallelize(follows)
val defaultUser = User("NA", 0)
val socialGraph = Graph(usersRDD, followsRDD, defaultUser)

numEdges is 15 (count the list). numVertices is 11 (vertices 1 to 10 plus vertex 11 with the default user).

Structural operatorsSearch on YouTubeBookmarkSaved

// signatures only
abstract class Graph[VD, ED] {
  def reverse: Graph[VD, ED]
  def subgraph(epred: EdgeTriplet[VD, ED] => Boolean,
    vpred: (VertexId, VD) => Boolean): Graph[VD, ED]
  def mask[VD2, ED2](other: Graph[VD2, ED2]): Graph[VD, ED]
  def groupEdges(merge: (ED, ED) => ED): Graph[VD, ED]
}

• reverse. Return a new graph with every edge direction flipped. Vertex and edge properties stay the same. Edge count stays the same. It needs no data movement. Use it for inverse PageRank.
• subgraph. Keep only vertices that pass vpred and edges that pass epred and join two kept vertices. Use it to restrict the graph to the part of interest or to drop broken edges.
• mask. Keep only vertices and edges that also exist in the other graph. Use it to intersect two graphs over the same vertex set.
• groupEdges. Merge parallel edges between the same vertices with the merge function. Use it after a union (as in the neighbour-sum example) to combine duplicate edges.

Join operatorsSearch on YouTubeBookmarkSaved

// signatures only
abstract class Graph[VD, ED] {
  def joinVertices[U](table: RDD[(VertexId, U)])
    (map: (VertexId, VD, U) => VD): Graph[VD, ED]
  def outerJoinVertices[U, VD2](table: RDD[(VertexId, U)])
    (map: (VertexId, VD, Option[U]) => VD2): Graph[VD2, ED]
}

• joinVertices. Join an RDD of (VertexId, value) pairs into the vertex properties. Only vertices present in the table change. The vertex type stays VD.
• outerJoinVertices. Same join, but every vertex gets a new property. Vertices missing from the table get None. The vertex type may change to VD2. Use it to attach optional new data to all vertices.

PageRankSearch on YouTubeBookmarkSaved

PageRank scores the importance of each vertex. An edge from u to v is an endorsement of v by u. A user followed by many users ranks high.

Run it on the bundled social-network files. Load the edge list, run PageRank, then join ranks back to usernames:

import org.apache.spark.graphx.GraphLoader
val graph = GraphLoader.edgeListFile(sc, "data/graphx/followers.txt")
val ranks = graph.pageRank(0.0001).vertices
val users = sc.textFile("data/graphx/users.txt").map { line =>
  val fields = line.split(",")
  (fields(0).toLong, fields(1))
}
val ranksByUsername = users.join(ranks).map {
  case (id, (username, rank)) => (username, rank)
}
println(ranksByUsername.collect().mkString("\n"))

pageRank(0.0001) iterates until ranks move less than the tolerance 0.0001. The join pairs each id with its (username, rank) tuple.

Connected componentsSearch on YouTubeBookmarkSaved

The connected-components algorithm labels each component with an id. GraphX ships it in the ConnectedComponents object. It is one use of the aggregation operator (like shortest paths and smallest reachable vertex id). The Pregel operator then runs it in super-steps: each vertex receives the sum (or minimum) of inbound messages per step.

import org.apache.spark.graphx.GraphLoader
val graph = GraphLoader.edgeListFile(sc, "data/graphx/followers.txt")
val cc = graph.connectedComponents().vertices
val users = sc.textFile("data/graphx/users.txt").map { line =>
  val fields = line.split(",")
  (fields(0).toLong, fields(1))
}
val ccByUsername = users.join(cc).map {
  case (id, (username, cc)) => (username, cc)
}
println(ccByUsername.collect().mkString("\n"))

Triangle countingSearch on YouTubeBookmarkSaved

triangleCount returns, per vertex, the count of triangles through it. A triangle measures clustering: it shows how tightly the neighbours of a vertex link to each other.

Two requirements hold. Edges must sit in canonical orientation ($\text{srcId} \lt \text{dstId}$). The graph must use partitionBy first:

import org.apache.spark.graphx.{GraphLoader, PartitionStrategy}
val graph = GraphLoader.edgeListFile(sc, "data/graphx/followers.txt", true)
  .partitionBy(PartitionStrategy.RandomVertexCut)
val triCounts = graph.triangleCount().vertices
val users = sc.textFile("data/graphx/users.txt").map { line =>
  val fields = line.split(",")
  (fields(0).toLong, fields(1))
}
val triCountByUsername = users.join(triCounts).map {
  case (id, (username, tc)) => (username, tc)
}
println(triCountByUsername.collect().mkString("\n"))

The true flag in edgeListFile requests canonical orientation. RandomVertexCut hashes the (srcId, dstId) pair, so an edge and its reverse land in the same partition.

Quick self-testSearch on YouTubeBookmarkSaved