SparkGraphX示例

Programming Guide中的例子

import org.apache.spark.graphx._val users: RDD[(VertexId, (String, String))] &＃61; sc.parallelize(Array((3L, ("rxin", "student")), (7L, ("jgonzal", "postdoc")),(5L, ("franklin", "prof")), (2L, ("istoica", "prof"))))

val relationships: RDD[Edge[String]] &＃61; sc.parallelize(Array(Edge(3L, 7L, "collab"),Edge(5L, 3L, "advisor"),Edge(2L, 5L, "colleague"), Edge(5L, 7L, "pi")))
val graph &＃61; Graph(users, relationships)
val rankedGraph &＃61; graph.staticPageRank(3)

实际的例子&＃xff0c;别踩白快的传输分析&＃xff1a;

val transfer &＃61; sc.textFile("hdfs://LDKJSERVER1046:8020/user/flume/transfer/*/*/*/*.tsv").filter(line &＃61;> line.contains("biecaibaikuai"))val structuredTransferRDD &＃61; transfer.map(line &＃61;> {val trunks &＃61; line.split("\t")if(trunks.length &＃61;&＃61; 35){(trunks(6), trunks(7), trunks(3), trunks(5), trunks(12), trunks(13)) }})
val rdd &＃61; structuredTransferRDD.filter(arg &＃61;> arg !&＃61; ()).map(arg &＃61;> arg.asInstanceOf[(String, String, String, String, String, String)])
val repatitionedRDD &＃61; rdd.repartition(100)
repatitionedRDD.cache
repatitionedRDD.countval mappedRDD &＃61; repatitionedRDD.map(arg &＃61;> ((arg._1, arg._1.hashCode.toLong),(arg._2, arg._2.hashCode.toLong), arg._3, arg._4, arg._5, arg._6))val vertexs &＃61; mappedRDD.flatMap(arg &＃61;> Array((arg._1._2, arg._1._1), (arg._2._2, arg._2._1))).distinct
vertexs.cache
vertexs.count
import org.apache.spark.graphx._
val edges &＃61; mappedRDD.map(arg &＃61;> (arg._1._2, arg._2._2, arg._3)).distinct.map(arg &＃61;> Edge(arg._1, arg._2, arg._3))
edges.cache
edges.count
val graph &＃61; Graph(vertexs, edges)
val rankedGraph &＃61; graph.staticPageRank(3)
rankedGraph.vertices.first
rankedGraph.vertices.filter(arg &＃61;> arg._2 !&＃61; 0.15).first