We can create a graph using the following steps:

This creates a GraphX-based structure called graph, which can now be used for multiple operations. Remember that, although these data samples are small, you can create extremely large graphs using this approach. Many of these algorithms are iterative applications, for instance, PageRank and triangle count, and as a result the programs will generate many iterative Spark jobs. Just like Spark's RDDs, graphs can be cached and un-cached in memory for better performance in iterative processing.

It is easy to count the number of vertices and edges in a graph with the following functions:

scala> println( "vertices count : " + graph.vertices.count )
vertices count : 6
scala> println( "edges count : " + graph.edges.count )
edges count : 12

Alternatively, counting can be done as follows:

scala> println(s"The graph has ${graph.numVertices} vertices")
The graph has 6 vertices
scala> println(s"The graph has ${graph.numEdges} edges")
The graph has 12 edges

We can filter graphs to create sub-graphs. For example, filter the vertices to filter people whose age is greater than 40 years as shown here:

scala> val cnt1 = graph.vertices.filter { case (id, (name, age)) => age.toLong > 40 }.count
cnt1: Long = 4

scala> println( "Vertices count : " + cnt1 )
Vertices count : 4

Now, filter the edges on the relationship property of Mother or Father and then print the output:

scala> val cnt2 = graph.edges.filter { case Edge(from, to, property) => property == "Father" | property == "Mother" }.count
cnt2: Long = 4

scala> println( "Edges count : " + cnt2 )
Edges count : 4

GraphX also defines a special data structure for node degree as inDegrees, outDegrees, and degrees. The degree of a node represents the number of links it has to other nodes. We can find the incoming degree of a node, or an in-degree, which is the number of its incoming links. Also, we can find the outgoing degree, or out-degree, which is the number of nodes that it points to. Let's compute the max degrees by defining a reduce operation to compute the highest degree vertex:

scala> def max(a: (VertexId, Int), b: (VertexId, Int)): (VertexId,    Int) = {
         if (a._2 > b._2) a else b
       }
max: (a: (org.apache.spark.graphx.VertexId, Int), b: (org.apache.spark.graphx.VertexId, Int))(org.apache.spark.graphx.VertexId, Int)

scala> val maxInDegree: (VertexId, Int)  = graph.inDegrees.reduce(max)
maxInDegree: (org.apache.spark.graphx.VertexId, Int) = (1,5)

scala> val maxOutDegree: (VertexId, Int) = graph.outDegrees.reduce(max)
maxOutDegree: (org.apache.spark.graphx.VertexId, Int) = (1,3)

scala> val maxDegrees: (VertexId, Int)   = graph.degrees.reduce(max)
maxDegrees: (org.apache.spark.graphx.VertexId, Int) = (1,8)

We can define a minimum function to find people with the minimum out-degrees as follows:

scala> val minDegrees = graph.outDegrees.filter(_._2 <= 1)
minDegrees: org.apache.spark.graphx.VertexRDD[Int] = VertexRDDImpl[62] at RDD at VertexRDD.scala:57
scala> minDegrees.collect()
Array[(org.apache.spark.graphx.VertexId, Int)] = Array((4,1), (6,1))

GraphX also exposes a triplet view, which logically joins the vertex and edge properties yielding an RDD[EdgeTriplet[VD, ED]]. Figure 9.2 expresses a triplet join graphically:

Figure 9.2: Triplet view of a graph

Let's print the triplets of our graph by using the srcAttr, attr, and dstAttr members:

scala> graph.triplets.map(
    triplet => triplet.srcAttr._1 + " is the " + triplet.attr + " of " + triplet.dstAttr._1).collect.foreach(println)

Jacob is the Husband of Jessica
Jessica is the Wife of Jacob
Andrew is the Son of Jacob
Emily is the Daughter of Jacob
Emily is the Daughter of Jessica
Lily is the Sister of Jacob
Jacob is the Father of Andrew
Jacob is the Father of Emily
Jessica is the Mother of Andrew
Jessica is the Mother of Emily
Andrew is the Son of Jessica
Ryan is the Friend of Jacob

The map is one of the main transformation functions for transforming RDDs in Spark. Similarly, graphs also have three map operators as follows:

class Graph[VD, ED] { 
  def mapVertices[VD2](mapFun: (VertexId, VD) => VD2): Graph[VD2, ED] 
  def mapEdges[ED2](mapFun: Edge[ED] => ED2): Graph[VD, ED2]
  def mapTriplets[ED2](mapFun: EdgeTriplet[VD, ED] => ED2): Graph[VD, ED2] } 

Each of these methods is called on a graph with the vertex attribute type, VD and edge attribute type, ED. Each of them also takes a user-defined mapping function, say myMap, that performs one of the following:

The GraphX library also provides four useful methods for changing the structure of graphs. Their method signatures are listed as follows:

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] 
}

The operators used in the preceding code are explained as follows:

GraphX also provides APIs for joining RDD datasets with graphs as well. This is really useful when we want to add extra information or merge vertex attributes to any vertex attributes. These tasks can be accomplished using the joinVertices and outerJoinVertices join operators. Let's explore an example of joining the movies RDD with the graph we previously created:

scala> case class MoviesWatched(Movie: String, Genre: String)

scala>  val movies: RDD[(VertexId, MoviesWatched)] = sc.parallelize(List( (1, MoviesWatched("Toy Story 3", "kids")), (2, MoviesWatched("Titanic", "Love")), (3, MoviesWatched("The Hangover", "Comedy"))))

