finished mappings (probably)

fixed some roundtriptests

Author: eisenbahnplatte

README.md

@@ -10,7 +10,7 @@ Example Application Deployment: Download the files of 5 datasets as given in the
 ## Current State
 
 **beta**: 
-most of the times it should produce expected results for compression and RDF conversion. Please expect some code refactoring and fluctuation. There will be an open-source licence, either GPL or Apache.  
+most of the times it should produce expected results for compression and RDF format.conversion. Please expect some code refactoring and fluctuation. There will be an open-source licence, either GPL or Apache.  
 
 
 ## Concept
@@ -28,7 +28,7 @@ The databus-client is designed to unify and convert data on the client-side in s
 * Level 1: all features finished, testing required
 * Level 2: using Apache Compress library covers most of the compression formats, more testing required
 * Level 3: Scalable RDF libraries from [SANSA-Stack](http://sansa-stack.net/) and [Databus Derive](https://github.com/dbpedia/databus-derive). Step by step, extension for all (quasi-)isomorphic [IANA mediatypes](https://www.iana.org/assignments/media-types/media-types.xhtml).
-* Level 4: In addition, we plan to provide a plugin mechanism to incorporate more sophisticated mapping engines as [Tarql](https://tarql.github.io/) (already implemented), [RML](http://rml.io), R2RML, [R2R](http://wifo5-03.informatik.uni-mannheim.de/bizer/r2r/) (for owl:equivalence translation) and XSLT. 
+* Level 4: In addition, we plan to provide a plugin mechanism to incorporate more sophisticated format.mapping engines as [Tarql](https://tarql.github.io/) (already implemented), [RML](http://rml.io), R2RML, [R2R](http://wifo5-03.informatik.uni-mannheim.de/bizer/r2r/) (for owl:equivalence translation) and XSLT. 
 
 
 ## Usage   

errorLog.log

@@ -201,7 +201,7 @@
                 <artifactId>scalatest-maven-plugin</artifactId>
                 <version>1.0</version>
                 <configuration>
-                    <suites>conversionTests.mapping.roundTripTests, conversionTests.conversion.roundTripTests</suites>
+                    <suites>archived.format.mapping.roundTripTests, format.conversion.format.conversion.roundTripTests</suites>
                     <reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory>
                     <junitxml>.</junitxml>
                     <filereports>WDF TestSuite.txt</filereports>

pom.xml

@@ -191,7 +191,7 @@ object FileUtil {
    * @return format
    */
   def getFormatType(inputFile: File, compressionInputFile: String): String = {
-    {
+    val format ={
       try {
         if (!(getFormatTypeWithDataID(inputFile) == "")) {
           getFormatTypeWithDataID(inputFile)
@@ -202,6 +202,23 @@ object FileUtil {
         case _: FileNotFoundException => getFormatTypeWithoutDataID(inputFile, compressionInputFile)
       }
     }
+
+    if (format == "rdf") "rdfxml"
+    else format
+  }
+
+  /**
+   * read a query file as string
+   *
+   * @param file query file
+   * @return query string
+   */
+  def readQueryFile(file: File): String = {
+    var queryString: String = ""
+    for (line <- file.lineIterator) {
+      queryString = queryString.concat(line).concat("\n")
+    }
+    queryString
   }
 
   /**

src/main/scala/org/dbpedia/databus/client/filehandling/FileUtil.scala

@@ -26,7 +26,7 @@ class SourceHandler(conf:CLI_Config) {
       val sourceFile: File = File(conf.source())
 
       if (sourceFile.hasExtension && sourceFile.extension.get.matches(".sparql|.query")) { // conf.source() is a query file
-        val queryString = readQueryFile(sourceFile)
+        val queryString = FileUtil.readQueryFile(sourceFile)
         handleQuery(queryString)
       }
       else { // conf.source() is an already existing file or directory
@@ -173,20 +173,6 @@ class SourceHandler(conf:CLI_Config) {
     handler.handleResponse(response)
   }
 
-  /**
-   * read a query file as string
-   *
-   * @param file query file
-   * @return query string
-   */
-  def readQueryFile(file: File): String = {
-    var queryString: String = ""
-    for (line <- file.lineIterator) {
-      queryString = queryString.concat(line).concat("\n")
-    }
-    queryString
-  }
-
   def printTask(sourceType: String, source: String, target: String):Unit = {
     val str =
       s"""

src/main/scala/org/dbpedia/databus/client/filehandling/SourceHandler.scala

@@ -14,6 +14,7 @@ import org.slf4j.LoggerFactory
 
 import scala.util.control.Breaks.{break, breakable}
 import org.apache.jena.graph.Triple
+import org.apache.jena.sparql.core.Quad
 import org.apache.spark.rdd.RDD
 
 import java.net.URLEncoder
@@ -90,7 +91,8 @@ object FormatConverter {
       //read process
       val quads = {
         if (RDF_QUADS.contains(conf.inputFormat))  quadsHandler.read(file.pathAsString, conf.inputFormat)
-        else RDF_Triples_Mapper.map_to_quads(new TripleHandler().read(file.pathAsString, conf.inputFormat), conf.graphURI)
+        else if (RDF_TRIPLES.contains(conf.inputFormat)) RDF_Triples_Mapper.map_to_quads(new TripleHandler().read(file.pathAsString, conf.inputFormat), conf.graphURI)
+        else Spark.context.emptyRDD[Quad]
       }
 
       //write process
@@ -113,19 +115,8 @@ object FormatConverter {
         }
       }
 
+      //write process
       tsdHandler.write(data, conf.outputFormat)
     }
   }
-//  FileUtil.unionFiles(tempDir, targetFile)
-//  if (mappingFile.exists && mappingFile != File("")) {
-//    val mapDir = File("./mappings/")
-//    mapDir.createDirectoryIfNotExists()
-//    mappingFile.moveTo(mapDir / FileUtil.getSha256(targetFile), overwrite = true)
-//  }
-//}
-//catch {
-//  case _: RuntimeException => LoggerFactory.getLogger("UnionFilesLogger").error(s"File $targetFile already exists") //deleteAndRestart(inputFile, inputFormat, outputFormat, targetFile: File)
-//}
-//
-//  targetFile
 }

src/main/scala/org/dbpedia/databus/client/filehandling/convert/FormatConverter.scala

@@ -6,7 +6,7 @@ import org.apache.spark.SparkContext
 import org.apache.spark.rdd.RDD
 import org.dbpedia.databus.client.filehandling.FileUtil
 import org.dbpedia.databus.client.filehandling.convert.format.EquivalenceClassHandler
-import org.dbpedia.databus.client.filehandling.convert.format.rdf.quads.format.{NQuads, Trig, Trix}
+import org.dbpedia.databus.client.filehandling.convert.format.rdf.quads.format.{JsonLD, NQuads, Trig, Trix}
 
 class QuadsHandler extends EquivalenceClassHandler[RDD[Quad]]{
 
@@ -23,6 +23,7 @@ class QuadsHandler extends EquivalenceClassHandler[RDD[Quad]]{
       case "nq" =>    new NQuads().read(source)
       case "trig" =>  new Trix().read(source)
       case "trix" =>  new Trig().read(source)
+      case "jsonld" => new JsonLD().read(source)
     }
   }
 
@@ -38,6 +39,7 @@ class QuadsHandler extends EquivalenceClassHandler[RDD[Quad]]{
       case "nq" =>   new NQuads().write(data)
       case "trig" => new Trig().write(data)
       case "trix" => new Trix().write(data)
+      case "jsonld" => new JsonLD().write(data)
     }
 
   }

src/main/scala/org/dbpedia/databus/client/filehandling/convert/format/rdf/quads/QuadsHandler.scala

@@ -1,23 +1,40 @@
 package org.dbpedia.databus.client.filehandling.convert.format.rdf.quads.format
 
 import better.files.File
-import org.apache.jena.graph.{NodeFactory, Triple}
-import org.apache.jena.rdf.model.{ModelFactory, ResourceFactory}
-import org.apache.jena.riot.{RDFDataMgr, RDFFormat}
+import org.apache.jena.riot.Lang
 import org.apache.jena.sparql.core.Quad
-import org.apache.spark.SparkContext
 import org.apache.spark.rdd.RDD
-import org.apache.spark.sql.SparkSession
 import org.dbpedia.databus.client.filehandling.convert.format.Format
 
-import java.io.{ByteArrayInputStream, ByteArrayOutputStream}
-import scala.io.{Codec, Source}
+class JsonLD extends Format[RDD[Quad]] {
+
+  override def read(source: String): RDD[Quad] = {
+    new Trix(Lang.JSONLD).read(source)
+  }
+
+  override def write(t: RDD[Quad]): File = {
+    new Trix(Lang.JSONLD).write(t)
+  }
+}
+//
+//import better.files.File
+//import org.apache.jena.graph.{NodeFactory, Triple}
+//import org.apache.jena.rdf.model.{ModelFactory, ResourceFactory}
+//import org.apache.jena.riot.{RDFDataMgr, RDFFormat}
+//import org.apache.jena.sparql.core.Quad
+//import org.apache.spark.SparkContext
+//import org.apache.spark.rdd.RDD
+//import org.apache.spark.sql.SparkSession
+//import org.dbpedia.databus.client.filehandling.convert.format.Format
+//
+//import java.io.{ByteArrayInputStream, ByteArrayOutputStream}
+//import scala.io.{Codec, Source}
 
 //class JsonLD extends Format[RDD[Quad]] {
 //
 ////  def readJSONL(spark: SparkSession, inputFile: File): RDD[Triple] = {
-////    val sc = spark.sparkContext
-////    val data = sc.textFile(inputFile.pathAsString)
+//    val sc = spark.sparkContext
+//    val data = sc.textFile(inputFile.pathAsString)
 ////    var tripleRDD = sc.emptyRDD[Triple]
 ////
 ////    //    data.foreach(println(_))

src/main/scala/org/dbpedia/databus/client/filehandling/convert/format/rdf/quads/format/JsonLD.scala

@@ -23,7 +23,7 @@ class TripleHandler extends EquivalenceClassHandler[RDD[Triple]] {
 
     inputFormat match {
       case "nt" => new NTriples().read(source)
-      case "rdf" => new RDFXML().read(source)
+      case "rdfxml" => new RDFXML().read(source)
       case "ttl" =>
         //wie geht das besser?
         try {

src/main/scala/org/dbpedia/databus/client/filehandling/convert/format/rdf/triples/TripleHandler.scala

@@ -3,7 +3,7 @@ package org.dbpedia.databus.client.filehandling.convert.mapping
 import org.apache.spark.rdd.RDD
 import org.apache.jena.graph.Triple
 import org.apache.jena.sparql.core.Quad
-import org.apache.spark.sql.functions.lit
+import org.apache.spark.sql.functions.{col, lit}
 import org.apache.spark.sql.{Column, DataFrame}
 import org.dbpedia.databus.client.filehandling.convert.Spark
 import org.dbpedia.databus.client.filehandling.convert.mapping.util.TriplesResult
@@ -31,18 +31,29 @@ object RDF_Quads_Mapper {
   }
 
   def map_to_tsd(data:RDD[Quad], createMapping:Boolean):DataFrame={
+    //calculate partly results
     val triplesData = map_to_triples(data)
     val dataFrameForEachGraph = triplesData.map(triplesResult => {
       val dataFrame = RDF_Triples_Mapper.map_to_tsd(triplesResult.graph, createMapping)
-      dataFrame.show()
       dataFrame.withColumn("graph", lit(triplesResult.graphName))
     })
 
-    val resultDataFrame = dataFrameForEachGraph.head
+    //join partly results
+    var resultDataFrame = dataFrameForEachGraph.head
 
-    dataFrameForEachGraph.foreach()
-    df1.join(df2, df1.col("column").equalTo(df2("column")))
-    dataFrameForEachGraph.reduce(_ join _)
+    dataFrameForEachGraph.foreach(df => {
+      var columns = Seq.empty[String]
+      resultDataFrame.columns.foreach(col => {
+        if (df.columns.contains(col)) columns = columns:+col
+      })
+      resultDataFrame=resultDataFrame.join(df, columns, "outer")
+    })
+
+    //sort DataFrame
+    val columns = resultDataFrame.columns
+    val graphColIndex = columns.indexOf("graph")
+    val cols = columns.updated(graphColIndex, columns.head).updated(0, "graph").toSeq
+    resultDataFrame.select(cols.map(x=>col(x)):_*).sort("graph")
   }
 }