GraphX编程指南-官方文档-整理

import org.apache.spark._
import  org.apache.spark.graphx._
// To make some of the examples work we will also need RDD
import  org.apache.spark.rdd.RDD

class  VertexProperty()
case  class  UserProperty( val name: String)  extends  VertexProperty
case  class  ProductProperty( val name: String,  val price: Double)  extends
VertexProperty
// The graph might then have the type:
var graph: Graph[VertexProperty, String] = null

class  Graph[VD, ED] {
    val vertices: VertexRDD[VD]
    val edges: EdgeRDD[ED, VD]
}

val userGraph: Graph[(String, String), String]

// Assume the SparkContext has already been constructed
val sc: SparkContext
// Create an RDD for the vertices
val users: RDD[(VertexId, (String, String))] =
    sc.parallelize( Array((3L, ("rxin", "student")), (7L, ("jgonzal","postdoc")),(5L, ("franklin", "prof")), (2L, ("istoica", "prof"))))
// Create an RDD for edges
val relationships: RDD[Edge[String]] =
    sc.parallelize( Array( Edge(3L, 7L, "collab"), Edge(5L, 3L, "advisor"),Edge(2L, 5L, "colleague"),  Edge(5L, 7L, "pi")))
// Define a default user in case there are relationship with missing user
val defaultUser = ("John Doe", "Missing")
// Build the initial Graph
val graph =  Graph(users, relationships, defaultUser)

val graph: Graph[(String, String), String]  // Constructed from above
// Count all users which are postdocs
graph.vertices.filter {  case (id, (name, pos))  => pos == "postdoc" }.count
// Count all the edges where src > dst
graph.edges.filter(e  => e.srcId > e.dstId).count

graph.edges.filter {  case  Edge(src, dst, prop)  => src > dst }.count

val graph: Graph[(String, String), String]  // Constructed from above
// Use the triplets view to create an RDD of facts.
val facts: RDD[String] =
    graph.triplets.map(triplet  =>
        triplet.srcAttr._1 + " is the " + triplet.attr + " of " + triplet.dstAttr._1)
facts.collect.foreach(println(_ _))

val graph: Graph[(String, String), String]
// Use the implicit GraphOps.inDegrees operator
val inDegrees: VertexRDD[Int] = graph.inDegrees