scala>  val movieOuterJoinedGraph = graph.outerJoinVertices(movies)((_,name, movies) => (name,movies))

scala> movieOuterJoinedGraph.vertices.map(t => t).collect.foreach(println)

(4,((Ryan,53),None))
(6,((Lily,52),None))
(2,((Jessica,45),Some(MoviesWatched(Titanic,Love))))
(1,((Jacob,48),Some(MoviesWatched(Toy Story 3,kids))))
(3,((Andrew,25),Some(MoviesWatched(The Hangover,Comedy))))
(5,((Emily,22),None))

You can see that movie information is added for Jacob, Andrew, and Jessica. We can use the getOrElse method to provide default attribute values for the vertices that are not present in the passed vertex:

scala> val movieOuterJoinedGraph = graph.outerJoinVertices(movies)((_,name, movies) => (name,movies.getOrElse(MoviesWatched("NA","NA"))))

scala> movieOuterJoinedGraph.vertices.map(t=>t).collect.foreach(println)

(4,((Ryan,53),MoviesWatched(NA,NA)))
(6,((Lily,52),MoviesWatched(NA,NA)))
(2,((Jessica,45),MoviesWatched(Titanic,Love)))
(1,((Jacob,48),MoviesWatched(Toy Story 3,kids)))
(3,((Andrew,25),MoviesWatched(The Hangover,Comedy)))
(5,((Emily,22),MoviesWatched(NA,NA)))

All of the graph operations in previous sections are invoked on a graph and return a new graph object. In this section, let's learn about operations that transform VertexRDD and EdgeRDD collections. The types of these collections are subtypes of RDD[(VertexID, VD)] and RDD[Edge[ED]] respectively.

def mapValues[VD2](map: VD => VD2): VertexRDD[VD2] 
def mapValues[VD2](map: (VertexId, VD) => VD2): VertexRDD[VD2]

def mapValues[ED2](f: Edge[ED] => ED2): EdgeRDD[ED2]

def filter(pred: (VertexId, VD) => Boolean): VertexRDD[VD]

def diff(another: VertexRDD[VD]): VertexRDD[VD]

def minus(another: RDD[(VertexId, VD)])

def innerJoin[U, VD2](other: RDD[(VertexId, U)])(f: (VertexId, VD, U) => VD2): VertexRDD[VD2] 
def leftJoin[U, VD2](other: RDD[(VertexId, VD2)])(f: (VertexId, VD, Option[U]) => VD2): VertexRDD[VD2]

def innerJoin[ED2, ED3](other: EdgeRDD[ED2])(f: (VertexId, VertexId, ED, ED2) => ED3): EdgeRDD[ED3] 

def reverse: EdgeRDD[ED]

The triangle count algorithm provides a vertex-based count of the number of triangles associated with a vertex. For instance, vertex Jacob (1) is connected to Emily (5), who is connected to Jessica (2). Jessica is connected to Jacob (1) and so, a triangle is formed. This can be useful for route finding, where minimum, triangle-free, spanning tree graphs need to be generated for route planning:

scala> val tCount = graph.triangleCount().vertices
scala> println( tCount.collect().mkString("
") )
(4,0)
(6,0)
(2,2)
(1,2)
(3,1)
(5,1)

The results of the application job show that the vertices called Lily (6) and Ryan (4) have no triangles, whereas Jacob (1) and Jessica (2) have the most, as expected, as they have the most relationships.

When a large graph is created from the data, it might contain unconnected subgraphs that are isolated from each other and contain no connecting edges between them. This algorithm provides a measure of this connectivity. Depending upon your processing, it is important to know that all the vertices are connected. Let's use two graph methods, connectedComponents and stronglyConnectedComponents, for this algorithm. The strong method required a maximum iteration count, which has been set to 1000. These counts are acting on the graph vertices:

scala> val iterations = 1000
scala> val connected = graph.connectedComponents().vertices
scala> val connectedS = graph.stronglyConnectedComponents(iterations).vertices

The vertex counts are then joined with the original vertex records, so that the connection counts can be associated with vertex information, such as the person's name:

scala> val connByPerson = vertices.join(connected).map {
case (id, ( (person,age) , conn )) => (conn, id, person)
}
scala> val connByPersonS = vertices.join(connectedS).map {
case (id, ( (person,age) , conn )) => (conn, id, person)
}

The results are then output using a case statement and formatted, printing:

scala> connByPerson.collect().foreach {
         case (conn, id, person) =>
         println ( f"Weak $conn $id $person" )
       }

Weak 1 4 Ryan
Weak 1 6 Lily
Weak 1 2 Jessica
Weak 1 1 Jacob
Weak 1 3 Andrew
Weak 1 5 Emily

As expected for the connectedComponents algorithm, the results show that, for each vertex, there is only one component.

The stronglyConnectedComponents algorithm gives a measure of the connectivity in a graph, taking into account the direction of the relationships between them:

scala> connByPersonS.collect().foreach {
         case (conn, id, person) =>
         println ( f"Strong $conn $id $person" )
       }

The result for the stronglyConnectedComponents algorithm output is as follows:

Strong 4 4 Ryan
Strong 6 6 Lily
Strong 1 2 Jessica
Strong 1 1 Jacob
Strong 1 3 Andrew
Strong 1 5 Emily

So, the strong method output shows that, for most vertices, there is only one graph component signified by the 1 in the second column. However, vertices 4 and 6 are not reachable due to the direction of their relationship, and so they have a vertex ID instead of a component ID.