0
0

Spark Graphx Pregel原理方法示例源码详解

Spark Graphx Pregel原理方法示例源码详解–点击此标题看全文

文章目录

基本思想

Pregel计算模型是一个分布式计算模型，主要用于大规模图计算。它的基本思想是迭代计算和顶点为中心，并采用消息传递机制来实现并行计算。

  1. 迭代计算和顶点为中心 在Pregel模型中，每个迭代都执行一个函数，这个函数会发送和接收消息，以实现迭代式的计算过程。这种以顶点为中心的模型允许每个顶点独立执行计算，不需要全局同步。
  2. 消息传递机制 Pregel采用消息传递机制来实现顶点之间的通信。顶点之间通过发送消息来进行通信，这些消息包含了顶点计算所需的值。消息会按照发送的顺序进行传递，但不会保证顺序的精确性。
  3. 并行和同步 Pregel模型支持并行执行，并通过对超步之间的同步来确保程序的正确性。超步之间的同步点确保所有消息都已经被传递和接收，而顶点状态的更新也不会发生冲突。
  4. 输入输出 Pregel支持多种输入输出格式，将输入输出与图计算分离。用户可以根据需要选择不同的输入输出格式。
  5. 面临挑战 (1) 通信开销：由于消息传递机制需要传输大量数据，因此通信开销是一个挑战。为降低通信开销，Pregel使用合并器来合并多个消息。 (2) 同步性：由于超步之间的同步点会导致延迟，因此同步性也是一个挑战。为降低同步性带来的影响，需要将顶点分配到不同的机器上，以分散负载。

综上所述，Pregel计算模型通过迭代计算和顶点为中心的消息传递机制实现了大规模图计算，并支持并行执行和输入输出。然而，它面临着通信开销和同步性的挑战。

实现的关键要点

在本文中，关于Pregel实现的关键要点包括：

  • Pregel将输入图划分为多个分区，每个分区包含一组顶点和它们的出边。顶点到分区的分配是基于顶点ID。
  • 主节点负责协调图分区到工作节点的分配。它还处理输入/输出、检查点记录和恢复。
  • 工作节点在内存中维护它们所分配的图分区的状态，包括顶点值、边值、传入消息和活动顶点标志。
  • Pregel采用以顶点为中心的模型，每个活动顶点在每个超级步中执行Compute()方法。该方法可以读取消息、发送消息、修改本地状态和请求突变。
  • 消息在工作节点之间异步发送，允许计算和通信重叠。然而，在下一个超级步之前，消息会按顺序传递。
  • 检查点记录用于实现容错性，在每个超级步开始时，工作节点将它们的部分状态保存到持久性存储中。在发生故障时，受影响的分区将被重新分配并从最新的检查点恢复。
  • 为了确保确定性，在一个超级步内以特定的顺序应用突变（首先是移除，然后是添加）。使用处理器解决冲突。
  • 聚合器使全局通信和监视成为可能，工作节点在主节点计算最终全局值之前，部分减少了值。
  • 在当前的实现中，整个计算状态都驻留在内存中。然而，作者计划将一些数据溢出到磁盘以处理更大的图。

底层实现

Pregel的C++ API主要包括以下关键特征：

  1. 定义Vertex类，提供Compute()方法用于实现用户定义的顶点计算逻辑。
  2. 提供Edge、Message类，用于处理边和消息。
  3. 定义Combiner类，用于实现消息的聚合，提高通信效率。
  4. 提供Aggregator类，用于实现全局聚合操作，提供全局统计信息。
  5. 支持拓扑变化操作，允许用户改变图的拓扑结构。
  6. 提供I/O接口，支持多种输入输出格式。 这些API设计提供了灵活的分布式图计算编程接口，降低了分布式计算的学习成本，为大规模图计算提供了良好的支持。

消息传递特征

Pregel中的消息传递机制具有以下关键特征：

  1. 采用纯消息传递机制，不支持远程读。
  2. 每个顶点可以在一个超步内发送任意数量的消息。
  3. 消息使用异步方式传递，采用批量发送以优化通信效率。
  4. 消息的接收是同步的，每个顶点在一个超步内只能接收到上一个超步发送的消息。
  5. 消息通常沿出边传递，但也可以发送到任何已知ID的顶点。

这种消息传递机制利用了Pregel的并行计算能力，为大规模图计算提供了高效的通信支持。

Combiners特征

Pregel中的组合器（Combiners）具有以下关键特征：

  1. 组合器是一种消息优化机制，可合并多个消息。
  2. 用户需定义组合器类，实现组合函数。
  3. 组合器要求操作具有结合律和交换律，以确保正确性。
  4. 组合器通常用于降低通信开销，提高性能。
  5. 组合器主要应用于单源最短路径等算法中。 这些组合器的设计利用了Pregel的并行计算能力，通过消息的合并提高了通信效率，为大规模图计算提供了优化支持。

Aggregators特征

Pregel中的聚合器（Aggregators）具有以下关键特征：

  1. 聚合器是一种全局聚合操作，在所有节点之间共享聚合值。
  2. 用户可以定义自己的聚合器类，实现聚合函数和聚合操作。
  3. 聚合器包括多种内置聚合操作，如min、max、sum等。
  4. 聚合器可用于统计信息、全局协调和并行计算。
  5. 聚合器的聚合值在每个超步结束后传递给所有节点。 这些聚合器的设计利用了Pregel的并行计算能力，支持复杂的图算法实现，提高了算法的效率。

