Merge pull request #66 from SANSA-Stack/bugfix/spark/rdfxml-dataframe

Fixes #33
SANSA-Stack · Jul 18, 2018 · 413fe98 · 413fe98
2 parents 4b1e3ab + b53d107
commit 413fe98
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 9 deletions.
diff --git a/sansa-rdf-spark/src/main/scala/net/sansa_stack/rdf/spark/io/rdfxml/JenaParser.scala b/sansa-rdf-spark/src/main/scala/net/sansa_stack/rdf/spark/io/rdfxml/JenaParser.scala
@@ -17,10 +17,9 @@ class JenaParser(val options: RdfXmlOptions) {
                recordLiteral: T => UTF8String)
   : Seq[InternalRow] = {
     try {
-      println(recordLiteral(record))
       val triples = new CollectorStreamRDF()
       createParser(parserBuilder, record).parse(triples)
-      triples.getTriples.asScala.map(convertTriple(_))
+      triples.getTriples.asScala.map(convertTriple)
     } catch {
       case e@(_: RuntimeException | _: RiotException) =>
         throw BadRecordException(() => recordLiteral(record), () => None, e)
@@ -30,7 +29,9 @@ class JenaParser(val options: RdfXmlOptions) {
 
   def convertTriple(triple: org.apache.jena.graph.Triple): InternalRow = {
     val row = new GenericInternalRow(3)
-    row.update(0, triple.getSubject)
+    row.update(0, UTF8String.fromBytes(triple.getSubject.toString.getBytes()))
+    row.update(1, UTF8String.fromBytes(triple.getPredicate.toString.getBytes()))
+    row.update(2, UTF8String.fromBytes(triple.getObject.toString.getBytes()))
     row
   }
 }
diff --git a/sansa-rdf-spark/src/main/scala/net/sansa_stack/rdf/spark/io/rdfxml/RdfXmlDataSource.scala b/sansa-rdf-spark/src/main/scala/net/sansa_stack/rdf/spark/io/rdfxml/RdfXmlDataSource.scala
@@ -60,7 +60,8 @@ object RdfXmlDataSource {
     if (options.wholeFile) {
       WholeFileRdfXmlDataSource
     } else {
-      TextInputRdfXmlDataSource
+      WholeFileRdfXmlDataSource
+//      TextInputRdfXmlDataSource
     }
   }
 
@@ -87,7 +88,7 @@ object RdfXmlDataSource {
 object TextInputRdfXmlDataSource extends RdfXmlDataSource[Text] {
   override val isSplitable: Boolean = {
     // splittable if the underlying source is
-    true
+    false
   }
 
   override protected def createBaseRdd(

diff --git a/sansa-rdf-spark/src/main/scala/net/sansa_stack/rdf/spark/io/rdfxml/RdfXmlFileFormat.scala b/sansa-rdf-spark/src/main/scala/net/sansa_stack/rdf/spark/io/rdfxml/RdfXmlFileFormat.scala
@@ -14,15 +14,17 @@ import org.apache.spark.sql.types.{StringType, StructField, StructType}
 class RdfXmlFileFormat extends TextBasedFileFormat with DataSourceRegister {
   override val shortName: String = "rdfxml"
 
-  override def isSplitable(
-      sparkSession: SparkSession,
-      options: Map[String, String],
-      path: Path): Boolean = {
+  override def isSplitable(sparkSession: SparkSession,
+                           options: Map[String, String],
+                           path: Path): Boolean = {
+
     val parsedOptions = new RdfXmlOptions(
       options,
       sparkSession.sessionState.conf.sessionLocalTimeZone,
       sparkSession.sessionState.conf.columnNameOfCorruptRecord)
+
     val rdfDataSource = RdfXmlDataSource(parsedOptions)
+
     rdfDataSource.isSplitable && super.isSplitable(sparkSession, options, path)
   }