Cleanup and align Semantic-based query for Spark

Author: GezimSejdiu

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

@@ -6,13 +6,13 @@ import java.nio.file.attribute.BasicFileAttributes
 
 import net.sansa_stack.query.spark.semantic.QuerySystem
 import net.sansa_stack.rdf.spark.io._
-import net.sansa_stack.rdf.spark.partition.semantic.RdfPartition
+import net.sansa_stack.rdf.spark.partition._
+import net.sansa_stack.rdf.common.partition.utils.Symbols
 import org.apache.jena.graph.Triple
 import org.apache.jena.riot.Lang
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.SparkSession
 
-
 /**
  * Run SPARQL queries over Spark using Semantic partitioning approach.
  *
@@ -23,83 +23,47 @@ object Semantic {
   def main(args: Array[String]) {
     parser.parse(args, Config()) match {
       case Some(config) =>
-        run(config.in, config.queries, config.partitions, config.out)
+        run(config.in, config.queries, config.out)
       case None =>
         println(parser.usage)
     }
   }
 
-  def run(input: String, queries: String, partitions: String, output: String): Unit = {
+  def run(input: String, queries: String, output: String): Unit = {
 
     println("===========================================")
     println("| SANSA - Semantic Partioning example     |")
     println("===========================================")
 
-    // variables initialization
-    val numOfFilesPartition: Int = 1
-    // val queryInputPath: String = queries //args(1)
-    // val partitionedDataPath: String = "src/main/resources/output/partitioned-data/"
-    // val queryResultPath: String = "src/main/resources/output/query-result/"
-    val symbol = Map(
-      "space" -> " " * 5,
-      "blank" -> " ",
-      "tabs" -> "\t",
-      "newline" -> "\n",
-      "colon" -> ":",
-      "comma" -> ",",
-      "hash" -> "#",
-      "slash" -> "/",
-      "question-mark" -> "?",
-      "exclamation-mark" -> "!",
-      "curly-bracket-left" -> "{",
-      "curly-bracket-right" -> "}",
-      "round-bracket-left" -> "(",
-      "round-bracket-right" -> ")",
-      "less-than" -> "<",
-      "greater-than" -> ">",
-      "at" -> "@",
-      "dot" -> ".",
-      "dots" -> "...",
-      "asterisk" -> "*",
-      "up-arrows" -> "^^")
-
     val spark = SparkSession.builder
       .master("local[*]")
       .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
       .appName("SANSA - Semantic Partioning")
       .getOrCreate()
 
-    // N-Triples reader
     val lang = Lang.NTRIPLES
-    val nTriplesRDD = spark.rdf(lang)(input)
+    val triples = spark.rdf(lang)(input)
 
     println("----------------------")
     println("Phase 1: RDF Partition")
     println("----------------------")
 
-    // class instance: Class RDFPartition and set the partition data
-    val partitionData = new RdfPartition(
-      symbol,
-      nTriplesRDD,
-      partitions,
-      numOfFilesPartition).partitionGraph
+    val partitionData = triples.partitionGraphAsSemantic()
 
     // count total number of N-Triples
-    countNTriples(Left(nTriplesRDD))
+    countNTriples(Left(triples))
     countNTriples(Right(partitionData))
 
-    println(symbol("newline"))
     println("----------------------")
     println("Phase 2: SPARQL System")
     println("----------------------")
 
-    // class instance: Class QuerySystem
     val qs = new QuerySystem(
-      symbol,
+      symbol = Symbols.symbol,
       partitionData,
       input,
       output,
-      numOfFilesPartition)
+      numOfFilesPartition = 1)
     qs.run()
 
     spark.close()
@@ -130,7 +94,7 @@ object Semantic {
       case Right(x) => println(s"Number of N-Triples after partition: ${x.distinct.count}")
     }
   }
-  case class Config(in: String = "", queries: String = "", partitions: String = "", out: String = "")
+  case class Config(in: String = "", queries: String = "", out: String = "")
 
   val parser = new scopt.OptionParser[Config]("SANSA - Semantic Partioning example") {
 
@@ -148,10 +112,6 @@ object Semantic {
       action((x, c) => c.copy(out = x)).
       text("the output directory")
 
-    opt[String]('p', "partitions").required().valueName("<directory>").
-      action((x, c) => c.copy(partitions = x)).
-      text("the partitions directory")
-
     help("help").text("prints this usage text")
   }