方法参数

类型

  • VD：顶点数据类型
  • ED：边数据类型
  • A：Pregel消息类型

参数：

  • graph：输入图
  • initialMsg：每个顶点在第一次迭代时接收到的消息
  • maxIterations：最大迭代次数
  • activeDirection：上一轮接收到消息的顶点周围边的方向，在其上运行 sendMsg 。 - 如果为 EdgeDirection.Out ，则只会在上一轮接收到消息的顶点的出边上运行 sendMsg 。- 默认值为 EdgeDirection.Either ，表示将在上一轮任何一侧接收到消息的边上运行 sendMsg 。- 如果为 EdgeDirection.Both ，则仅在上一轮两个顶点都接收到消息的边上运行 sendMsg 。
  • vprog：用户定义的顶点程序，它在每个顶点上运行，并接收传入的消息并计算新的顶点值。在第一次迭代时，顶点程序被应用于所有顶点，并传递默认的消息。在后续迭代中，只有收到消息的顶点才会调用顶点程序。
  • sendMsg：用户提供的函数，应用于当前迭代中收到消息的顶点的出边
  • mergeMsg：用户提供的函数，将两个类型为 A 的传入消息合并为一个类型为 A 的单个消息。“该函数必须是可交换和可结合的，并且最好 A 的大小不增加。”

返回：计算结束后的图对象

示例

这是一个使用Pregel实现PageRank算法的示例代码:

val pagerankGraph: Graph[Double,Double]= graph
  // 将每个顶点与其度关联起来.outerJoinVertices(graph.outDegrees){(vid, vdata, deg)=> deg.getOrElse(0)}// 根据度设置边的权重.mapTriplets(e =>1.0/ e.srcAttr)// 将顶点属性设置为初始的PageRank值.mapVertices((id, attr)=>1.0)def vertexProgram(id: VertexId, attr:Double, msgSum:Double):Double=
  resetProb +(1.0- resetProb)* msgSum

def sendMessage(id: VertexId, edge: EdgeTriplet[Double,Double]): Iterator[(VertexId,Double)]=
  Iterator((edge.dstId, edge.srcAttr * edge.attr))def messageCombiner(a:Double, b:Double):Double= a + b

val initialMessage =0.0// 执行固定次数的Pregel迭代。val resultGraph = Pregel(pagerankGraph, initialMessage, numIter)(
  vertexProgram, sendMessage, messageCombiner)

以上代码中，我们首先将每个顶点的出度与顶点关联起来，然后根据出度设置边的权重。接下来，我们将每个顶点的属性初始化为初始的PageRank值。

然后，定义了三个函数：

  • vertexProgram ：在每个顶点上执行的顶点程序，计算新的顶点值。
  • sendMessage ： 用于计算每个顶点发送的消息。
  • messageCombiner： 将两个消息合并为一个。

最后，我们使用Pregel函数对图执行固定次数的迭代，其中传入了定义的顶点程序、消息发送函数和消息合并函数。迭代完成后，得到的结果图即为PageRank算法的计算结果。

源码

object Pregel

