Leveraging the Spark GraphX API

In this chapter, we will learn how to create a graph from a data source. We will then carry out experiments with the Edge API and Vertex API. By the end of this chapter, you will know how to calculate the degree of vertex and PageRank.

In this chapter, we will cover the following topics:

• Creating a graph from a data source
• Using the Vertex API
• Using the Edge API
• Calculating the degree of vertex
• Calculating PageRank

Creating a graph from a data source

We will be creating a loader component that will be used to load the data, revisit the graph format, and load a Spark graph from file.

Creating the loader component

The graph.g file consists of a structure of vertex to vertex. In the following graph.g file, if we align 1 to 2, this means that there is an edge between vertex ID 1 and vertex ID 2. The second line means that there's an edge from vertex ID 1 to 3, then from 2 to 3, and finally 3 to 5:

1  2
1  3
2  3
3  5

We will take the graph.g file, load it, and see how it will provide results in Spark. First, we need to get a resource to our graph.g file. We will do this using the getClass.getResource() method to get the path to it, as...

Using the Vertex API

In this section, we will construct the graph using edge. We will learn to use the Vertex API and also leverage edge transformations.

Constructing a graph using the vertex

Constructing a graph is not a trivial task; we need to supply vertices and edges between them. Let's focus on the first part. The first part consists of our users, users is an RDD of VertexId and String as follows:

package com.tomekl007.chapter_7

import org.apache.spark.SparkContext
import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.scalatest.FunSuite

class VertexAPI extends FunSuite {
  val spark: SparkContext = SparkSession.builder().master("local...

Using the Edge API

In this section, we will construct the graph using the Edge API. We'll also use the vertex, but this time we'll focus on the edge transformations.

Constructing the graph using edge

As we saw in the previous sections, we have edges and vertices, which is an RDD. As this is an RDD, we can get an edge. We have a lot of methods that are available on the normal RDD. We can use the max method, min method, sum method, and all other actions. We will apply the reduce method, so the reduce method will take two edges, we will take e1, e2, and we can perform some logic on it.

The e1 edge is an edge that has an attribute, destination, and a source, as shown in the following screenshot:

Since the edge is...

Calculating the degree of the vertex

In this section, we will cover the total degree, then we'll split it into two parts—an in-degree and an out-degree—and we will understand how this works in the code.

For our first test, let's construct the graph that we already know about:

package com.tomekl007.chapter_7

import org.apache.spark.SparkContext
import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.scalatest.FunSuite
import org.scalatest.Matchers._

class CalculateDegreeTest extends FunSuite {
  val spark: SparkContext = SparkSession.builder().master("local[2]").getOrCreate().sparkContext

  test("should calculate degree of vertices") {
    //given
    val users: RDD[(VertexId, (String))] =
      spark.parallelize(Array(
        (1L, "a"),
        (2L, "...

Calculating PageRank

In this section, we will load data about users and reload data about their followers. We will use the graph API and the structure of our data, and we will calculate PageRank to calculate the rank of users.

First, we need to load edgeListFile, as follows:

package com.tomekl007.chapter_7

import org.apache.spark.graphx.GraphLoader
import org.apache.spark.sql.SparkSession
import org.scalatest.FunSuite
import org.scalatest.Matchers._

class PageRankTest extends FunSuite {
  private val sc = SparkSession.builder().master("local[2]").getOrCreate().sparkContext

  test("should calculate page rank using GraphX API") {
    //given
    val graph = GraphLoader.edgeListFile(sc, getClass.getResource("/pagerank/followers.txt").getPath)

We have a followers.txt file; the following screenshot shows the format of the file, which is similar to the file we...

Summary

In this chapter, we delved into transformations and actions, and then we learned about Spark's immutable design. We studied how to avoid shuffle and how to reduce operational expenses. Then, we looked at how to save the data in the correct format. We also learned how to work with the Spark key/value API, and how to test Apache Spark jobs. After that, we learned how to create a graph from a data source, and then we investigated and experimented with the edge and vertex APIs. We learned how to calculate the degree of the vertex. Finally, we looked at PageRank and how we are able to calculate it using the Spark GraphicX API.