/** Summary of the functionality in the property graph */
class  Graph[VD, ED] {
    //  Information  about  the  Graph
    val numEdges: Long
    val numVertices: Long
    val inDegrees: VertexRDD[Int]
    val outDegrees: VertexRDD[Int]
    val degrees: VertexRDD[Int]
    //  Views  of  the  graph  as  collections
    val vertices: VertexRDD[VD]
    val edges: EdgeRDD[ED, VD]
    val triplets: RDD[EdgeTriplet[VD, ED]]
    //  Functions  for  caching  graphs
    def persist(newLevel: StorageLevel =  StorageLevel. MEMORY_ONLY): Graph[VD, ED]
    def cache(): Graph[VD, ED]
    def unpersistVertices(blocking: Boolean =  true): Graph[VD, ED]
    //  Change  the  partitioning  heuristic
    def partitionBy(partitionStrategy: PartitionStrategy): Graph[VD, ED]
    //  Transform  vertex  and  edge  attributes
    def mapVertices[VD2](map: (VertexID, VD)  =>  VD2): Graph[VD2, ED]
    def mapEdges[ED2](map: Edge[ED]  =>  ED2): Graph[VD, ED2]
    def mapEdges[ED2](map: (PartitionID, Iterator[Edge[ED]])  =>  Iterator[ED2]):Graph[VD, ED2]
    def mapTriplets[ED2](map: EdgeTriplet[VD, ED]  =>  ED2): Graph[VD, ED2]
    def mapTriplets[ED2](map: (PartitionID, Iterator[EdgeTriplet[VD, ED]])  =>
    Iterator[ED2]): Graph[VD, ED2]
    //  Modify  the  graph  structure
    def reverse: Graph[VD, ED]
    def subgraph(
            epred: EdgeTriplet[VD,ED]  =>  Boolean = (x  =>  true),
            vpred: (VertexID, VD)  =>  Boolean = ((v, d)  =>  true)
        ): Graph[VD, ED]
    def mask[VD2, ED2](other: Graph[VD2, ED2]): Graph[VD, ED]
    def groupEdges(merge: (ED, ED)  =>  ED): Graph[VD, ED]
    //  Join  RDDs  with  the  graph
    def joinVertices[U](table: RDD[(VertexID, U)])(mapFunc: (VertexID, VD, U)
    =>  VD): Graph[VD, ED]
    def outerJoinVertices[U, VD2](other: RDD[(VertexID, U)])
    (mapFunc: (VertexID, VD,  Option[U])  =>  VD2)
    : Graph[VD2, ED]
    //  Aggregate  information  about  adjacent  triplets
    def  collectNeighborIds(edgeDirection:  EdgeDirection): VertexRDD[Array[VertexID]]
    def  collectNeighbors(edgeDirection:  EdgeDirection):VertexRDD[Array[(VertexID, VD)]]
    def mapReduceTriplets[A: ClassTag](
            mapFunc: EdgeTriplet[VD, ED]  =>  Iterator[(VertexID, A)],
            reduceFunc: (A, A)  => A,
            activeSetOpt: Option[(VertexRDD[_ _], EdgeDirection)] =  None
        ): VertexRDD[A]
    //  Iterative  graph-parallel  computation
    def pregel[A](initialMsg: A, maxIterations: Int, activeDirection:  EdgeDirection)(
            vprog: (VertexID, VD, A)  =>  VD,
            sendMsg: EdgeTriplet[VD, ED]  =>  Iterator[(VertexID,A)],
            mergeMsg: (A, A)  => A
        ): Graph[VD, ED]
    //  Basic  graph  algorithms
    def pageRank(tol: Double, resetProb: Double = 0.15): Graph[Double, Double]
    def connectedComponents(): Graph[VertexID, ED]
    def triangleCount(): Graph[Int, ED]
    def stronglyConnectedComponents(numIter: Int): Graph[VertexID, ED]
}

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

val newVertices = graph.vertices.map {  case (id, attr)  => (id, mapUdf(id,
attr)) }
val newGraph =  Graph(newVertices, graph.edges)

val newGraph = graph.mapVertices((id, attr)  => mapUdf(id, attr))

// Given a graph where the vertex property is the out-degree
val inputGraph: Graph[Int, String] =
    graph.outerJoinVertices(graph.outDegrees)((vid,  _ _,  degOpt)  => degOpt.getOrElse(0))
// Construct a graph where each edge contains the weight
// and each vertex is the initial PageRank
val outputGraph: Graph[Double, Double] =
    inputGraph.mapTriplets(triplet  => 1.0 / triplet.srcAttr).mapVertices((id, _ _) => 1.0)

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

// Create an RDD for the vertices
val users: RDD[(VertexId, (String, String))] =
        sc.parallelize( Array((3L, ("rxin", "student")), (7L, ("jgonzal","postdoc")),(5L, ("franklin", "prof")), (2L, ("istoica", "prof")),(4L, ("peter", "student"))))
// Create an RDD for edges
val relationships: RDD[Edge[String]] =
        sc.parallelize( Array( Edge(3L, 7L, "collab"), Edge(5L, 3L, "advisor"),Edge(2L, 5L, "colleague"),  Edge(5L, 7L, "pi"),Edge(4L, 0L, "student"), Edge(5L, 0L, "colleague")))
// Define a default user in case there are relationship with missing user
val defaultUser = ("John Doe", "Missing")
// Build the initial Graph
val graph =  Graph(users, relationships, defaultUser)
// Notice that there is a user 0 (for which we have no information) connected to users
// 4 (peter) and 5 (franklin).
graph.triplets.map(
    triplet  => triplet.srcAttr._1 + " is the " + triplet.attr + " of " + triplet.dstAttr._1
).collect.foreach(println(_ _))
// Remove missing vertices as well as the edges to connected to them
val validGraph = graph.subgraph(vpred = (id, attr)  => attr._2 != "Missing")
// The valid subgraph will disconnect users 4 and 5 by removing user 0
validGraph.vertices.collect.foreach(println(_ _))
validGraph.triplets.map(
    triplet  => triplet.srcAttr._1 + " is the " + triplet.attr + " of " + triplet.dstAttr._1
).collect.foreach(println(_ _))