/**
 * 执行类似Pregel的迭代式顶点并行计算。用户定义的顶点程序 `vprog` 在每个顶点上并行执行，
 * 接收传入的任何入站消息，并计算顶点的新值。然后，在所有出边上调用 `sendMsg` 函数，
 * 用于计算目标顶点的可选消息。`mergeMsg` 函数是一个交换和结合性函数，用于将针对同一目标顶点的多个消息合并为单个消息。
 *
 * 在第一次迭代时，所有顶点都会接收到 `initialMsg` ，在后续迭代中，如果顶点没有收到消息，则不会调用顶点程序。
 *
 * 该函数迭代直到没有剩余消息或达到 `maxIterations` 次数。
 *
 * @tparam VD 顶点数据类型
 * @tparam ED 边数据类型
 * @tparam A Pregel消息类型
 *
 * @param graph 输入图
 *
 * @param initialMsg 每个顶点在第一次迭代时接收到的消息
 *
 * @param maxIterations 最大迭代次数
 *
 * @param activeDirection 上一轮接收到消息的顶点周围边的方向，在其上运行 `sendMsg` 。
 *                      例如，如果为 `EdgeDirection.Out` ，则只会在上一轮接收到消息的顶点的出边上运行 `sendMsg` 。
 *                      默认值为 `EdgeDirection.Either` ，表示将在上一轮任何一侧接收到消息的边上运行 `sendMsg` 。
 *                      如果为 `EdgeDirection.Both` ，则仅在上一轮两个顶点都接收到消息的边上运行 `sendMsg` 。
 *
 * @param vprog 用户定义的顶点程序，它在每个顶点上运行，并接收传入的消息并计算新的顶点值。
 *              在第一次迭代时，顶点程序被应用于所有顶点，并传递默认的消息。在后续迭代中，只有收到消息的顶点才会调用顶点程序。
 *
 * @param sendMsg 用户提供的函数，应用于当前迭代中收到消息的顶点的出边
 *
 * @param mergeMsg 用户提供的函数，将两个类型为 A 的传入消息合并为一个类型为 A 的单个消息。
 *                 "该函数必须是可交换和可结合的，并且最好 A 的大小不增加。"
 *
 * @return 计算结束后的图对象
 */object Pregel extends Logging {/**
   * 执行类似Pregel的迭代式顶点并行计算。用户定义的顶点程序 `vprog` 在每个顶点上并行执行，
   * 接收传入的任何入站消息，并计算顶点的新值。然后，在所有出边上调用 `sendMsg` 函数，
   * 用于计算目标顶点的可选消息。`mergeMsg` 函数是一个交换和结合性函数，用于将针对同一目标顶点的多个消息合并为单个消息。
   *
   * 在第一次迭代时，所有顶点都会接收到 `initialMsg` ，在后续迭代中，如果顶点没有收到消息，则不会调用顶点程序。
   *
   * 该函数迭代直到没有剩余消息或达到 `maxIterations` 次数。
   *
   * @tparam VD 顶点数据类型
   * @tparam ED 边数据类型
   * @tparam A Pregel消息类型
   *
   * @param graph 输入图
   *
   * @param initialMsg 每个顶点在第一次迭代时接收到的消息
   *
   * @param maxIterations 最大迭代次数
   *
   * @param activeDirection 上一轮接收到消息的顶点周围边的方向，在其上运行 `sendMsg` 。
   *                      例如，如果为 `EdgeDirection.Out` ，则只会在上一轮接收到消息的顶点的出边上运行 `sendMsg` 。
   *                      默认值为 `EdgeDirection.Either` ，表示将在上一轮任何一侧接收到消息的边上运行 `sendMsg` 。
   *                      如果为 `EdgeDirection.Both` ，则仅在上一轮两个顶点都接收到消息的边上运行 `sendMsg` 。
   *
   * @param vprog 用户定义的顶点程序，它在每个顶点上运行，并接收传入的消息并计算新的顶点值。
   *              在第一次迭代时，顶点程序被应用于所有顶点，并传递默认的消息。在后续迭代中，只有收到消息的顶点才会调用顶点程序。
   *
   * @param sendMsg 用户提供的函数，应用于当前迭代中收到消息的顶点的出边
   *
   * @param mergeMsg 用户提供的函数，将两个类型为 A 的传入消息合并为一个类型为 A 的单个消息。
   *                 "该函数必须是可交换和可结合的，并且最好 A 的大小不增加。"
   *
   * @return 计算结束后的图对象
   */def apply[VD: ClassTag, ED: ClassTag, A: ClassTag](graph: Graph[VD, ED],
      initialMsg: A,
      maxIterations:Int=Int.MaxValue,
      activeDirection: EdgeDirection = EdgeDirection.Either)(vprog:(VertexId, VD, A)=> VD,
      sendMsg: EdgeTriplet[VD, ED]=> Iterator[(VertexId, A)],
      mergeMsg:(A, A)=> A): Graph[VD, ED]={
    require(maxIterations >0, s"最大迭代次数必须大于0，但得到的是${maxIterations}")val checkpointInterval = graph.vertices.sparkContext.getConf
      .getInt("spark.graphx.pregel.checkpointInterval",-1)var g = graph.mapVertices((vid, vdata)=> vprog(vid, vdata, initialMsg))val graphCheckpointer =new PeriodicGraphCheckpointer[VD, ED](
      checkpointInterval, graph.vertices.sparkContext)
    graphCheckpointer.update(g)// 计算消息var messages = GraphXUtils.mapReduceTriplets(g, sendMsg, mergeMsg)val messageCheckpointer =new PeriodicRDDCheckpointer[(VertexId, A)](
      checkpointInterval, graph.vertices.sparkContext)
    messageCheckpointer.update(messages.asInstanceOf[RDD[(VertexId, A)]])var isActiveMessagesNonEmpty =!messages.isEmpty()// 循环var prevG: Graph[VD, ED]=nullvar i =0while(isActiveMessagesNonEmpty && i < maxIterations){// 接收消息并更新顶点。
      prevG = g
      g = g.joinVertices(messages)(vprog)
      graphCheckpointer.update(g)val oldMessages = messages
      // 发送新消息，跳过未收到消息的边。我们必须缓存消息，以便在下一行中实现材料化，// 从而允许我们取消缓存上一个迭代。
      messages = GraphXUtils.mapReduceTriplets(
        g, sendMsg, mergeMsg, Some((oldMessages, activeDirection)))// 调用count()方法会实际材料化`messages`和`g`的顶点。这隐藏了`oldMessages`// （被`g`的顶点所依赖）和`prevG`的顶点（被`oldMessages`和`g`的顶点所依赖）。
      messageCheckpointer.update(messages.asInstanceOf[RDD[(VertexId, A)]])
      isActiveMessagesNonEmpty =!messages.isEmpty()

      logInfo("Pregel 完成迭代 "+ i)// 取消缓存被新材料化RDD隐藏的RDD
      oldMessages.unpersist()
      prevG.unpersistVertices()
      prevG.edges.unpersist()// 计数迭代次数
      i +=1}
    messageCheckpointer.unpersistDataSet()
    graphCheckpointer.deleteAllCheckpoints()
    messageCheckpointer.deleteAllCheckpoints()
    g
  }// end of apply}// end of class Pregel

