More robust N-Triples parser

Author: LorenzBuehmann

sansa-inference-common/src/main/scala/net/sansa_stack/inference/utils/NTriplesStringToRDFTriple.scala

@@ -1,5 +1,9 @@
 package net.sansa_stack.inference.utils
 
+import java.util.regex.Pattern
+
+import scala.util.Try
+
 import net.sansa_stack.inference.data.RDFTriple
 
 /**
@@ -8,10 +12,76 @@ import net.sansa_stack.inference.data.RDFTriple
   * @author Lorenz Buehmann
   */
 class NTriplesStringToRDFTriple
-    extends Function1[String, RDFTriple]
-    with java.io.Serializable {
-  override def apply(s: String): RDFTriple = {
-    val tokens = s.replace("<", "").replace(">", "").split(" ") // split by white space
-    RDFTriple(tokens(0), tokens(1), tokens(2))
+    extends Function1[String, Option[RDFTriple]]
+    with java.io.Serializable
+    with Logging {
+
+//  val pattern: Pattern = Pattern.compile("^<([^>]+)>|(?<!<)([^>]+)(?<!>)\\s*<([^>]+)>\\s*<?([^>]+)>?\\s*\\.$")
+
+  val pattern: Pattern = Pattern.compile(
+    """|^
+       |(<([^>]*)>|(?<!<)([^>]+)(?<!>))
+       |\s*
+       |<([^>]+)>
+       |\s*
+       |(<([^>]+)>|(.*))
+       |\s*\.$
+    """.stripMargin.replaceAll("\n", "").trim)
+
+  override def apply(s: String): Option[RDFTriple] = {
+   parseRegexPattern(s)
+//    parseRegexSplit(s)
   }
+
+  def parseRegexPattern(s : String): Option[RDFTriple] = Try {
+    val matcher = pattern.matcher(s)
+
+    if (matcher.matches) {
+//      for(i <- 0 to matcher.groupCount())
+//        println(i + ":" + matcher.group(i))
+
+      val subject = if (matcher.group(2) == null) {
+        matcher.group(1)
+      } else {
+        matcher.group(2)
+      }
+
+      val obj = if (matcher.group(6) == null) {
+        matcher.group(7).trim
+      } else {
+        matcher.group(6)
+      }
+
+      RDFTriple(subject, matcher.group(4), obj)
+    } else {
+      warn(s"WARN: Illegal N-Triples syntax. Ignoring triple $s")
+      throw new Exception(s"WARN: Illegal N-Triples syntax. Ignoring triple $s")
+    }
+
+  }.toOption
+
+  def parseRegexSplit(s: String): Option[RDFTriple] = Try {
+    val s1 = s.trim
+    val split = s1.substring(0, s1.lastIndexOf('.')).split("\\s", 3)
+    var obj = split(2).trim
+    obj = obj.substring(0, obj.lastIndexOf('.'))
+    RDFTriple(split(0), split(1), obj)
+  }.toOption
+
 }
+
+object NTriplesStringToRDFTriple {
+  def main(args: Array[String]): Unit = {
+    val s1 = "<> <http://www.w3.org/1999/02/22-rdf-syntax-ns#rest> <http://www.w3.org/1999/02/22-rdf-syntax-ns#nil> ."
+    val s2 = "_:genid32 <http://www.w3.org/1999/02/22-rdf-syntax-ns#rest> <http://www.w3.org/1999/02/22-rdf-syntax-ns#nil> ."
+    val s3 = "<http://example.org/a> <http://www.w3.org/1999/02/22-rdf-syntax-ns#rest> <http://www.w3.org/1999/02/22-rdf-syntax-ns#nil> ."
+    val s4 = "<http://example.org/a> <http://www.w3.org/1999/02/22-rdf-syntax-ns#rest> \"3\"^^<http://ex.org/int> ."
+
+    println(new NTriplesStringToRDFTriple().apply(s1))
+    println(new NTriplesStringToRDFTriple().apply(s2))
+    println(new NTriplesStringToRDFTriple().apply(s3))
+    println(new NTriplesStringToRDFTriple().apply(s4))
+  }
+}
+
+

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

@@ -46,7 +46,9 @@ object RDFGraphLoader {
 
     val tmp: List[String] = paths.map(path => path.toString).toList
 
-    val triples = tmp.map(f => env.readTextFile(f).map(new NTriplesStringToRDFTriple())).reduce(_ union _).name("triples")
+    val converter = new NTriplesStringToRDFTriple()
+
+    val triples = tmp.map(f => env.readTextFile(f).flatMap(line => converter.apply(line))).reduce(_ union _).name("triples")
 
     RDFGraph(triples)
   }

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

@@ -5,10 +5,7 @@ import java.net.URI
 import scala.language.implicitConversions
 
 import org.apache.spark.SparkContext
-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.apache.spark.sql.{Dataset, SparkSession}
 import org.slf4j.LoggerFactory
 
 import net.sansa_stack.inference.data.{RDFTriple, SQLSchema, SQLSchemaDefault}
@@ -41,9 +38,11 @@ object RDFGraphLoader {
     logger.info("loading triples from disk...")
     val startTime = System.currentTimeMillis()
 
+    val converter = new NTriplesStringToRDFTriple()
+
     val triples = session.sparkContext
       .textFile(path, minPartitions) // read the text file
-      .map(new NTriplesStringToRDFTriple()) // convert to triple object
+      .flatMap(line => converter.apply(line)) // convert to triple object
 //      .repartition(minPartitions)
 
 //  logger.info("finished loading " + triples.count() + " triples in " + (System.currentTimeMillis()-startTime) + "ms.")
@@ -88,9 +87,11 @@ object RDFGraphLoader {
     logger.info("loading triples from disk...")
     val startTime = System.currentTimeMillis()
 
+    val converter = new NTriplesStringToRDFTriple()
+
     val triples = session.sparkContext
       .textFile(path, minPartitions) // read the text file
-      .map(new NTriplesStringToRDFTriple()) // convert to triple object
+      .flatMap(line => converter.apply(line)) // convert to triple object
 
     // logger.info("finished loading " + triples.count() + " triples in " +
     // (System.currentTimeMillis()-startTime) + "ms.")

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

@@ -30,7 +30,9 @@ class NTriplesRelation(location: String, userSchema: StructType)
         .sparkContext
         .textFile(location)
 
-      val rows = rdd.map(new NTriplesStringToRDFTriple()).map(t => Row.fromSeq(Seq(t.s, t.p, t.o)))
+      val converter = new NTriplesStringToRDFTriple()
+
+      val rows = rdd.flatMap(x => converter.apply(x)).map(t => Row.fromSeq(Seq(t.s, t.p, t.o)))
 
       rows
     }