Split RDD.

Author: LorenzBuehmann

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

@@ -4,6 +4,7 @@ import org.apache.jena.vocabulary.{RDF, RDFS}
 import org.apache.spark.SparkContext
 import net.sansa_stack.inference.data.RDFTriple
 import net.sansa_stack.inference.spark.data.RDFGraph
+import net.sansa_stack.inference.spark.utils.RDDUtils
 import net.sansa_stack.inference.utils.CollectionUtils
 import org.slf4j.LoggerFactory
 
@@ -52,6 +53,12 @@ class ForwardRuleReasonerRDFS(sc: SparkContext) extends ForwardRuleReasoner{
     val subClassOfMapBC = sc.broadcast(subClassOfMap)
     val subPropertyMapBC = sc.broadcast(subPropertyMap)
 
+    import net.sansa_stack.inference.spark.utils.RDDUtils.RDDOps
+
+    // split by rdf:type
+    val split = triplesRDD.partitionBy(t => t.predicate == RDF.`type`.getURI)
+    var typeTriples = split._1
+    var otherTriples = split._2
 
     // 2. SubPropertyOf inheritance according to rdfs7 is computed
 
@@ -60,14 +67,14 @@ class ForwardRuleReasonerRDFS(sc: SparkContext) extends ForwardRuleReasoner{
             xxx aaa yyy .                   	xxx bbb yyy .
      */
     val triplesRDFS7 =
-      triplesRDD // all triples (s p1 o)
+      otherTriples // all triples (s p1 o)
         .filter(t => subPropertyMapBC.value.contains(t.predicate)) // such that p1 has a super property p2
         .flatMap(t => subPropertyMapBC.value(t.predicate)
         .map(supProp => RDFTriple(t.subject, supProp, t.`object`))) // create triple (s p2 o)
         .setName("rdfs7")
 
     // add triples
-    triplesRDD = triplesRDD.union(triplesRDFS7)
+    otherTriples = otherTriples.union(triplesRDFS7)
 
     // 3. Domain and Range inheritance according to rdfs2 and rdfs3 is computed
 
@@ -80,7 +87,7 @@ class ForwardRuleReasonerRDFS(sc: SparkContext) extends ForwardRuleReasoner{
     val domainMapBC = sc.broadcast(domainMap)
 
     val triplesRDFS2 =
-      triplesRDD
+      otherTriples
         .filter(t => domainMapBC.value.contains(t.predicate))
         .map(t => RDFTriple(t.subject, RDF.`type`.getURI, domainMapBC.value(t.predicate)))
         .setName("rdfs2")
@@ -94,15 +101,11 @@ class ForwardRuleReasonerRDFS(sc: SparkContext) extends ForwardRuleReasoner{
     val rangeMapBC = sc.broadcast(rangeMap)
 
     val triplesRDFS3 =
-      triplesRDD
+      otherTriples
         .filter(t => rangeMapBC.value.contains(t.predicate))
         .map(t => RDFTriple(t.`object`, RDF.`type`.getURI, rangeMapBC.value(t.predicate)))
         .setName("rdfs3")
 
-    // extract the rdf:type triples
-    var typeTriples = triplesRDD
-      .filter(t => t.predicate == RDF.`type`.getURI)
-
     // rdfs2 and rdfs3 generated rdf:type triples which we'll add to the existing ones
     val triples23 = triplesRDFS2.union(triplesRDFS3)
 
@@ -123,6 +126,7 @@ class ForwardRuleReasonerRDFS(sc: SparkContext) extends ForwardRuleReasoner{
 
     // 5. merge triples and remove duplicates
     val allTriples = sc.union(Seq(
+                          otherTriples,
                           subClassOfTriplesTrans,
                           subPropertyOfTriplesTrans,
                           typeTriples,
@@ -131,8 +135,8 @@ class ForwardRuleReasonerRDFS(sc: SparkContext) extends ForwardRuleReasoner{
                         .distinct()
 
     logger.info("...finished materialization in " + (System.currentTimeMillis() - startTime) + "ms.")
-    val newSize = allTriples.count()
-    logger.info(s"|G_inf|=$newSize")
+//    val newSize = allTriples.count()
+//    logger.info(s"|G_inf|=$newSize")
 
     // return graph with inferred triples
     new RDFGraph(allTriples)