在 spark 和 graphx 中使用 mapReduceTriplets 将函数应用于图形数据

Question

我在使用 graphx 将 mapReduceTriplets 应用到我在 spark 中的图形网络时遇到了一些问题。

我一直在学习教程并读入我自己的数据，这些数据以 [Array[String],Int] 的形式组合在一起，所以例如我的顶点是：

org.apache.spark.graphx.VertexRDD[Array[String]] 例如(3999,数组(17,低,9))

我的优势是：

org.apache.spark.graphx.EdgeRDD[Int] 例如边(3999,4500,1)

我正在尝试使用 mapReduceTriplets 应用聚合类型函数，该函数计算顶点数组中的最后一个整数（在上面的示例 9 中）与第一个整数（在以上示例 17) 的所有连接顶点。

所以你最终会得到一个匹配或不匹配数量的计数列表。

我遇到的问题是使用 mapReduceTriplets 应用任何函数，我对 scala 很陌生，所以这可能非常明显，但在 graphx 教程中有一个示例，它使用格式为 Graph[Double] 的图形, Int]，但是我的图表采用 Graph[Array[String],Int] 的格式，所以我只是作为第一步尝试弄清楚如何在示例中使用我的图表，然后从那里开始工作。

graphx网站上的例子如下：

    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
)

任何建议将不胜感激，或者如果您认为有比使用 mapreducetriplets 更好的方法，我将很高兴听到。

编辑了新代码

val nodes = (sc.textFile("C~nodeData.csv")
.map(line => line.split(",")).map( parts => (parts.head.toLong, parts.tail) ))

val edges = GraphLoader.edgeListFile(sc, "C:~edges.txt")


val graph = edges.outerJoinVertices(nodes) {
case (uid, deg, Some(attrList)) => attrList
case (uid, deg, None) => Array.empty[String]
}


val countsRdd = graph.collectNeighbors(EdgeDirection.Either).leftOuterJoin(graph.vertices).map {
  case (id, t) => {
    val neighbors: Array[(VertexId, Array[String])] = t._1
    val nodeAttr = (t._2)
    neighbors.map(_._2).count( x => x.apply(x.size - 1) == nodeAttr(0))

  }
}

Answer 1

我认为您想使用 GraphOps.collectNeighbors 而不是 mapReduceTriplets 或 aggregateMessages。

collectNeighbors 将为您提供一个 RDD，对于图中的每个 VertexId，连接的节点作为一个数组。只需根据您的需要减少 Array 即可。类似于：

val countsRdd = graph.collectNeighbors(EdgeDirection.Either)
  .join(graph.vertices)
  .map{ case (vid,t) => {
    val neighbors = t._1
    val nodeAttr = t._2
    neighbors.map(_._2).filter( <add logic here> ).size
  }

如果这不能让您朝着正确的方向前进，或者您遇到困难，请告诉我（例如，“”部分）。

在 spark 和 graphx 中使用 mapReduceTriplets 将函数应用于图形数据

applying a function to graph data using mapReduceTriplets in spark and graphx

scala

network-programming

apache-spark

graph-algorithm

spark-graphx