CLI parser for Flink.

Author: LorenzBuehmann

sansa-inference-flink/pom.xml

@@ -97,6 +97,12 @@ under the License.
 			<scope>test</scope>
 		</dependency>
 
+		<!-- Scopt -->
+		<dependency>
+			<groupId>com.github.scopt</groupId>
+			<artifactId>scopt_${scala.binary.version}</artifactId>
+		</dependency>
+
 	</dependencies>
 
 

sansa-inference-flink/src/main/scala/net/sansa_stack/inference/flink/RDFGraphMaterializer.scala

@@ -1,7 +1,11 @@
 package net.sansa_stack.inference.flink
 
+import java.io.File
+
 import net.sansa_stack.inference.flink.data.{RDFGraphLoader, RDFGraphWriter}
-import net.sansa_stack.inference.flink.forwardchaining.ForwardRuleReasonerRDFS
+import net.sansa_stack.inference.flink.forwardchaining.{ForwardRuleReasonerOWLHorst, ForwardRuleReasonerRDFS}
+import net.sansa_stack.inference.rules.ReasoningProfile
+import net.sansa_stack.inference.rules.ReasoningProfile._
 import org.apache.flink.api.java.utils.ParameterTool
 import org.apache.flink.api.scala.ExecutionEnvironment
 
@@ -13,38 +17,81 @@ import org.apache.flink.api.scala.ExecutionEnvironment
   */
 object RDFGraphMaterializer {
 
-
   def main(args: Array[String]) {
+    parser.parse(args, Config()) match {
+      case Some(config) =>
+        run(args, config.in, config.out, config.profile, config.writeToSingleFile, config.sortedOutput)
+      case None =>
+        println(parser.usage)
+    }
+  }
 
+  def run(args: Array[String], input: Seq[File], output: File, profile: ReasoningProfile, writeToSingleFile: Boolean, sortedOutput: Boolean): Unit = {
     // get params
     val params: ParameterTool = ParameterTool.fromArgs(args)
 
-    if(params.has("input") && params.has("output")) {
+    // set up the execution environment
+    val env = ExecutionEnvironment.getExecutionEnvironment
 
-      // set up the execution environment
-      val env = ExecutionEnvironment.getExecutionEnvironment
+    // make parameters available in the web interface
+    env.getConfig.setGlobalJobParameters(params)
 
-      // make parameters available in the web interface
-      env.getConfig.setGlobalJobParameters(params)
+    // load triples from disk
+    val graph = RDFGraphLoader.loadFromDisk(input(0), env)
+    print(graph.size())
 
-      // load triples from disk
-      val graph = RDFGraphLoader.loadFromFile(params.get("input"), env)
+    // create reasoner
+    val reasoner = profile match {
+      case RDFS => new ForwardRuleReasonerRDFS(env)
+      case OWL_HORST => new ForwardRuleReasonerOWLHorst(env)
+    }
+
+    // compute inferred graph
+    val inferredGraph = reasoner.apply(graph)
+    println(s"|G| = $inferredGraph.size()")
 
-      // create reasoner
-      val reasoner = new ForwardRuleReasonerRDFS(env)
+    // write triples to disk
+    RDFGraphWriter.writeToDisk(inferredGraph, output)
 
-      // compute inferred graph
-      val inferredGraph = reasoner.apply(graph)
+    env.execute(s"RDF ${profile} Reasoning")
 
-      // write triples to disk
-      RDFGraphWriter.writeToFile(inferredGraph, params.get("output"))
+  }
 
-      env.execute("RDF graph materialization")
+  // the config object
+  case class Config(
+                     in: Seq[File] = Seq(),
+                     out: File = new File("."),
+                     profile: ReasoningProfile = ReasoningProfile.RDFS,
+                     writeToSingleFile: Boolean = false,
+                     sortedOutput: Boolean = false)
 
-    } else {
-      System.err.println("Usage: RDFGraphMaterializer --input <sourceFile> --output <targetFile>")
-      System.exit(1)
-    }
+  // read ReasoningProfile enum
+  implicit val profilesRead: scopt.Read[ReasoningProfile.Value] =
+    scopt.Read.reads(ReasoningProfile forName _.toLowerCase())
+
+  // the CLI parser
+  val parser = new scopt.OptionParser[Config]("RDFGraphMaterializer") {
+    head("RDFGraphMaterializer", "0.1.0")
+
+    opt[Seq[File]]('i', "input").required().valueName("<path1>,<path2>,...").
+      action((x, c) => c.copy(in = x)).
+      text("path to file or directory that contains the input files (in N-Triple format)")
+
+    opt[File]('o', "out").required().valueName("<directory>").
+      action((x, c) => c.copy(out = x)).
+      text("the output directory")
+
+    opt[Unit]("single-file").optional().action( (_, c) =>
+      c.copy(writeToSingleFile = true)).text("write the output to a single file in the output directory")
+
+    opt[Unit]("sorted").optional().action( (_, c) =>
+      c.copy(sortedOutput = true)).text("sorted output of the triples (per file)")
+
+    opt[ReasoningProfile]('p', "profile").required().valueName("{rdfs | owl-horst | owl-el | owl-rl}").
+      action((x, c) => c.copy(profile = x)).
+      text("the reasoning profile")
+
+    help("help").text("prints this usage text")
   }
 
 }