// Run Connected Components
val ccGraph = graph.connectedComponents()  // No longer contains missing field
// Remove missing vertices as well as the edges to connected to them
val validGraph = graph.subgraph(vpred = (id, attr)  => attr._2 != "Missing")
// Restrict the answer to the valid subgraph
val validCCGraph = ccGraph.mask(validGraph)

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

val nonUniqueCosts: RDD[(VertexID, Double)]
val uniqueCosts: VertexRDD[Double] = 
        graph.vertices.aggregateUsingIndex(nonUnique, (a,b) => a + b)
val joinedGraph = graph.joinVertices(uniqueCosts)(
        (id, oldCost, extraCost)  => oldCost + extraCost)

val outDegrees: VertexRDD[Int] = graph.outDegrees
val degreeGraph = graph.outerJoinVertices(outDegrees) { 
    (id, oldAttr, outDegOpt)  =>
        outDegOpt  match {
            case  Some(outDeg)  => outDeg
            case  None  => 0  // No outDegree means zero outDegree
        }
}

val joinedGraph = graph.joinVertices(uniqueCosts,
    (id: VertexID, oldCost: Double, extraCost: Double)  => oldCost + extraCost)

class Graph[VD, ED] {
    def mapReduceTriplets[A](
        map: EdgeTriplet[VD, ED]  =>  Iterator[(VertexId, A)],
        reduce: (A, A)  => A 
    ): VertexRDD[A]
}

activeSetOpt: Option[(VertexRDD[_], EdgeDirection)] = None

// Import random graph generation library
import  org.apache.spark.graphx.util.GraphGenerators
// Create a graph with "age" as the vertex property. Here we use a random graph for simplicity.
val graph: Graph[Double, Int] =
        GraphGenerators.logNormalGraph(sc, numVertices = 100).mapVertices( 
            (id, _ _) => id.toDouble )
// Compute the number of older followers and their total age
val olderFollowers: VertexRDD[(Int, Double)] = graph.mapReduceTriplets[(Int,Double)](
    triplet  => {  // Map Function
        if (triplet.srcAttr > triplet.dstAttr) {
            // Send message to destination vertex containing counter and age
            Iterator((triplet.dstId, (1, triplet.srcAttr)))
        }  else {
            // Don't send a message for this triplet
            Iterator.empty
        }
    },
    // Add counter and age
    (a, b)  => (a._1 + b._1, a._2 + b._2)  // Reduce Function
)
// Divide total age by number of older followers to get average age of older followers
val avgAgeOfOlderFollowers: VertexRDD[Double] =
    olderFollowers.mapValues( (id, value)  => value  
        match {  case (count, totalAge) => totalAge / count } )
// Display the results
avgAgeOfOlderFollowers.collect.foreach(println(_ _))

// Define a reduce operation to compute the highest degree vertex
def max(a: (VertexId, Int), b: (VertexId, Int)): (VertexId, Int) = {
    if (a._2 > b._2) a  else b
}
// Compute the max degrees
val maxInDegree: (VertexId, Int) = graph.inDegrees.reduce(max)
val maxOutDegree: (VertexId, Int) = graph.outDegrees.reduce(max)
val maxDegrees: (VertexId, Int) = graph.degrees.reduce(max)

class GraphOps[VD, ED] {
    def  collectNeighborIds(edgeDirection:  EdgeDirection): VertexRDD[Array[VertexId]]
    def  collectNeighbors(edgeDirection:  EdgeDirection): VertexRDD[ Array[(VertexId, VD)] ]
}

