用spark和graphx中的mapReduceTriplets对图形数据应用函数

Question

我一直在遵循教程，并读取了我自己的数据，这些数据被组合成[array[String]，int]，因此，例如，我的顶点是：

org.apache.spark.graphx.vertexrdd[Array[String]]例如（3999,Array（17,Low,9））

我的边缘是：

因此，您最终会得到一个匹配或不匹配数量的计数列表。

我遇到的问题是使用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
)

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

我认为您应该使用grapops.collectneights而不是MapReduceTriplets或AggregateMessages。

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

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
  }

如果这不能让你朝着正确的方向前进，或者你被卡住了，让我知道（例如“”部分）。