N-Triple writer extended.

Author: LorenzBuehmann

sansa-inference-common/src/main/scala/net/sansa_stack/inference/utils/RDFTripleToNTripleString.scala

@@ -0,0 +1,15 @@
+package net.sansa_stack.inference.utils
+
+import net.sansa_stack.inference.data.RDFTriple
+
+/**
+  * Convert an RDFTriple object to an N-Triple string.
+  *
+  * @author Lorenz Buehmann
+  */
+class RDFTripleToNTripleString extends Function1[RDFTriple, String] with java.io.Serializable {
+  override def apply(t: RDFTriple): String = {
+    val objStr = if(t.`object`.startsWith("http:") || t.`object`.startsWith("ftp:")) s"<${t.`object`}>" else t.`object`
+    s"<${t.subject}> <${t.predicate}> ${objStr} ."
+  }
+}

sansa-inference-flink/src/main/scala/net/sansa_stack/inference/flink/data/RDFGraphWriter.scala

@@ -7,7 +7,7 @@ import org.apache.flink.api.common.operators.Order
 import org.apache.flink.api.scala._
 import org.apache.flink.core.fs.FileSystem
 import org.apache.jena.rdf.model.{Model, ModelFactory}
-import net.sansa_stack.inference.utils.RDFTripleOrdering
+import net.sansa_stack.inference.utils.{RDFTripleOrdering, RDFTripleToNTripleString}
 import org.slf4j.LoggerFactory
 
 /**
@@ -27,7 +27,7 @@ object RDFGraphWriter {
     implicit val ordering = RDFTripleOrdering
 
     graph.triples.map(t=>(t,t)).sortPartition(1, Order.DESCENDING).map(_._1)
-      .map(t => "<" + t.subject + "> <" + t.predicate + "> <" + t.`object` + "> .") // to N-TRIPLES string
+      .map(new RDFTripleToNTripleString()) // to N-TRIPLES string
       .writeAsText(path, writeMode = FileSystem.WriteMode.OVERWRITE)
 
     logger.info("finished writing triples to disk in " + (System.currentTimeMillis()-startTime) + "ms.")
@@ -48,15 +48,14 @@ object RDFGraphWriter {
     implicit val ordering = RDFTripleOrdering
 
     graph.triples.map(t=>(t,t)).sortPartition(1, Order.DESCENDING).map(_._1)
-      .map(t => "<" + t.subject + "> <" + t.predicate + "> <" + t.`object` + "> .") // to N-TRIPLES string
+      .map(new RDFTripleToNTripleString()) // to N-TRIPLES string
       .writeAsText(path.getAbsolutePath, writeMode = FileSystem.WriteMode.OVERWRITE)
 
     logger.info("finished writing triples to disk in " + (System.currentTimeMillis()-startTime) + "ms.")
   }
 
   def convertToModel(graph: RDFGraph) : Model = {
-    val modelString = graph.triples.map(t =>
-      "<" + t.subject + "> <" + t.predicate + "> <" + t.`object` + "> .") // to N-TRIPLES string
+    val modelString = graph.triples.map(new RDFTripleToNTripleString())
       .collect().mkString("\n")
 
     val model = ModelFactory.createDefaultModel()

sansa-inference-spark/src/main/scala/net/sansa_stack/inference/spark/data/RDFGraphWriter.scala

@@ -7,7 +7,7 @@ import org.apache.jena.rdf.model.{Model, ModelFactory}
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.DataFrame
 import net.sansa_stack.inference.data.RDFTriple
-import net.sansa_stack.inference.utils.RDFTripleOrdering
+import net.sansa_stack.inference.utils.{RDFTripleOrdering, RDFTripleToNTripleString}
 import org.slf4j.LoggerFactory
 
 /**
@@ -54,7 +54,7 @@ object RDFGraphWriter {
               }
 
     // convert to N-Triple format
-    var triplesNTFormat = tmp.map(t => "<" + t.subject + "> <" + t.predicate + "> <" + t.`object` + "> .")
+    var triplesNTFormat = tmp.map(new RDFTripleToNTripleString())
 
     // convert to single file, i.e. move al lto one partition
     // (might be very expensive and contradicts the Big Data paradigm on Hadoop in general)
@@ -73,8 +73,7 @@ object RDFGraphWriter {
   }
 
   def convertToModel(graph: RDFGraph) : Model = {
-    val modelString = graph.triples.map(t =>
-      "<" + t.subject + "> <" + t.predicate + "> <" + t.`object` + "> .") // to N-TRIPLES string
+    val modelString = graph.triples.map(new RDFTripleToNTripleString())
       .collect().mkString("\n")
 
     val model = ModelFactory.createDefaultModel()