该代码定义了一个名为Pregel的对象，实现了类似Pregel的迭代式顶点并行计算。具体来说：

  1. apply方法：执行类似Pregel的迭代式顶点并行计算。用户定义的顶点程序 vprog 在每个顶点上并行执行，接收传入的任何入站消息，并计算顶点的新值。然后，在所有出边上调用 sendMsg 函数，用于计算目标顶点的可选消息。mergeMsg 函数是一个交换和结合性函数，用于将针对同一目标顶点的多个消息合并为单个消息。
  2. 在第一次迭代时，所有顶点都会接收到 initialMsg ，在后续迭代中，如果顶点没有收到消息，则不会调用顶点程序。
  3. 该函数迭代直到没有剩余消息或达到 maxIterations 次数。
  4. graph表示输入图，initialMsg表示每个顶点在第一次迭代时接收到的消息，maxIterations表示最大迭代次数，activeDirection表示上一轮接收到消息的顶点周围边的方向，在其上运行 sendMsg。
  5. vprog是用户定义的顶点程序，它在每个顶点上运行，并接收传入的消息并计算新的顶点值。在第一次迭代时，顶点程序被应用于所有顶点，并传递默认的消息。在后续迭代中，只有收到消息的顶点才会调用顶点程序。
  6. sendMsg是用户提供的函数，应用于当前迭代中收到消息的顶点的出边。
  7. mergeMsg是用户提供的函数，将两个类型为 A 的传入消息合并为一个类型为 A 的单个消息。
  8. 返回计算结束后的图对象。

object GraphXUtils

object GraphXUtils {/**
   * 注册GraphX使用的类到Kryo。
   */def registerKryoClasses(conf: SparkConf):Unit={
    conf.registerKryoClasses(Array(
      classOf[Edge[Object]],
      classOf[(VertexId, Object)],
      classOf[EdgePartition[Object, Object]],
      classOf[ShippableVertexPartition[Object]],
      classOf[RoutingTablePartition],
      classOf[BitSet],
      classOf[VertexIdToIndexMap],
      classOf[VertexAttributeBlock[Object]],
      classOf[PartitionStrategy],
      classOf[BoundedPriorityQueue[Object]],
      classOf[EdgeDirection],
      classOf[GraphXPrimitiveKeyOpenHashMap[VertexId,Int]],
      classOf[OpenHashSet[Int]],
      classOf[OpenHashSet[Long]]))}/**
   * 将过时的API映射到新API的代理方法。
   */private[graphx]def mapReduceTriplets[VD: ClassTag, ED: ClassTag, A: ClassTag](
      g: Graph[VD, ED],
      mapFunc: EdgeTriplet[VD, ED]=> Iterator[(VertexId, A)],
      reduceFunc:(A, A)=> A,
      activeSetOpt: Option[(VertexRDD[_], EdgeDirection)]= None): VertexRDD[A]={def sendMsg(ctx: EdgeContext[VD, ED, A]):Unit={
      mapFunc(ctx.toEdgeTriplet).foreach { kv =>val id = kv._1
        val msg = kv._2
        if(id == ctx.srcId){
          ctx.sendToSrc(msg)}else{
          assert(id == ctx.dstId)
          ctx.sendToDst(msg)}}}
    g.aggregateMessagesWithActiveSet(
      sendMsg, reduceFunc, TripletFields.All, activeSetOpt)}}

以上是

GraphXUtils

对象的代码。其中包含了以下两个方法：

  • registerKryoClasses：用于将GraphX使用的类注册到Kryo序列化器。
  • mapReduceTriplets：一个代理方法，用于映射过时的API到新的API。它接受一个图、一个映射函数、一个合并函数以及一个可选的活跃顶点集合，并返回一个顶点RDD。

这些方法都是为了支持GraphX的内部计算和消息传递操作而设计的。

class Graph

