Added: Separate implementation of schema extractors

Author: LorenzBuehmann

sansa-inference-flink/src/main/scala/net/sansa_stack/inference/flink/extraction/OWLHorstSchemaExtractor.scala

@@ -0,0 +1,53 @@
+package net.sansa_stack.inference.flink.extraction
+
+import org.apache.jena.vocabulary.{OWL2, RDFS}
+
+/**
+  * An extractor of the schema for OWL Horst.
+  *
+  * Currently, it's supports the extraction of triples `(s,p,o)` with `p` being
+  *
+  *  - rdfs:subClassOf
+  *  - rdfs:subPropertyOf
+  *  - rdfs:domain
+  *  - rdfs:range
+  *  - owl:equivalentProperty
+  *  - owl:equivalentClass
+  *  - owl:inverseOf
+  *  - owl:someValuesFrom
+  *  - owl:allValuesFrom
+  *  - owl:hasValue
+  *  - owl:onProperty
+  *
+  *  or `o` being
+  *
+  *  - owl:TransitiveProperty
+  *  - owl:FunctionalProperty
+  *  - owl:InverseFunctionalProperty
+  *  - owl:SymmetricProperty
+  *
+  * @author Lorenz Buehmann
+  */
+class OWLHorstSchemaExtractor()
+    extends SchemaExtractor()(
+      Set(
+        RDFS.subClassOf,
+        RDFS.subPropertyOf,
+        RDFS.domain,
+        RDFS.range,
+        OWL2.equivalentProperty,
+        OWL2.equivalentClass,
+        OWL2.inverseOf,
+        OWL2.someValuesFrom,
+        OWL2.allValuesFrom,
+        OWL2.hasValue,
+        OWL2.onProperty
+      ).map(p => p.getURI)
+    )(
+      Set(
+        OWL2.TransitiveProperty,
+        OWL2.FunctionalProperty,
+        OWL2.InverseFunctionalProperty,
+        OWL2.SymmetricProperty
+      ).map(p => p.getURI)
+    ) {}

sansa-inference-flink/src/main/scala/net/sansa_stack/inference/flink/extraction/RDFSSchemaExtractor.scala

@@ -0,0 +1,22 @@
+package net.sansa_stack.inference.flink.extraction
+
+import org.apache.jena.vocabulary.RDFS
+
+/**
+  * An extractor of the schema for RDFS.
+  *
+  * Currently, it's supports the extraction of triples `(s,p,o)` with `p` being
+  *
+  *  - rdfs:subClassOf
+  *  - rdfs:subPropertyOf
+  *  - rdfs:domain
+  *  - rdfs:range
+  *
+  * @author Lorenz Buehmann
+  */
+class RDFSSchemaExtractor()
+  extends SchemaExtractor()(Set(RDFS.subClassOf, RDFS.subPropertyOf, RDFS.domain, RDFS.range).map(p => p.getURI))() {}
+
+object RDFSSchemaExtractor {
+  def apply: RDFSSchemaExtractor = new RDFSSchemaExtractor()
+}

sansa-inference-flink/src/main/scala/net/sansa_stack/inference/flink/extraction/SchemaExtractor.scala

@@ -0,0 +1,45 @@
+package net.sansa_stack.inference.flink.extraction
+
+import org.apache.flink.api.scala.DataSet
+import org.apache.jena.vocabulary.RDFS
+
+import net.sansa_stack.inference.data.RDFTriple
+import net.sansa_stack.inference.flink.data.RDFGraph
+import net.sansa_stack.inference.utils.Logging
+
+/**
+  * @author Lorenz Buehmann
+  */
+abstract class SchemaExtractor
+  (subjects: Set[String] = Set())
+  (predicates: Set[String] = Set())
+  (objects: Set[String] = Set())
+    extends Logging with Serializable{
+
+  val subjectsFilter: ((RDFTriple) => Boolean) = t => subjects.contains(t.s)
+  val predicatesFilter: ((RDFTriple) => Boolean) = t => predicates.contains(t.p)
+  val objectsFilter: ((RDFTriple) => Boolean) = t => objects.contains(t.o)
+
+  private def or(ps: (RDFTriple => Boolean)*) = (a: RDFTriple) => ps.exists(_(a))
+
+  /**
+    * Extract a graph that contains only the schema triples.
+    *
+    * @param graph the graph
+    * @return a graph containing only the schema triples
+    */
+  def extract(graph: RDFGraph): RDFGraph =
+    new RDFGraph(extract(graph.triples))
+
+  /**
+    * Extract a DataSet that contains only the schema triples.
+    *
+    * @param triples the triples
+    * @return the schema triples
+    */
+  def extract(triples: DataSet[RDFTriple]): DataSet[RDFTriple] =
+    triples
+      .filter(or(subjectsFilter, predicatesFilter, objectsFilter))
+      .name("schema-triples")
+
+}