fixed bug in calculation truss

Author: ricarda-schueler

graph-mining-flink/src/main/scala/de/hpi/dbda/graph_mining/MaximalTruss.scala

@@ -19,17 +19,17 @@ object MaximalTruss {
     while (k != maxK && k != minK){
 
       print ("############################ k is " + k +" #################################")
-      val filteredGraph = graphs.filter(e => e.vertex1.degree >= k-2 && e.vertex2.degree >= k-2)
-//      filteredGraph.print()
+//      val filteredGraph = graphs.filter(e => e.vertex1.degree >= k-2 && e.vertex2.degree >= k-2)
+////      filteredGraph.print()
 
-      val trusses = Truss.calculateTruss(k, filteredGraph)
+      val trusses = Truss.calculateTruss(k, graphs)
 
       val foundTrusses:DataSet[Edge] = trusses.map{truss =>
-          truss._2.truss = truss._1
+//          truss._2.truss = truss._1
           truss._2
         }
 
-      val trussCount = foundTrusses.count
+      val trussCount = trusses.count()
 
       if ( trussCount == 0){
         val newK = minK + (k-minK)/2
@@ -51,65 +51,8 @@ object MaximalTruss {
 
     }
 
+    print ("############################ final k is " + k +" #################################")
     graphs
 
   }
-
-
-  def maxTruss1(graph: DataSet[Edge], stringk:String, executionEnvironment: ExecutionEnvironment): DataSet[Edge] ={
-
-    var k = stringk.toInt
-    var maxK = 0
-    var minK = 2
-
-    val graphs = graph
-
-    val possibleKs = executionEnvironment.fromCollection(2 until 10000)
-
-    graphs.iterateWithTermination(1000)({ currentGraph =>
-
-      print ("############################ k is " + k +" #################################")
-      val filteredGraph = currentGraph.filter(e => e.vertex1.degree >= k-2 && e.vertex2.degree >= k-2)
-      //      filteredGraph.print()
-
-      val trusses = Truss.calculateTruss(k, filteredGraph)
-
-      val foundTrusses:DataSet[Edge] = trusses.map{truss =>
-        truss._2.truss = truss._1
-        truss._2
-      }
-
-
-      val trussCount = foundTrusses.count()
-
-      if ( trussCount == 0){
-        val newPossibleKs = possibleKs.filter(possibleK => possibleK < k)
-
-        val newK = minK + (k-minK)/2
-        maxK = k
-        k = newK
-        (currentGraph, newPossibleKs)
-
-      } else {
-        val newPossibleKs = possibleKs.filter(possibleK => possibleK > k)
-        if (maxK == 0){
-          val newK = 2*k
-          minK = k
-          k = newK
-        } else {
-          val newK = k + (maxK -k)/2
-          minK = k
-          k = newK
-        }
-        (foundTrusses, newPossibleKs)
-      }
-    })
-
-
-  }
-
-
-
-
-
 }

graph-mining-flink/src/main/scala/de/hpi/dbda/graph_mining/Truss.scala

@@ -15,7 +15,7 @@ import scala.concurrent.duration.durationToPair
  */
 
 case class Vertex(id: Int, degree:Int)
-case class Edge(vertex1:Vertex,vertex2:Vertex, var truss:Int = 1, var triangleCount:Int = -1)
+case class Edge(vertex1:Vertex,vertex2:Vertex, var triangleCount:Int = -1)
 case class Triangle(edge1:Edge, edge2:Edge, edge3:Edge)
 
 
@@ -26,23 +26,21 @@ object Truss {
         val splitted = line.split(seperator)
         val f = new Vertex(splitted(0).toInt, 1)
         val s = new Vertex(splitted(1).toInt, 1)
-        createEdge(f, s, 1)
+        createEdge(f, s)
       }
     ).name("convert Graph")
     convertedGraph
   }
 