/**
 * Graph类抽象地表示了一个具有任意对象关联的顶点和边的图。该图提供了基本操作来访问和操作与顶点和边相关联的数据以及底层结构。与Spark RDD类似，图是一个功能性数据结构，在进行变异操作时会返回新的图。
 *
 * @note [[GraphOps]]包含了额外的方便操作和图算法。
 *
 * @tparam VD 顶点属性类型
 * @tparam ED 边属性类型
 */abstractclass Graph[VD: ClassTag, ED: ClassTag]protected()extends Serializable {/**
   * 一个包含顶点及其关联属性的RDD。
   *
   * @note 顶点ID是唯一的。
   * @return 包含此图中顶点的RDD
   */val vertices: VertexRDD[VD]/**
   * 一个包含边及其关联属性的RDD。RDD中的条目仅包含源ID、目标ID和边数据。
   *
   * @return 包含此图中边的RDD
   *
   * @see `Edge`获取边类型。
   * @see `Graph#triplets`获取包含所有边及其顶点数据的RDD。
   *
   */val edges: EdgeRDD[ED]/**
   * 包含边三元组的RDD，即带有与相邻顶点关联的顶点数据的边。如果不需要顶点数据，应使用`edges`方法，即仅需要边数据和相邻顶点ID。
   *
   * @return 包含边三元组的RDD
   *
   * @example 此操作可用于评估图着色，我们想要检查两个顶点是否具有不同的颜色。
   * {{{
   * type Color = Int
   * val graph: Graph[Color, Int] = GraphLoader.edgeListFile("hdfs://file.tsv")
   * val numInvalid = graph.triplets.map(e => if (e.src.data == e.dst.data) 1 else 0).sum
   * }}}
   */val triplets: RDD[EdgeTriplet[VD, ED]]/**
   * 使用指定的存储级别将顶点和边缓存在内存中，忽略之前设置的目标存储级别。
   *
   * @param newLevel 缓存图的级别。
   *
   * @return 为方便起见，返回对此图的引用。
   */def persist(newLevel: StorageLevel = StorageLevel.MEMORY_ONLY): Graph[VD, ED]/**
   * 使用之前指定的目标存储级别，将顶点和边缓存在内存中，默认为`MEMORY_ONLY`。这可用于将图固定在内存中，以便多个查询可以重用相同的构建过程。
   */def cache(): Graph[VD, ED]/**
   * 对此图进行检查点标记。它将保存到SparkContext.setCheckpointDir()设置的检查点目录中，并删除其父RDD的所有引用。强烈建议将此图持久化在内存中，否则将需要重新计算才能将其保存到文件中。
   */def checkpoint():Unit/**
   * 返回此图是否已经进行了检查点标记。
   * 当且仅当顶点RDD和边RDD都已经进行了检查点标记时，返回true。
   */def isCheckpointed:Boolean/**
   * 获取此图被检查点标记后保存的文件名。
   * （顶点RDD和边RDD会被单独检查点标记。）
   */def getCheckpointFiles: Seq[String]/**
   * 取消缓存图的顶点和边。这对于在每次迭代中构建新图的迭代算法非常有用。
   *
   * @param blocking 是否阻塞直到所有数据都未缓存（默认值为false）
   */def unpersist(blocking:Boolean=false): Graph[VD, ED]/**
   * 只取消缓存图的顶点，而保留边不变。这对于修改顶点属性但重用边的迭代算法非常有用。此方法可用于在不再需要之前迭代的顶点属性时，取消缓存先前迭代的顶点属性，从而提高GC性能。
   *
   * @param blocking 是否阻塞直到所有数据都未缓存（默认值为false）
   */def unpersistVertices(blocking:Boolean=false): Graph[VD, ED]/**
   * 根据`partitionStrategy`对图中的边进行重新分区。
   *
   * @param partitionStrategy 分区策略，在对图的边进行分区时使用。
   */def partitionBy(partitionStrategy: PartitionStrategy): Graph[VD, ED]/**
   * 根据`partitionStrategy`对图中的边进行重新分区。
   *
   * @param partitionStrategy 分区策略，在对图的边进行分区时使用。
   * @param numPartitions 新图中的边分区数。
   */def partitionBy(partitionStrategy: PartitionStrategy, numPartitions:Int): Graph[VD, ED]/**
   * 使用map函数转换图中的每个顶点属性。
   *
   * @note 新图具有相同的结构。因此，底层索引结构可以重用。
   *
   * @param map 从顶点对象到新顶点值的函数
   *
   * @tparam VD2 新顶点数据类型
   *
   * @example 我们可以使用此操作将顶点值从一种类型更改为另一种类型以初始化算法。
   * {{{
   * val rawGraph: Graph[(), ()] = Graph.textFile("hdfs://file")
   * val root = 42
   * var bfsGraph = rawGraph.mapVertices[Int]((vid, data) => if (vid == root) 0 else Math.MaxValue)
   * }}}
   *
   */def mapVertices[VD2: ClassTag](map:(VertexId, VD)=> VD2)(implicit eq: VD =:= VD2 =null): Graph[VD2, ED]/**
   * 使用map函数将图中的每个边属性进行转换。map函数不会传递与边相邻的顶点值。
   * 如果需要顶点值，请使用`mapTriplets`。
   *
   * @note 此图不会更改，并且新图具有相同的结构。因此，可以重用底层索引结构。
   *
   * @param map 将边对象映射为新边值的函数。
   *
   * @tparam ED2 新的边数据类型
   *
   * @example 此函数可用于初始化边属性。
   *
   */def mapEdges[ED2: ClassTag](map: Edge[ED]=> ED2): Graph[VD, ED2]={
    mapEdges((pid, iter)=> iter.map(map))}/**
   * 使用map函数将每个边属性转换，一次传递一个分区。map函数接收到一个迭代器，该迭代器包含逻辑分区内的所有边，
   * 并且应返回一个新的迭代器，其中包含每个边的新值，其元素必须与旧迭代器的元素一一对应。
   * 如果需要相邻顶点值，请使用`mapTriplets`。
   *
   * @note 这不会更改图的结构或修改该图的值。因此，可以重用底层索引结构。
   *
   * @param map 接收分区ID和一个迭代器作为参数的函数，迭代器包含分区中的所有边，
   *            并且必须返回一个迭代器，其中包含输入迭代器的元素的新值。
   *
   * @tparam ED2 新的边数据类型
   *
   */def mapEdges[ED2: ClassTag](map:(PartitionID, Iterator[Edge[ED]])=> Iterator[ED2]): Graph[VD, ED2]/**
   * 使用map函数将每个边属性转换，并传递相邻的顶点属性。如果不需要相邻顶点值，请考虑使用`mapEdges`。
   *
   * @note 这不会更改图的结构或修改该图的值。因此，可以重用底层索引结构。
   *
   * @param map 将边对象映射为新边值的函数。
   *
   * @tparam ED2 新的边数据类型
   *
   * @example 此函数可用于基于与每个顶点关联的属性初始化边属性。
   * {{{
   * val rawGraph: Graph[Int, Int] = someLoadFunction()
   * val graph = rawGraph.mapTriplets[Int]( edge =>
   *   edge.src.data - edge.dst.data)
   * }}}
   *
   */def mapTriplets[ED2: ClassTag](map: EdgeTriplet[VD, ED]=> ED2): Graph[VD, ED2]={
    mapTriplets((pid, iter)=> iter.map(map), TripletFields.All)}/**
   * 使用map函数将每个边属性转换，并传递相邻的顶点属性。如果不需要相邻顶点值，请考虑使用`mapEdges`。
   *
   * @note 这不会更改图的结构或修改该图的值。因此，可以重用底层索引结构。
   *
   * @param map 将边对象映射为新边值的函数。
   * @param tripletFields 应该包含在传递给map函数的边三元组中的字段。如果不需要所有字段，指定此项可以提高性能。
   *
   * @tparam ED2 新的边数据类型
   *
   * @example 此函数可用于基于与每个顶点关联的属性初始化边属性。
   * {{{
   * val rawGraph: Graph[Int, Int] = someLoadFunction()
   * val graph = rawGraph.mapTriplets[Int]( edge =>
   *   edge.src.data - edge.dst.data)
   * }}}
   *
   */def mapTriplets[ED2: ClassTag](
      map: EdgeTriplet[VD, ED]=> ED2,
      tripletFields: TripletFields): Graph[VD, ED2]={
    mapTriplets((pid, iter)=> iter.map(map), tripletFields)}/**
   * 使用map函数将每个边属性进行转换，一次传递一个分区，并传递相邻的顶点属性。
   * map函数接收到一个迭代器，该迭代器包含逻辑分区内的所有边三元组，
   * 并且应返回一个新的迭代器，其中包含每个边的新值，其元素必须与旧迭代器的元素一一对应。
   * 如果不需要相邻顶点值，请考虑使用`mapEdges`。
   *
   * @note 这不会更改图的结构或修改该图的值。因此，可以重用底层索引结构。
   *
   * @param map 迭代器转换函数
   * @param tripletFields 应该包含在传递给map函数的边三元组中的字段。如果不需要所有字段，指定此项可以提高性能。
   *
   * @tparam ED2 新的边数据类型
   *
   */def mapTriplets[ED2: ClassTag](
      map:(PartitionID, Iterator[EdgeTriplet[VD, ED]])=> Iterator[ED2],
      tripletFields: TripletFields): Graph[VD, ED2]/**
   * 反转图中的所有边。如果图中存在从a到b的边，则返回的图中将包含从b到a的边。
   */def reverse: Graph[VD, ED]/**
   * 将图限制为仅满足谓词的顶点和边。生成的子图满足以下条件：
   *
   * {{{
   * V' = {v : for all v in V where vpred(v)}
   * E' = {(u,v): for all (u,v) in E where epred((u,v)) && vpred(u) && vpred(v)}
   * }}}
   *
   * @param epred 边谓词，它接收一个边三元组并在边保留在子图中时返回true。请注意，仅考虑满足顶点谓词的边。
   *
   * @param vpred 顶点谓词，它接收一个顶点对象，并在将顶点包含在子图中时返回true
   *
   * @return 仅包含满足谓词的顶点和边的子图
   */def subgraph(
      epred: EdgeTriplet[VD, ED]=>Boolean=(x =>true),
      vpred:(VertexId, VD)=>Boolean=((v, d)=>true)): Graph[VD, ED]/**
   * 将图限制为仅包含在`other`图中的顶点和边，但保留来自此图的属性。
   * @param other 要投影当前图形到的图形
   * @return 包含只在当前图和`other`图中存在的顶点和边的图形，其中顶点和边的数据来自当前图
   */def mask[VD2: ClassTag, ED2: ClassTag](other: Graph[VD2, ED2]): Graph[VD, ED]/**
 * 将两个顶点之间的多个边合并为单个边。为了得到正确的结果，图必须使用`partitionBy`进行分区。
 *
 * @param merge 用于合并重复边的边属性的用户提供的可交换、可结合函数。
 *
 * @return 每个（源，目标）顶点对应的单个边的结果图。
 */def groupEdges(merge:(ED, ED)=> ED): Graph[VD, ED]/**
 * 从每个顶点的相邻边和相邻顶点中聚合值。在图的每条边上调用用户提供的`sendMsg`函数，生成0或多个消息发送到边上的任一顶点。
 * 然后使用`mergeMsg`函数来合并发送给同一个顶点的所有消息。
 *
 * @tparam A 发送到每个顶点的消息的类型
 *
 * @param sendMsg 在每条边上运行，使用[[EdgeContext]]向相邻顶点发送消息。
 * @param mergeMsg 用于合并由`sendMsg`发送给同一顶点的消息。此组合器应该是可交换和可结合的。
 * @param tripletFields 应该包含在传递给`sendMsg`函数的[[EdgeContext]]中的字段。如果不需要所有字段，指定这些可以提高性能。
 *
 * @example 我们可以使用此函数计算每个顶点的入度
 * {{{
 * val rawGraph: Graph[_, _] = Graph.textFile("twittergraph")
 * val inDeg: RDD[(VertexId, Int)] =
 *   rawGraph.aggregateMessages[Int](ctx => ctx.sendToDst(1), _ + _)
 * }}}
 *
 * @note 通过在边级别表达计算，我们实现了最大并行性。这是图API中实现邻域级别计算的核心函数之一。
 * 例如，此函数可用于计算满足谓词的邻居数量或实现PageRank。
 *
 */def aggregateMessages[A: ClassTag](
    sendMsg: EdgeContext[VD, ED, A]=>Unit,
    mergeMsg:(A, A)=> A,
    tripletFields: TripletFields = TripletFields.All): VertexRDD[A]={
  aggregateMessagesWithActiveSet(sendMsg, mergeMsg, tripletFields, None)}/**
 * 从每个顶点的相邻边和相邻顶点中聚合值。在图的每条边上调用用户提供的`sendMsg`函数，生成0或多个消息发送到边上的任一顶点。
 * 然后使用`mergeMsg`函数来合并发送给同一个顶点的所有消息。
 *
 * 此变体可以接受一个活动集来限制计算，并且仅供内部使用。
 *
 * @tparam A 发送到每个顶点的消息的类型
 *
 * @param sendMsg 在每条边上运行，使用[[EdgeContext]]向相邻顶点发送消息。
 * @param mergeMsg 用于合并由`sendMsg`发送给同一顶点的消息。此组合器应该是可交换和可结合的。
 * @param tripletFields 应该包含在传递给`sendMsg`函数的[[EdgeContext]]中的字段。如果不需要所有字段，指定这些可以提高性能。
 * @param activeSetOpt 用于在需要时在边的子集上运行聚合的有效方法。这是通过指定一组“活动”顶点和一个边方向来完成的。
 *   然后，`sendMsg`函数仅在以指定方向连接到活动顶点的边上运行。如果方向是`In`，则`sendMsg`仅在目标在活动集中的边上运行。
 *   如果方向是`Out`，则`sendMsg`仅在源自活动集中的顶点的边上运行。如果方向是`Either`，则`sendMsg`将在具有*任一*顶点在活动集中的边上运行。
 *   如果方向是`Both`，则`sendMsg`将在具有*两个*顶点在活动集中的边上运行。活动集必须具有与图的顶点相同的索引。
 */private[graphx]def aggregateMessagesWithActiveSet[A: ClassTag](
    sendMsg: EdgeContext[VD, ED, A]=>Unit,
    mergeMsg:(A, A)=> A,
    tripletFields: TripletFields,
    activeSetOpt: Option[(VertexRDD[_], EdgeDirection)]): VertexRDD[A]/**
 * 将顶点与`table` RDD中的条目进行连接，并使用`mapFunc`合并结果。
 * 输入表格应该最多包含每个顶点的一个条目。如果没有为图中的特定顶点提供`other`条目，则映射函数接收到`None`。
 *
 * @tparam U 更新表格中的条目类型
 * @tparam VD2 新顶点值类型
 *
 * @param other 要与图中的顶点连接的表格。表格应该最多包含每个顶点的一个条目。
 * @param mapFunc 用于计算新顶点值的函数。
 *                映射函数对所有顶点都进行调用，甚至那些在表格中没有相应条目的顶点也会调用。
 *
 * @example 此函数用于基于外部数据更新顶点的新值。
 *          例如，我们可以将出度添加到每个顶点记录中：
 *
 * {{{
 * val rawGraph: Graph[_, _] = Graph.textFile("webgraph")
 * val outDeg: RDD[(VertexId, Int)] = rawGraph.outDegrees
 * val graph = rawGraph.outerJoinVertices(outDeg) {
 *   (vid, data, optDeg) => optDeg.getOrElse(0)
 * }
 * }}}
 */def outerJoinVertices[U: ClassTag, VD2: ClassTag](other: RDD[(VertexId, U)])(mapFunc:(VertexId, VD, Option[U])=> VD2)(implicit eq: VD =:= VD2 =null): Graph[VD2, ED]val ops =new GraphOps(this)}// end of Graph