class  GraphOps[VD, ED] {
    def pregel[A](
    initialMsg: A,
    maxIter: Int =  Int. MaxValue,
    activeDir: EdgeDirection =  EdgeDirection. Out)(
        vprog: (VertexId, VD, A)  =>  VD,
        sendMsg: EdgeTriplet[VD, ED]  =>  Iterator[(VertexId, A)],
        mergeMsg: (A, A)  => A
    ): Graph[VD, ED] = {
        // Receive the initial message at each vertex
        var  g  =  mapVertices(  (vid,  vdata)  =>  vprog(vid,  vdata, initialMsg) ).cache()
        // compute the messages
        var messages = g.mapReduceTriplets(sendMsg, mergeMsg)
        var activeMessages = messages.count()
        // Loop until no messages remain or maxIterations is achieved
        var i = 0
        while (activeMessages > 0 && i < maxIterations) {
            //  Receive  the  messages:------
            // Run the vertex program on all vertices that receive messages
            val newVerts = g.vertices.innerJoin(messages)(vprog).cache()
            // Merge the new vertex values back into the graph
            g  =  g.outerJoinVertices(newVerts)  {  (vid,  old,  newOpt)  =>  newOpt.getOrElse(old) }.cache()
            //  Send  Messages:-----
            --
            // Vertices that didn't receive a message above don't appear in newVerts and therefore don't
            // get to send messages. More precisely the map phase of mapReduceTriplets is only invoked
            // on edges in the activeDir of vertices in newVerts
            messages = g.mapReduceTriplets(sendMsg, mergeMsg,  Some((newVerts, activeDir))).cache()
            activeMessages = messages.count()
            i += 1
        }
        g
    }
}

import  org.apache.spark.graphx._
// Import random graph generation library
import  org.apache.spark.graphx.util.GraphGenerators
// A graph with edge attributes containing distances
val graph: Graph[Int, Double] =
        GraphGenerators.logNormalGraph(sc, numVertices = 100).mapEdges(e => e.attr.toDouble)
val sourceId: VertexId = 42  // The ultimate source
// Initialize the graph such that all vertices except the root have distance infinity.
val initialGraph = graph.mapVertices((id, _ _)  =>  if (id == sourceId) 0.0  else Double.PositiveInfinity)
val sssp = initialGraph.pregel( Double. PositiveInfinity)(
    (id, dist, newDist)  => math.min(dist, newDist),  // Vertex Program
    triplet  => { // Send Message
        if (triplet.srcAttr + triplet.attr < triplet.dstAttr) {
            Iterator((triplet.dstId, triplet.srcAttr + triplet.attr))
        }  else {
            Iterator.empty
        }
    },
    (a,b)  => math.min(a,b)  // Merge Message
)
println(sssp.vertices.collect.mkString("\n"))

object  GraphLoader {
    def edgeListFile(
        sc: SparkContext,
        path: String,
        canonicalOrientation: Boolean =  false,
        minEdgePartitions: Int = 1
    ): Graph[Int, Int]
}

object Graph {
    def apply[VD, ED](
            vertices: RDD[(VertexId, VD)],
            edges: RDD[Edge[ED]],
            defaultVertexAttr: VD =  null
    ): Graph[VD, ED]
    
    def fromEdges[VD, ED](
            edges: RDD[Edge[ED]],
            defaultValue: VD): Graph[VD, ED]
    
    def fromEdgeTuples[VD](
            rawEdges: RDD[(VertexId, VertexId)],
            defaultValue: VD,
            uniqueEdges: Option[PartitionStrategy] =  None
    ): Graph[VD, Int]
}

class  VertexRDD[VD]  extends  RDD[(VertexID, VD)] {
    // Filter the vertex set but preserves the internal index
    def filter(pred: Tuple2[VertexId, VD]  =>  Boolean): VertexRDD[VD]
    // Transform the values without changing the ids (preserves the internal index)
    def mapValues[VD2](map: VD =>  VD2): VertexRDD[VD2]
    def mapValues[VD2](map: (VertexId, VD)  =>  VD2): VertexRDD[VD2]
    // Remove vertices from this set that appear in the other set
    def diff(other: VertexRDD[VD]): VertexRDD[VD]
    // Join operators that take advantage of the internal indexing to accelerate joins (substantially)
    def leftJoin[VD2, VD3](other: RDD[(VertexId, VD2)])(f: (VertexId, VD, Option[VD2])  =>  VD3): VertexRDD[VD3]
    def innerJoin[U, VD2](other: RDD[(VertexId, U)])(f: (VertexId, VD, U)  => VD2): VertexRDD[VD2]
    // Use the index on this RDD to accelerate a `reduceByKey` operation on the input RDD.
    def aggregateUsingIndex[VD2](other: RDD[(VertexId, VD2)], reduceFunc: (VD2, VD2)  =>  VD2): VertexRDD[VD2]
}

