Load RDF graph into DataFrame via separate Reader implementation.

Author: LorenzBuehmann

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

@@ -5,10 +5,13 @@ import java.net.URI
 import scala.language.implicitConversions
 
 import org.apache.spark.SparkContext
-import org.apache.spark.sql.{Dataset, SparkSession}
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.sources.{BaseRelation, RelationProvider, SchemaRelationProvider, TableScan}
+import org.apache.spark.sql.types.{StringType, StructField, StructType}
+import org.apache.spark.sql.{Dataset, Row, SQLContext, SparkSession}
 import org.slf4j.LoggerFactory
 
-import net.sansa_stack.inference.data.RDFTriple
+import net.sansa_stack.inference.data.{RDFTriple, SQLSchema, SQLSchemaDefault}
 import net.sansa_stack.inference.spark.data.model.{RDFGraph, RDFGraphDataFrame, RDFGraphDataset, RDFGraphNative}
 import net.sansa_stack.inference.utils.NTriplesStringToRDFTriple
 
@@ -192,7 +195,15 @@ object RDFGraphLoader {
     * @param minPartitions min number of partitions for Hadoop RDDs ([[SparkContext.defaultMinPartitions]])
     * @return an RDF graph based on a [[org.apache.spark.sql.DataFrame]]
     */
-  def loadFromDiskAsDataFrame(session: SparkSession, path: String, minPartitions: Int): RDFGraphDataFrame = {
-    new RDFGraphDataFrame(loadFromDiskAsRDD(session, path, minPartitions).toDataFrame(session))
+  def loadFromDiskAsDataFrame(session: SparkSession, path: String, minPartitions: Int, sqlSchema: SQLSchema = SQLSchemaDefault): RDFGraphDataFrame = {
+    val df = session
+      .read
+      .format("net.sansa_stack.inference.spark.data.loader.sql")
+      .load(path)
+
+    // register the DataFrame as a table
+    df.createOrReplaceTempView(sqlSchema.triplesTable)
+
+    new RDFGraphDataFrame(df)
   }
 }

sansa-inference-spark/src/main/scala/net/sansa_stack/inference/spark/data/loader/sql/DefaultSource.scala

@@ -0,0 +1,19 @@
+package net.sansa_stack.inference.spark.data.loader.sql
+
+import org.apache.spark.sql.SQLContext
+import org.apache.spark.sql.sources.{BaseRelation, RelationProvider, SchemaRelationProvider}
+import org.apache.spark.sql.types.StructType
+
+
+class DefaultSource extends RelationProvider with SchemaRelationProvider {
+    override def createRelation(sqlContext: SQLContext, parameters: Map[String, String])
+    : BaseRelation = {
+      createRelation(sqlContext, parameters, null)
+    }
+    override def createRelation(sqlContext: SQLContext, parameters: Map[String, String]
+                                , schema: StructType)
+    : BaseRelation = {
+      parameters.getOrElse("path", sys.error("'path' must be specified for our data."))
+      return new NTriplesRelation(parameters.get("path").get, schema)(sqlContext)
+    }
+  }

sansa-inference-spark/src/main/scala/net/sansa_stack/inference/spark/data/loader/sql/NTriplesRelation.scala

@@ -0,0 +1,37 @@
+package net.sansa_stack.inference.spark.data.loader.sql
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.{Row, SQLContext}
+import org.apache.spark.sql.sources.{BaseRelation, TableScan}
+import org.apache.spark.sql.types.{StringType, StructField, StructType}
+
+import net.sansa_stack.inference.utils.NTriplesStringToRDFTriple
+
+class NTriplesRelation(location: String, userSchema: StructType)
+                      (@transient val sqlContext: SQLContext)
+    extends BaseRelation
+      with TableScan
+      with Serializable {
+    override def schema: StructType = {
+      if (this.userSchema != null) {
+        this.userSchema
+      }
+      else {
+        StructType(
+          Seq(
+            StructField("s", StringType, true),
+            StructField("p", StringType, true),
+            StructField("o", StringType, true)
+        ))
+      }
+    }
+    override def buildScan(): RDD[Row] = {
+      val rdd = sqlContext
+        .sparkContext
+        .textFile(location)
+
+      val rows = rdd.map(new NTriplesStringToRDFTriple()).map(t => Row.fromSeq(Seq(t.s, t.p, t.o)))
+
+      rows
+    }
+  }