Read directory.

Write to disk.

Author: LorenzBuehmann

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

@@ -1,7 +1,10 @@
 package net.sansa_stack.inference.flink.data
 
+import java.io.File
+
 import org.apache.flink.api.scala.{ExecutionEnvironment, _}
 import net.sansa_stack.inference.data.RDFTriple
+import org.apache.flink.configuration.Configuration
 
 /**
   * @author Lorenz Buehmann
@@ -16,4 +19,34 @@ object RDFGraphLoader {
     RDFGraph(triples)
   }
 
+  def loadFromDisk(path: File, env: ExecutionEnvironment): RDFGraph = {
+    val paths = if(path.isFile) Seq(path.getAbsolutePath) else path.listFiles().map(f => f.getAbsolutePath).toSeq
+
+    // create a configuration object
+    val parameters = new Configuration
+
+    // set the recursive enumeration parameter
+    parameters.setBoolean("recursive.file.enumeration", true)
+
+    // pass the configuration to the data source
+    val triples = env.readTextFile(path.getAbsolutePath).withParameters(parameters)
+      .map(line => line.replace(">", "").replace("<", "").split("\\s+")) // line to tokens
+      .map(tokens => RDFTriple(tokens(0), tokens(1), tokens(2))) // tokens to triple
+
+    RDFGraph(triples)
+  }
+
+  def loadFromDisk(paths: Seq[File], env: ExecutionEnvironment): RDFGraph = {
+
+    val tmp: Seq[String] = paths.map(path => {
+      if(path.isFile) Seq(path.getAbsolutePath) else path.listFiles().map(f => f.getAbsolutePath).toSeq
+    }).flatMap(identity)
+
+    val triples = env.fromCollection(tmp)
+      .map(line => line.replace(">", "").replace("<", "").split("\\s+")) // line to tokens
+      .map(tokens => RDFTriple(tokens(0), tokens(1), tokens(2))) // tokens to triple
+
+    RDFGraph(triples)
+  }
+
 }

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

@@ -1,6 +1,6 @@
 package net.sansa_stack.inference.flink.data
 
-import java.io.ByteArrayInputStream
+import java.io.{ByteArrayInputStream, File}
 import java.nio.charset.StandardCharsets
 
 import org.apache.flink.api.common.operators.Order
@@ -33,6 +33,19 @@ object RDFGraphWriter {
     logger.info("finished writing triples to disk in " + (System.currentTimeMillis()-startTime) + "ms.")
   }
 
+  def writeToDisk(graph: RDFGraph, path: File): Unit = {
+    logger.info("writing triples to disk...")
+    val startTime  = System.currentTimeMillis()
+
+    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
+      .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