Clean ups

Author: GezimSejdiu

sansa-examples-spark/src/main/scala/net/sansa_stack/examples/spark/ml/clustering/RDFGraphPIClustering.scala

@@ -14,7 +14,6 @@ import net.sansa_stack.ml.spark.clustering.RDFGraphPowerIterationClustering
 object RDFGraphPIClustering {
 
   def main(args: Array[String]) {
-    //config.in-> path for input nt file or txt file. It may be a local path or hdfs path.
     parser.parse(args, Config()) match {
       case Some(config) =>
         run(config.in, config.out, config.k, config.maxIterations)
@@ -31,24 +30,22 @@ object RDFGraphPIClustering {
       .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
       .getOrCreate()
     System.setProperty("spark.akka.frameSize", "2000")
-    
 
     println("============================================")
     println("| Power Iteration Clustering   example     |")
     println("============================================")
 
     val lang = Lang.NTRIPLES
     val triples = spark.rdf(lang)(input)
-    
+
     val graph = triples.asStringGraph()
-    
+
     val cluster = RDFGraphPowerIterationClustering(spark, graph, output, k, maxIterations)
     cluster.saveAsTextFile(output)
 
     spark.stop
 
   }
-  
 
   case class Config(in: String = "", out: String = "", k: Int = 2, maxIterations: Int = 5)
 
@@ -59,14 +56,13 @@ object RDFGraphPIClustering {
     head("PowerIterationClusteringExample: an example PIC app using concentric circles.")
 
     opt[String]('i', "input").required().valueName("<path>")
-      .text(s"path to file that contains the input files (in N-Triple format)")
+      .text(s"path (local/hdfs) to file that contains the input files (in N-Triple format)")
       .action((x, c) => c.copy(in = x))
 
     opt[String]('o', "out").required().valueName("<directory>").
       action((x, c) => c.copy(out = x)).
       text("the output directory")
 
-
     opt[Int]('k', "k")
       .text(s"number of circles (/clusters), default: ${defaultParams.k}")
       .action((x, c) => c.copy(k = x))

sansa-examples-spark/src/main/scala/net/sansa_stack/examples/spark/query/GraphQuery.scala

@@ -1,7 +1,7 @@
 package net.sansa_stack.examples.spark.query
 
 import net.sansa_stack.query.spark.graph.jena.SparqlParser
-import net.sansa_stack.query.spark.graph.jena.model.{IntermediateResult, SparkExecutionModel, Config => modelConfig}
+import net.sansa_stack.query.spark.graph.jena.model.{ IntermediateResult, SparkExecutionModel, Config => modelConfig }
 import net.sansa_stack.rdf.spark.partition.graph.algo._
 import org.apache.jena.graph.Node
 import org.apache.jena.riot.Lang
@@ -16,7 +16,7 @@ object GraphQuery {
 
     parser.parse(args, Config()) match {
       case Some(config) => run(config)
-      case None => println(parser.usage)
+      case None         => println(parser.usage)
     }
   }
 
@@ -50,85 +50,85 @@ object GraphQuery {
 
     // Set number of partitions (if config.numParts is 0, number of partitions equals to that of previous graph)
     config.numParts match {
-      case 0 => numParts = prevG.edges.partitions.length
+      case 0     => numParts = prevG.edges.partitions.length
       case other => numParts = other
     }
 
     config.numIters match {
-      case 0 =>
+      case 0     =>
       case other => numIters = other
     }
 
     var partAlgo: PartitionAlgo[Node, Node] = null
 
     config.algo match {
       case "SSHP" =>
-        if(numIters == 0){
+        if (numIters == 0) {
           // Partition algorithm will use default number of iterations
           partAlgo = new SubjectHashPartition[Node, Node](prevG, session, numParts)
         } else {
           partAlgo = new SubjectHashPartition[Node, Node](prevG, session, numParts).setNumIterations(numIters)
         }
         msg = "Start to execute subject semantic hash partitioning"
       case "OSHP" =>
-        if(numIters == 0){
+        if (numIters == 0) {
           partAlgo = new ObjectHashPartition[Node, Node](prevG, session, numParts)
         } else {
           partAlgo = new ObjectHashPartition[Node, Node](prevG, session, numParts).setNumIterations(numIters)
         }
         msg = "Start to execute object semantic hash partitioning"
       case "SOSHP" =>
-        if(numIters == 0){
+        if (numIters == 0) {
           partAlgo = new SOHashPartition[Node, Node](prevG, session, numParts)
         } else {
           partAlgo = new SOHashPartition[Node, Node](prevG, session, numParts).setNumIterations(numIters)
         }
         msg = "Start to execute subject-object semantic hash partitioning"
       case "PP" =>
-        if(numIters == 0){
+        if (numIters == 0) {
           partAlgo = new PathPartition[Node, Node](prevG, session, numParts)
         } else {
           partAlgo = new PathPartition[Node, Node](prevG, session, numParts).setNumIterations(numIters)
         }
         msg = "Start to execute path partitioning"
-      case "" =>
+      case ""    =>
       case other => println(s"the input $other doesn't match any options, no algorithm will be applied.")
     }
 
     var start = 0L
     var end = 0L
 
-    if(partAlgo != null) {
+    if (partAlgo != null) {
       log.info(msg)
       start = System.currentTimeMillis()
       g = partAlgo.partitionBy().cache()
-      //SparkExecutionModel.loadGraph(g)
+      SparkExecutionModel.loadGraph(g)
       end = System.currentTimeMillis()
-      log.info("Graph partitioning execution time: "+Duration(end - start, "millis").toMillis+" ms")
+      log.info("Graph partitioning execution time: " + Duration(end - start, "millis").toMillis + " ms")
     }
 
     // query executing
     log.info("Start to execute queries")
 
-    config.query.foreach{ path =>
-      log.info("Query file: "+path)
+    config.query.foreach { path =>
+      log.info("Query file: " + path)
       modelConfig.setInputQueryFile(path)
       val sp = new SparqlParser(modelConfig.getInputQueryFile)
-      sp.getOps.foreach{ ops =>
+      sp.getOps.foreach { ops =>
         val tag = ops.getTag
-        log.info("Operation "+tag+" start")
+        log.info("Operation " + tag + " start")
         start = System.currentTimeMillis()
         ops.execute()
         end = System.currentTimeMillis()
-        log.info(tag+" execution time: "+Duration(end - start, "millis").toMillis+" ms")
+        log.info(tag + " execution time: " + Duration(end - start, "millis").toMillis + " ms")
       }
     }
 
     // print results to console
-    if(config.print){
+    if (config.print) {
       log.info("print final result")
       val results = IntermediateResult.getFinalResult.cache()
-      if(results.count() >= 10){
+      if (results.count() >= 10) {
         log.info("Too long results(more than 10)")
       } else {
         results.collect().foreach(println(_))