val setA: VertexRDD[Int] =  VertexRDD(sc.parallelize(0L until 100L).map(id  => (id, 1)))
val rddB: RDD[(VertexId, Double)] = sc.parallelize(0L until 100L).flatMap(id =>  List((id, 1.0), (id, 2.0)))
// There should be 200 entries in rddB
rddB.count
val setB: VertexRDD[Double] = setA.aggregateUsingIndex(rddB, _ _ + _ _)
// There should be 100 entries in setB
setB.count
// Joining A and B should now be fast!
val setC: VertexRDD[Double] = setA.innerJoin(setB)((id, a, b)  => a + b)

// Transform the edge attributes while preserving the structure
def mapValues[ED2](f: Edge[ED]  =>  ED2): EdgeRDD[ED2, VD]
// Revere the edges reusing both attributes and structure
def reverse: EdgeRDD[ED, VD]
// Join two `EdgeRDD`s partitioned using the same partitioning strategy.
def innerJoin[ED2, ED3](other: EdgeRDD[ED2, VD])(f: (VertexId, VertexId,  ED, ED2) => ED3): EdgeRDD[ED3, VD]

// Load the edges as a graph
val graph =  GraphLoader.edgeListFile(sc, "graphx/data/followers.txt")
// Run PageRank
val ranks = graph.pageRank(0.0001).vertices
// Join the ranks with the usernames
val users = sc.textFile("graphx/data/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)
}
// Print the result
println(ranksByUsername.collect().mkString("\n"))

// Load the graph as in the PageRank example
val graph =  GraphLoader.edgeListFile(sc, "graphx/data/followers.txt")
// Find the connected components
val cc = graph.connectedComponents().vertices
// Join the connected components with the usernames
val users = sc.textFile("graphx/data/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)
}
// Print the result
println(ccByUsername.collect().mkString("\n"))

// Load the edges in canonical order and partition the graph for triangle count
val graph =  GraphLoader.edgeListFile(sc, "graphx/data/followers.txt", true).partitionBy( PartitionStrategy. RandomVertexCut)
// Find the triangle count for each vertex
val triCounts = graph.triangleCount().vertices
// Join the triangle counts with the usernames
val users = sc.textFile("graphx/data/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)
    }
// Print the result
println(triCountByUsername.collect().mkString("\n"))

// Connect to the Spark cluster
val sc =  new  SparkContext("spark://master.amplab.org", "research")
// Load my user data and parse into tuples of user id and attribute list
val users = (sc.textFile("graphx/data/users.txt").map(line  =>  
        line.split(",")).map(  
            parts  =>  (parts.head.toLong,parts.tail) 
        )
    )
// Parse the edge data which is already in userId -> userId format
val  followerGraph  =  GraphLoader.edgeListFile(sc,"graphx/data/followers.txt")
// Attach the user attributes
val graph = followerGraph.outerJoinVertices(users) {
        case (uid, deg,  Some(attrList))  => attrList
        // Some users may not have attributes so we set them as empty
        case (uid, deg,  None)  =>  Array.empty[String]
}
// Restrict the graph to users with usernames and names
val subgraph = graph.subgraph(vpred = (vid, attr)  => attr.size == 2)
// Compute the PageRank
val pagerankGraph = subgraph.pageRank(0.001)
// Get the attributes of the top pagerank users
val  userInfoWithPageRank  =
        subgraph.outerJoinVertices(pagerankGraph.vertices) {
            case (uid, attrList,  Some(pr))  => (pr, attrList.toList)
            case (uid, attrList,  None)  => (0.0, attrList.toList)
        }
println(userInfoWithPageRank.vertices.top(5)( 
    Ordering.by(_ _._2._1)).mkString("\n")
)