Added: Option to extract schema in advance

Author: LorenzBuehmann

sansa-inference-spark/src/main/scala/net/sansa_stack/inference/spark/forwardchaining/ForwardRuleReasonerRDFS.scala

@@ -10,6 +10,7 @@ import net.sansa_stack.inference.data.RDFTriple
 import net.sansa_stack.inference.rules.RDFSLevel._
 import net.sansa_stack.inference.spark.data.RDFGraph
 import net.sansa_stack.inference.spark.utils.RDDUtils.RDDOps
+import net.sansa_stack.inference.spark.utils.RDFSSchemaExtractor
 import net.sansa_stack.inference.utils.CollectionUtils
 
 /**
@@ -26,6 +27,8 @@ class ForwardRuleReasonerRDFS(sc: SparkContext, parallelism: Int = 2) extends Tr
 
   var level: RDFSLevel = DEFAULT
 
+  var extractSchemaTriplesInAdvance: Boolean = true
+
   override def apply(graph: RDFGraph): RDFGraph = {
     logger.info("materializing graph...")
     val startTime = System.currentTimeMillis()
@@ -34,6 +37,11 @@ class ForwardRuleReasonerRDFS(sc: SparkContext, parallelism: Int = 2) extends Tr
     triplesRDD.cache()
     // RDFS rules dependency was analyzed in \todo(add references) and the same ordering is used here
 
+    // as an optimization, we can extract all schema triples first which avoids to run on the whole dataset
+    // for each schema triple later
+    val schemaTriples = if (extractSchemaTriplesInAdvance) new RDFSSchemaExtractor(sc).extract(triplesRDD)
+                        else triplesRDD
+
 
     // 1. we first compute the transitive closure of rdfs:subPropertyOf and rdfs:subClassOf
 

sansa-inference-spark/src/main/scala/net/sansa_stack/inference/spark/forwardchaining/ForwardRuleReasonerRDFSDataframe.scala

@@ -31,9 +31,9 @@ class ForwardRuleReasonerRDFSDataframe(session: SparkSession, parallelism: Int =
     logger.info("materializing graph...")
     val startTime = System.currentTimeMillis()
 
-    val extractor = new RDFSSchemaExtractor(session)
+    val extractor = new RDFSSchemaExtractor(session.sparkContext)
 
-    var index = extractor.extract(graph)
+    var index = extractor.extractWithIndex(graph)
 
     var triples = graph.toDataFrame(session).alias("DATA")
 

sansa-inference-spark/src/main/scala/net/sansa_stack/inference/spark/utils/RDFSSchemaExtractor.scala

@@ -1,11 +1,13 @@
 package net.sansa_stack.inference.spark.utils
 
 import org.apache.jena.vocabulary.RDFS
+import org.apache.spark.SparkContext
 import org.apache.spark.broadcast.Broadcast
 import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.{DataFrame, SparkSession}
+import org.apache.spark.sql.DataFrame
 
-import net.sansa_stack.inference.spark.data.{RDFGraphDataFrame, RDFGraphNative}
+import net.sansa_stack.inference.data.RDFTriple
+import net.sansa_stack.inference.spark.data.{RDFGraph, RDFGraphDataFrame, RDFGraphNative}
 import net.sansa_stack.inference.utils.{CollectionUtils, Logging}
 
 /**
@@ -20,9 +22,41 @@ import net.sansa_stack.inference.utils.{CollectionUtils, Logging}
   *
   * @author Lorenz Buehmann
   */
-class RDFSSchemaExtractor(session : SparkSession) extends Logging{
+class RDFSSchemaExtractor(sc : SparkContext) extends Logging{
 
-  val properties = List(RDFS.subClassOf, RDFS.subPropertyOf, RDFS.domain, RDFS.range).map(p => p.getURI)
+  val properties = Set(RDFS.subClassOf, RDFS.subPropertyOf, RDFS.domain, RDFS.range).map(p => p.getURI)
+
+  /**
+    * Extracts the RDF graph containing only the schema triples from the RDF graph.
+    *
+    * @param graph the RDF graph
+    * @return the RDF graph containing only the schema triples
+    */
+  def extract(graph: RDFGraph): RDFGraph = {
+    log.info("Started schema extraction...")
+
+    val filteredTriples = graph.triples.filter(t => properties.contains(t.p))
+
+    log.info("Finished schema extraction.")
+
+    new RDFGraph(filteredTriples)
+  }
+
+  /**
+    * Extracts the schema triples from the given triples.
+    *
+    * @param triples the triples
+    * @return the schema triples
+    */
+  def extract(triples: RDD[RDFTriple]): RDD[RDFTriple] = {
+    log.info("Started schema extraction...")
+
+    val filteredTriples = triples.filter(t => properties.contains(t.p))
+
+    log.info("Finished schema extraction.")
+
+    filteredTriples
+  }
 
 
   /**
@@ -32,7 +66,7 @@ class RDFSSchemaExtractor(session : SparkSession) extends Logging{
     * @param graph the RDF graph
     * @return a mapping from the corresponding schema property to the RDD of s-o pairs
     */
-  def extract(graph: RDFGraphNative): Map[String, RDD[(String, String)]] = {
+  def extractWithIndex(graph: RDFGraphNative): Map[String, RDD[(String, String)]] = {
     log.info("Started schema extraction...")
 
     // for each schema property p
@@ -59,7 +93,7 @@ class RDFSSchemaExtractor(session : SparkSession) extends Logging{
     * @param graph the RDF graph
     * @return a mapping from the corresponding schema property to the Dataframe of s-o pairs
     */
-  def extract(graph: RDFGraphDataFrame): Map[String, DataFrame] = {
+  def extractWithIndex(graph: RDFGraphDataFrame): Map[String, DataFrame] = {
     log.info("Started schema extraction...")
 
     // for each schema property p
@@ -87,8 +121,8 @@ class RDFSSchemaExtractor(session : SparkSession) extends Logging{
     * @return a mapping from the corresponding schema property to the broadcast variable that wraps the multimap
     *         with s-o pairs
     */
-  def extractAndDistribute(graph: RDFGraphNative): Map[String, Broadcast[Map[String, Set[String]]]] = {
-    val schema = extract(graph)
+  def extractWithIndexAndDistribute(graph: RDFGraphNative): Map[String, Broadcast[Map[String, Set[String]]]] = {
+    val schema = extractWithIndex(graph)
 
     log.info("Started schema distribution...")
     val index =
@@ -100,7 +134,7 @@ class RDFSSchemaExtractor(session : SparkSession) extends Logging{
         val mmap = CollectionUtils.toMultiMap(rdd.collect())
 
         // broadcast
-        val bv = session.sparkContext.broadcast(mmap)
+        val bv = sc.broadcast(mmap)
 
         // add to index
         (p -> bv)