-  def createEdge(vert1:Vertex, vert2:Vertex, truss:Int): Edge = {
-    if (vert1.degree < vert2.degree) new Edge(vert1, vert2, truss)
+  def createEdge(vert1:Vertex, vert2:Vertex): Edge = {
+    if (vert1.degree < vert2.degree) new Edge(vert1, vert2)
     else
     if (vert1.degree == vert2.degree && vert1.id < vert2.id)
-      new Edge(vert1, vert2, truss)
-    else new Edge(vert2, vert1, truss)
+      new Edge(vert1, vert2)
+    else new Edge(vert2, vert1)
   }
 
   def addDegrees(graph:DataSet[Edge]): DataSet[Edge] ={
-    val truss = 1
-
     val degrees = graph
       .flatMap{e => List(e.vertex1, e.vertex2)}
       .groupBy(0)
@@ -52,10 +50,10 @@ object Truss {
     val degreedGraph = graph
       .map(e => (e.vertex1.id, e.vertex2.id, e))
       .join(degrees, JoinHint.BROADCAST_HASH_SECOND).where(0).equalTo(0) { //Repartition
-      (e, v) => (e._2, new Edge(v._2, e._3.vertex2, truss))
+      (e, v) => (e._2, new Edge(v._2, e._3.vertex2))
     }
       .join(degrees, JoinHint.BROADCAST_HASH_SECOND).where(0).equalTo(0) {
-      (e, v) => createEdge(e._2.vertex1, v._2, truss)
+      (e, v) => createEdge(e._2.vertex1, v._2)
     }.name("join Degrees")
 
     degreedGraph
@@ -75,7 +73,7 @@ object Truss {
     val allEdges = graph.map(edge => (edge, edge)).name("triangleCalculation: map singlge edges")
 
     //TODO Parameterization of #nodes?
-    val joinStrategy = if (k >= 8) JoinHint.BROADCAST_HASH_SECOND else JoinHint.REPARTITION_HASH_SECOND
+    val joinStrategy = if (k >= 8) JoinHint.BROADCAST_HASH_SECOND else JoinHint.BROADCAST_HASH_SECOND //REPARTITION_HASH_SECOND
 
     val triangles = triads.join(allEdges, joinStrategy).where(0).equalTo(0) {
       (triadPart, edgePart) => Triangle(edgePart._2, triadPart._2, triadPart._3)
@@ -85,7 +83,7 @@ object Truss {
   }
 
   def getOuterTriangleVertices(edge1:Edge, edge2:Edge): Edge ={
-    createEdge(edge1.vertex2, edge2.vertex2, edge1.truss)
+    createEdge(edge1.vertex2, edge2.vertex2)
   }
 
 
@@ -99,6 +97,7 @@ object Truss {
     val triangles = getTriangles(filteredGraph, k)
 
     val filteredTriangles = triangles.iterateWithTermination(Int.MaxValue)({triangles =>
+
       val singleEdges = triangles.flatMap(triangle => List(triangle.edge1, triangle.edge2, triangle.edge3)).map((_, 1)).name("prepare triangle count per edge")
 
       val triangleCountPerEdge = singleEdges.groupBy(0).reduce{
@@ -108,10 +107,11 @@ object Truss {
         edgeInt._1.triangleCount  = edgeInt._2
         edgeInt._1
       }.name("give edge triangle count")
-        .filter(edge => edge.triangleCount >= k-2).name("filter edge with too small triangleCount")
 
       val removableEdges = graph.filter(edge => edge.triangleCount < k-2)
 
+      graph = graph.filter(edge => edge.triangleCount >= k-2).name("filter edge with too small triangleCount")
+
       var joinedtriangles = triangles.join(graph).where({triangle => (triangle.edge1.vertex1, triangle.edge1.vertex2)}).equalTo("vertex1", "vertex2"){
         (triangle, edge) =>
           triangle.edge1.triangleCount = edge.triangleCount

trussMini.txt

@@ -16,3 +16,12 @@
 8	9
 8	10
 10	9
+1	7
+1	8
+2	9
+3	6
+3	10
+4	7
+4	9
+5	8
+5	10