object Graph

/**
 * Graph对象包含一组用于从RDD构建图的常规操作。
 */object Graph {/**
   * 从以顶点id对编码的边的集合构建图。
   *
   * @param rawEdges 边的集合，格式为(src, dst)
   * @param defaultValue 用于创建由边引用的顶点的属性
   * @param uniqueEdges 如果找到多个相同的边，则将它们合并，并且边属性设置为它们的和。否则，重复的边被视为独立的。要启用`uniqueEdges`，必须提供一个[[PartitionStrategy]]。
   * @param edgeStorageLevel 如果需要，所需的缓存边的存储级别
   * @param vertexStorageLevel 如果需要，所需的缓存顶点的存储级别
   *
   * @return 一个包含边属性为重复边的计数或1（如果`uniqueEdges`为`None`）和顶点属性为每个顶点的总度的图
   */def fromEdgeTuples[VD: ClassTag](
      rawEdges: RDD[(VertexId, VertexId)],
      defaultValue: VD,
      uniqueEdges: Option[PartitionStrategy]= None,
      edgeStorageLevel: StorageLevel = StorageLevel.MEMORY_ONLY,
      vertexStorageLevel: StorageLevel = StorageLevel.MEMORY_ONLY): Graph[VD,Int]={val edges = rawEdges.map(p => Edge(p._1, p._2,1))val graph = GraphImpl(edges, defaultValue, edgeStorageLevel, vertexStorageLevel)
    uniqueEdges match{case Some(p)=> graph.partitionBy(p).groupEdges((a, b)=> a + b)case None => graph
    }}/**
   * 从边的集合构建图。
   *
   * @param edges 包含图中所有边的RDD
   * @param defaultValue 每个顶点使用的默认顶点属性
   * @param edgeStorageLevel 如果需要，所需的缓存边的存储级别
   * @param vertexStorageLevel 如果需要，所需的缓存顶点的存储级别
   *
   * @return 一个由`edges`描述的边属性和由`edges`中的所有顶点和值为`defaultValue`的顶点组成的图
   */def fromEdges[VD: ClassTag, ED: ClassTag](
      edges: RDD[Edge[ED]],
      defaultValue: VD,
      edgeStorageLevel: StorageLevel = StorageLevel.MEMORY_ONLY,
      vertexStorageLevel: StorageLevel = StorageLevel.MEMORY_ONLY): Graph[VD, ED]={
    GraphImpl(edges, defaultValue, edgeStorageLevel, vertexStorageLevel)}/**
   * 从一组顶点和带有属性的边构建图。任意选择重复的顶点，并且在边集合中但不在输入顶点中找到的顶点被分配默认属性。
   *
   * @tparam VD 顶点属性类型
   * @tparam ED 边属性类型
   * @param vertices "集"中的顶点及其属性
   * @param edges 图中的边的集合
   * @param defaultVertexAttr 对于在边中但不在顶点中提到的顶点，使用的默认顶点属性
   * @param edgeStorageLevel 如果需要，所需的缓存边的存储级别
   * @param vertexStorageLevel 如果需要，所需的缓存顶点的存储级别
   */def apply[VD: ClassTag, ED: ClassTag](
      vertices: RDD[(VertexId, VD)],
      edges: RDD[Edge[ED]],
      defaultVertexAttr: VD =null.asInstanceOf[VD],
      edgeStorageLevel: StorageLevel = StorageLevel.MEMORY_ONLY,
      vertexStorageLevel: StorageLevel = StorageLevel.MEMORY_ONLY): Graph[VD, ED]={
    GraphImpl(vertices, edges, defaultVertexAttr, edgeStorageLevel, vertexStorageLevel)}/**
   * 从图中隐式提取[[GraphOps]]成员。
   *
   * 为了提高模块化性，Graph类型只包含一小组基本操作。所有方便的操作都定义在[[GraphOps]]类中，它可以在多个图实现之间共享。
   */implicitdef graphToGraphOps[VD: ClassTag, ED: ClassTag](g: Graph[VD, ED]): GraphOps[VD, ED]= g.ops
}// Graph对象结束

