自定义spark GraphX中的collectNeighborIds方法

Posted 佟学强

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了自定义spark GraphX中的collectNeighborIds方法相关的知识,希望对你有一定的参考价值。

/**
* 自定义收集VertexId的neighborIds
* @author TongXueQiang
*/
def collectNeighborIds[T,U](edgeDirection:EdgeDirection,graph:Graph[T,U])(implicit m:scala.reflect.ClassTag[T],n:scala.reflect.ClassTag[U]):VertexRDD[Array[VertexId]] = {
val nbrs = graph.mapReduceTriplets[Array[VertexId]](
//map函数
edgeTriplets => {
val msgTosrc = (edgeTriplets.srcId,Array(edgeTriplets.dstId));
val msgTodst = (edgeTriplets.dstId,Array(edgeTriplets.srcId));
edgeDirection match {
case EdgeDirection.Either =>Iterator(msgTosrc,msgTodst)
case EdgeDirection.Out => Iterator(msgTosrc)
case EdgeDirection.In => Iterator(msgTodst)
case EdgeDirection.Both => throw new SparkException("It doesn‘t make sense to collect neighbors without a " + "direction.(EdgeDirection.Both is not supported.use EdgeDirection.Either instead.)")
}
},_ ++ _)//reduce函数
nbrs
}
测试:
object Test {
  
  System.setProperty("hadoop.home.dir","D://hadoop-2.6.2");
  val conf = new SparkConf().setMaster("local").setAppName("SparkGraph");
  val sc = new SparkContext(conf);

  def main(args:Array[String]):Unit = {
    val graph = GraphGenerators.logNormalGraph(sc,numVertices = 100).map((id,_) => id.toDouble);
    collectNeighborIds(EdgeDirection.In,graph).foreach(line => {print(line._1+":"); for (elem <- line._2) {print(elem + " ")};println;});

}



}

以上是关于自定义spark GraphX中的collectNeighborIds方法的主要内容,如果未能解决你的问题,请参考以下文章

Spark项目---利用GraphX解决飞机航班问题

Spark中文手册8:spark GraphX编程指南

Spark图处理GraphX学习笔记!

Spark GraphX 聚合操作

Spark Graphx图计算案例实战之aggregateMessages求社交网络中的最大年纪追求者和平均年纪!

Spark GraphX学习笔记