Spark GraphX social graph analysis

Author: artem0

.gitignore

@@ -1,2 +1,5 @@
 *.class
 *.log
+/.idea/
+/target/
+/project/target/

build.sbt

@@ -0,0 +1,10 @@
+name := "spark-course"
+
+version := "1.0"
+
+scalaVersion := "2.11.8"
+
+libraryDependencies ++= Seq(
+  "org.apache.spark" %% "spark-core" % "2.0.0",
+  "org.apache.spark" %% "spark-graphx" % "2.0.0"
+)

project/build.properties

@@ -0,0 +1 @@
+sbt.version = 0.13.15

project/plugins.sbt

@@ -0,0 +1 @@
+logLevel := Level.Warn

project/project/target/config-classes/$cdd3643bef030a3f6703.cache

@@ -0,0 +1 @@
+sbt.internals.DslEntry

src/main/resources/UserGraph.tsv

@@ -0,0 +1,66 @@
+package graphx
+
+import org.apache.spark._
+import org.apache.spark.graphx.{Graph, _}
+
+object InputDataFlow {
+
+  def parseNames(line: String): Option[(VertexId, String)] = {
+    val fields = line.split('\t')
+    if (fields.length > 1)
+      Some(fields(0).trim().toLong, fields(1))
+    else None
+  }
+
+  def makeEdges(line: String) : List[Edge[Int]] = {
+    import scala.collection.mutable.ListBuffer
+    var edges = new ListBuffer[Edge[Int]]()
+    val fields = line.split(" ")
+    val origin = fields(0)
+    (1 until fields.length)
+      .foreach { p => edges += Edge(origin.toLong, fields(p).toLong, 0) }
+    edges.toList
+  }
+
+}
+
+class GraphX(sc: SparkContext) {
+  private def verts = sc.textFile(USER_NAMES_FILE).flatMap(InputDataFlow.parseNames)
+
+  private def edges = sc.textFile(USER_GRAPH_FILE).flatMap(InputDataFlow.makeEdges)
+
+  private def graph = Graph(verts, edges).cache()
+
+
+  def getMostConnectedUsers(amount:Int): Array[(VertexId, (PartitionID, String))] = {
+    graph.degrees.join(verts)
+      .sortBy( {case( (_, (userName, _))) => userName }, ascending=false ).take(amount)
+  }
+
+  private def getBfs(root:VertexId) = {
+    val initialGraph = graph.mapVertices((id, _) => if (id == root) 0.0 else Double.PositiveInfinity)
+
+    val bfs = initialGraph.pregel(Double.PositiveInfinity, 10)(
+      (_, attr, msg) => math.min(attr, msg),
+      triplet => {
+        if (triplet.srcAttr != Double.PositiveInfinity) {
+          Iterator((triplet.dstId, triplet.srcAttr+1))
+        } else {
+          Iterator.empty
+        }
+      },
+      (a,b) => math.min(a,b)).cache()
+    bfs
+  }
+
+  def degreeOfSeparationSingleUser(root:VertexId) = {
+    getBfs(root).vertices.join(verts).take(100)
+  }
+
+  def degreeOfSeparationTwoUser(firstUser:VertexId, secondUser:VertexId ) = {
+    getBfs(firstUser)
+      .vertices
+      .filter{case (vertexId, _) => vertexId == secondUser}
+      .collect
+  }
+}

src/main/resources/UserNames.tsv

@@ -0,0 +1,23 @@
+package graphx
+
+import org.apache.log4j.{Level, Logger}
+import org.apache.spark.SparkContext
+
+object GraphXDemo {
+
+  def main(args: Array[String]): Unit = {
+    Logger.getLogger("org").setLevel(Level.ERROR)
+    val sc = new SparkContext("local[*]", "GraphX")
+
+    val graph = new GraphX(sc)
+
+    println("\nTop 10 most-connected users:")
+    graph.getMostConnectedUsers(10) foreach println
+
+    println("\nComputing degrees of separation for user Arch")
+    graph.degreeOfSeparationSingleUser(5306) foreach println
+
+    println("\nComputing degrees of separation for user Arch and Fred")
+    graph.degreeOfSeparationTwoUser(5306, 14) foreach println
+  }
+}

src/main/scala/graphx/GraphX.scala

@@ -0,0 +1,9 @@
+package object graphx {
+
+  val USER_NAMES_FILE_NAME = "/UserNames.tsv"
+  val USER_GRAPH_FILE_NAME = "/UserGraph.tsv"
+
+  val USER_NAMES_FILE: String = getClass.getResource(USER_NAMES_FILE_NAME).getPath
+  val USER_GRAPH_FILE: String = getClass.getResource(USER_GRAPH_FILE_NAME).getPath
+
+}