这段源码是定义了一个名为Graph的对象，用于构建和操作图。下面对每个方法进行简要说明：

  1. fromEdgeTuples方法：从以顶点id对编码的边的集合构建图。其中参数rawEdges表示边的集合，defaultValue表示顶点的默认属性，uniqueEdges表示是否合并重复的边，edgeStorageLevel和vertexStorageLevel表示缓存边和顶点的存储级别。
  2. fromEdges方法：从边的集合构建图。其中参数edges表示包含图中所有边的RDD，defaultValue表示每个顶点使用的默认顶点属性，edgeStorageLevel和vertexStorageLevel表示缓存边和顶点的存储级别。
  3. apply方法：从一组顶点和带有属性的边构建图。其中参数vertices表示顶点的RDD，edges表示边的RDD，defaultVertexAttr表示在边中但不在顶点中提到的顶点使用的默认顶点属性，edgeStorageLevel和vertexStorageLevel表示缓存边和顶点的存储级别。
  4. graphToGraphOps方法：从图中隐式提取GraphOps成员。Graph类型只包含基本操作，而方便的操作定义在GraphOps类中，该方法用于将Graph转换为GraphOps。

论文链接

https://kowshik.github.io/JPregel/pregel_paper.pdf

高频引用文章

  1. Gradle中文实用教程【持续更新收藏版】
  2. spark ML机器学习 spark原理示例用法源码学习总结目录【珍藏版】
标签： spark java 算法
本文转载自: https://blog.csdn.net/wang2leee/article/details/135661691
版权归原作者 BigDataMLApplication 所有， 如有侵权，请联系我们删除。
登录后发布评论

“Spark Graphx Pregel原理方法示例源码详解”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

NGCBot-打造基于Hook机制的微信机器人，Windows本地部署

Docker完整卸载的两种方法详细教程例子解析

spark-libFM 项目常见问题解决方案

基于大数据 Python NBA球员数据分析系统(源码+LW+部署讲解+数据库+ppt)

软件测试笔记｜web自动化测试｜Web 自动化测试中，XPath 定位的方法有哪些？如何编写 XPath？

Spring Boot 日志打印配置详解

MySQL中日期和时间戳的转换：字符到DATE和TIMESTAMP的相互转换

文章导航