> For the complete documentation index, see [llms.txt](https://alvintoh.gitbook.io/apache-2-0-spark-with-scala/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://alvintoh.gitbook.io/apache-2-0-spark-with-scala/section-9-intro-to-graphx/50.-graphx-pregel-and-breadth-first-search-with-pregel..md).

# 50. GraphX, Pregel, and Breadth-First-Search with Pregel.

## Analyzing and transversing graphs

Using Spark GraphX

### GRAPHX

* Not that kind of graph...
* Graphs like oursocial network of superheroes - graphs in the computer science / network sense.
* However, it's only useful for specific things.
  * It can measure things like "connectedness", degree distribution, average path length, triangle counts - high level measures of a graph
  * It can count triangles in the graph, and apply the PageRank algorithm to it.
  * It can also join graphs together and transform graphs quickly
  * For things like our "degrees of seperation" example, you won't find built-in support. But it does support the Pregel API for transversing a graph...
* Introduces VertexRDD and EdgeRDD, and the Edge data type
* Otherwise, GraphX code looks like any other Spark code for the most part

### Creating Vertex RDD's

```
    // Function to extract hero ID -> hero name tuples (or None in case of failure)
    def parseNames(line: String) : Option[(VertexId, String)] = {
        var fields = line.split('\'")
        if (fields.length > 1){
            val heroID: Long = fields(0).trim().toLong
            if (heroID < 6487) { // ID's above 6486 aren't real characters
                return Some(fields(0).trim().toLong, fields(1))
            }
        }

        return None // flatmap will just discard None results, and extract data from Some results.
    }
```

### Creating Edge RDD's

```
    /** Transform an input line from marvel-graph.txt into a List of Edges */
    def makeEdges(line: String) : List[Edge[Int]] = {
        import scala.collection.mutable.ListBuffer
        var edges = new ListBuffer[Edge[Int]]()
        val fields = line.split(" ")
        val origin = fields(0)
        for (x <- 1 to (fields.length - 1)) {
            // Our attribute field is unused, but in other graphs could
            // be used to deep track of physical distances etc.
            edges += Edge(origin.toLong, fields(x).toLong, 0)
        }

        return edges.toList
    }
```

### Creating A Graph

```
    // Build up our vertices
    val names = sc.textFile("../marvel-names.txt")
    val verts = names.flatMap(parseNames)

    // Build up our edges
    val lines = sc.textFile("../marvel-graph.txt")
    val edges = lines.flatMap(makeEdges)    

    // Build up our graph, and cache it as we're going to do a bunch of stuff with it.
    val default = "Nobody"
    val graph = Graph(verts, edges, default).cache()
```

### Doing Stuff

* Top 10 most-connected heroes:

  ```
    graph.degrees.join(verts).sortBy(_._2._1, ascending=false).take(10).foreach(println)
  ```

### Breadth-First Search With Pregel

* Before, we actually implemented BFS search in Spark the hard and manual way.
* Now with Pregel, that gives us an general algorithm for actually doing iterations through a graph in a general manner and we can actually implement breadth first search using the Pregel algorithm, in a little bit of a more simple way than we did it before

### How Pregel Works

* Vertices send messages to other vertices (their neighbours)
* The graph is processed in iterations called supersteps
* Each superstep does the following:
  * Messages from the previous iteration are received by each vertex
  * Each vertex runs a program to transform itself
  * Each vertex sends messages to other vertices

### BFS: Initialization

```
    val initialGraph = graph.mapVertices(id, _) => if (id == root) 0.0 else Double.PositiveInfinity)
```

### Sending Messages

```
    triplet => { 
        if (triplet.srcAttr != Double.PositiveInfinity) { 
            Iterator((triplet.dstId, triplet.srcAttr+1)) 
        } else { 
        Iterator.empty 
        } 
    },
```

### Preserving The Minimum Distance At Each Step

* Pregel's vertex program will preserve the minimum distance between the one it receives and what it has:
  * (id, attr, msg) => math.min(attr, msg)
* Its reduce operation preserves the minimum distance if multiple messages are received for the same vertex:
  * (a,b) => math.min(a,b)

### Putting It All Together

```
    // Initialize each node with a distance of infinity, unless it's our starting point
    val initialGraph = graph.mapVertices((id, _) => if (id == root) 0.0 else Double.PositiveInfinity)

    // Now the Pregel magic
    val bfs = initialGraph.pregel(Double.PositiveInfinity, 10)( 
        // Our "vertex program" preserves the shortest distance
        // between an inbound message and its current value.
        // It receives the vertex ID we are operating on,
        // the attribute already stored with the vertex, and
        // the inbound message from this iteration.
        (id, attr, msg) => math.min(attr, msg), 

        // Our "send message" function propagates out to all neighbors
        // with the distance incremented by one.
        triplet => { 
            if (triplet.srcAttr != Double.PositiveInfinity) { 
                Iterator((triplet.dstId, triplet.srcAttr+1)) 
            } else { 
            Iterator.empty 
            } 
        }, 

        // The "reduce" operation preserves the minimum
        // of messages received by a vertex if multiple
        // messages are received by one vertex
        (a,b) => math.min(a,b) )
    }
```

### Let's Do It