Removed redundant RDF loading code

This has already been moved to RDF layer and can cause conflicts if both
RDF and Inference dependency are used.

Closes #25

Author: LorenzBuehmann

pom.xml

@@ -69,9 +69,9 @@
     <scala.version>2.11.11</scala.version>
     <scala.binary.version>2.11</scala.binary.version>
     <spark.version>2.2.1</spark.version>
-    <flink.version>1.3.2</flink.version>
+    <flink.version>1.4.0</flink.version>
     <jena.version>3.5.0</jena.version>
-    <sansa.stack.version>0.3.0</sansa.stack.version>
+    <sansa.stack.version>0.3.1-SNAPSHOT</sansa.stack.version>
     <sansa.rdf.version>${sansa.stack.version}</sansa.rdf.version>
     <sansa.query.version>${sansa.stack.version}</sansa.query.version>
     <sansa.owl.version>${sansa.stack.version}</sansa.owl.version>
@@ -93,7 +93,7 @@
       <!-- RDF Layer -->
       <dependency>
         <groupId>${project.groupId}</groupId>
-        <artifactId>sansa-rdf-spark-core</artifactId>
+        <artifactId>sansa-rdf-spark_${scala.binary.version}</artifactId>
         <version>${sansa.rdf.version}</version>
       </dependency>
       <dependency>

sansa-inference-common/pom.xml

@@ -65,6 +65,7 @@
           <artifactId>google-collections</artifactId>
         </exclusion>
       </exclusions>
+      <scope>provided</scope>
     </dependency>
 
     <!-- Guava -->

sansa-inference-spark/pom.xml

@@ -33,10 +33,20 @@
         </dependency>
 
         <!-- RDF Layer -->
-        <!--<dependency>-->
-            <!--<groupId>${project.groupId}</groupId>-->
-            <!--<artifactId>sansa-rdf-spark-core</artifactId>-->
-        <!--</dependency>-->
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>sansa-rdf-spark_${scala.binary.version}</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.apache.hadoop</groupId>
+                    <artifactId>hadoop-common</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>org.apache.hadoop</groupId>
+                    <artifactId>hadoop-mapreduce-client-core</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
         <!--<dependency>-->
             <!--<groupId>net.sansa-stack</groupId>-->
             <!--<artifactId>sansa-rdf-partition-core</artifactId>-->
@@ -199,7 +209,7 @@
         <dependency>
             <groupId>org.apache.calcite</groupId>
             <artifactId>calcite-core</artifactId>
-            <scope>provided</scope>
+            <!--<scope>provided</scope>-->
         </dependency>
 
     </dependencies>

sansa-inference-spark/src/main/resources/META-INF/services/org.apache.spark.sql.sources.DataSourceRegister

@@ -7,11 +7,11 @@ import net.sansa_stack.inference.spark.data.model.{RDFGraph, RDFGraphDataFrame,
 import net.sansa_stack.inference.utils.NTriplesStringToJenaTriple
 import org.apache.jena.graph.Triple
 import org.apache.jena.riot.Lang
-import org.apache.spark.sql.{Dataset, SaveMode, SparkSession}
+import org.apache.spark.sql.{Dataset, Encoder, SaveMode, SparkSession}
 import org.apache.spark.{SparkConf, SparkContext}
 import org.slf4j.LoggerFactory
-import scala.language.implicitConversions
 
+import scala.language.implicitConversions
 import org.apache.jena.vocabulary.RDF
 
 /**
@@ -42,6 +42,7 @@ object RDFGraphLoader {
 
     val triples = session.sparkContext
       .textFile(path, minPartitions) // read the text file
+      .filter(line => !line.trim().isEmpty & !line.startsWith("#"))
       .map(new NTriplesStringToJenaTriple()) // convert to triple object
 //      .repartition(minPartitions)
 
@@ -127,7 +128,7 @@ object RDFGraphLoader {
       Array(splitted(0), splitted(1), splitted(2))
     })
 
-    implicit val rdfTripleEncoder = org.apache.spark.sql.Encoders.kryo[Triple]
+    implicit val rdfTripleEncoder: Encoder[Triple] = org.apache.spark.sql.Encoders.kryo[Triple]
     val spark = session.sqlContext
 
 
@@ -195,7 +196,7 @@ 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, sqlSchema: SQLSchema = SQLSchemaDefault): RDFGraphDataFrame = {
+  def loadFromDiskAsDataFrame(session: SparkSession, path: String, minPartitions: Int = 4, sqlSchema: SQLSchema = SQLSchemaDefault): RDFGraphDataFrame = {
     val df = session
       .read
       .format("net.sansa_stack.inference.spark.data.loader.sql")
@@ -208,7 +209,7 @@ object RDFGraphLoader {
   }
 
   def main(args: Array[String]): Unit = {
-    import net.sansa_stack.inference.spark.data.loader.sql.rdf._
+    import net.sansa_stack.rdf.spark.io.rdf._
 
     val path = args(0)
     val lang = args(1) match {
@@ -247,8 +248,6 @@ object RDFGraphLoader {
 
 
 
-    import net.sansa_stack.inference.spark.data.loader.rdd.rdf._
-
     val triplesRDD = session.sparkContext.rdf(lang)(path)
     triples.show(10)
     println(triples.count())