<commit_before>package net.sansa_stack.inference.spark

import java.net.URI

import net.sansa_stack.inference.rules.ReasoningProfile._
import net.sansa_stack.inference.rules.{RDFSLevel, ReasoningProfile}
import net.sansa_stack.inference.spark.data.loader.RDFGraphLoader
import net.sansa_stack.inference.spark.data.writer.RDFGraphWriter
import net.sansa_stack.inference.spark.forwardchaining.triples.{ForwardRuleReasonerOWLHorst, ForwardRuleReasonerRDFS, TransitiveReasoner}
import org.apache.jena.graph.{Node, NodeFactory}
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

/**
  * The main entry class to compute the materialization on an RDF graph.
  * Currently, only RDFS and OWL-Horst are supported.
  *
  * @author <NAME>
  *
  */
object RDFGraphMaterializer {

  def main(args: Array[String]) {
    parser.parse(args, Config()) match {
      case Some(config) =>
        run(config.in, config.out, config.profile, config.properties, config.writeToSingleFile, config.sortedOutput, config.parallelism)
      case None =>
        // scalastyle:off println
        println(parser.usage)
        // scalastyle:on println
    }
  }

  def run(input: Seq[URI], output: URI, profile: ReasoningProfile, properties: Seq[Node] = Seq(),
          writeToSingleFile: Boolean, sortedOutput: Boolean, parallelism: Int): Unit = {
    // register the custom classes for Kryo serializer
    val conf = new SparkConf()
    conf.registerKryoClasses(Array(classOf[org.apache.jena.graph.Triple]))
    conf.set("spark.extraListeners", "net.sansa_stack.inference.spark.utils.CustomSparkListener")

    // the SPARK config
    val session = SparkSession.builder
      .appName(s"SPARK $profile Reasoning")
      .master("local[4]")
      .config("spark.eventLog.enabled", "true")
      .config("spark.hadoop.validateOutputSpecs", "false") // override output files
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .config("spark.default.parallelism", parallelism)
      .config("spark.ui.showConsoleProgress", "false")
      .config("spark.sql.shuffle.partitions", parallelism)
      .config(conf)
      .getOrCreate()

//    println(session.conf.getAll.mkString("\n"))

//    val g = RDFGraphLoader.loadFromDiskAsDataset(session, input).distinct()
//    val g_inf = new ForwardRuleReasonerRDFSDataset(session).apply(g)
//    println(s"|G_inf| = ${g_inf.size()}")

    // load triples from disk
    val graph = RDFGraphLoader.loadFromDisk(session, input, parallelism)
//    println(s"|G| = ${graph.size()}")

    // create reasoner
    val reasoner = profile match {
      case TRANSITIVE => new TransitiveReasoner(session.sparkContext, properties, parallelism)
      case RDFS => new ForwardRuleReasonerRDFS(session.sparkContext, parallelism)
      case RDFS_SIMPLE =>
        val r = new ForwardRuleReasonerRDFS(session.sparkContext, parallelism)<commit_msg>Disabled default Spark event logging<commit_after>package net.sansa_stack.inference.spark

import java.net.URI

import net.sansa_stack.inference.rules.ReasoningProfile._
import net.sansa_stack.inference.rules.{RDFSLevel, ReasoningProfile}
import net.sansa_stack.inference.spark.data.loader.RDFGraphLoader
import net.sansa_stack.inference.spark.data.writer.RDFGraphWriter
import net.sansa_stack.inference.spark.forwardchaining.triples.{ForwardRuleReasonerOWLHorst, ForwardRuleReasonerRDFS, TransitiveReasoner}
import org.apache.jena.graph.{Node, NodeFactory}
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

/**
  * The main entry class to compute the materialization on an RDF graph.
  * Currently, only RDFS and OWL-Horst are supported.
  *
  * @author <NAME>
  *
  */
object RDFGraphMaterializer {

  def main(args: Array[String]) {
    parser.parse(args, Config()) match {
      case Some(config) =>
        run(config.in, config.out, config.profile, config.properties, config.writeToSingleFile, config.sortedOutput, config.parallelism)
      case None =>
        // scalastyle:off println
        println(parser.usage)
        // scalastyle:on println
    }
  }

  def run(input: Seq[URI], output: URI, profile: ReasoningProfile, properties: Seq[Node] = Seq(),
          writeToSingleFile: Boolean, sortedOutput: Boolean, parallelism: Int): Unit = {
    // register the custom classes for Kryo serializer
    val conf = new SparkConf()
    conf.registerKryoClasses(Array(classOf[org.apache.jena.graph.Triple]))
    conf.set("spark.extraListeners", "net.sansa_stack.inference.spark.utils.CustomSparkListener")

    // the SPARK config
    val session = SparkSession.builder
      .appName(s"SPARK $profile Reasoning")
      .master("local[4]")
//      .config("spark.eventLog.enabled", "true")
      .config("spark.hadoop.validateOutputSpecs", "false") // override output files
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .config("spark.default.parallelism", parallelism)
      .config("spark.ui.showConsoleProgress", "false")
      .config("spark.sql.shuffle.partitions", parallelism)
      .config(conf)
      .getOrCreate()

//    println(session.conf.getAll.mkString("\n"))

//    val g = RDFGraphLoader.loadFromDiskAsDataset(session, input).distinct()
//    val g_inf = new ForwardRuleReasonerRDFSDataset(session).apply(g)
//    println(s"|G_inf| = ${g_inf.size()}")

    // load triples from disk
    val graph = RDFGraphLoader.loadFromDisk(session, input, parallelism)
//    println(s"|G| = ${graph.size()}")

    // create reasoner
    val reasoner = profile match {
      case TRANSITIVE => new TransitiveReasoner(session.sparkContext, properties, parallelism)
      case RDFS => new ForwardRuleReasonerRDFS(session.sparkContext, parallelism)
      case RDFS_SIMPLE =>
        val r = new ForwardRuleReasonerRDFS(session.sparkContext, parallelism)
{'fraction_non_alphanumeric': 0.12073400788886983, 'fraction_numerical': 0.0003429943405933802, 'mean_word_length': 4.014617368873603, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1924113', 'n_tokens_mistral': 1905, 'n_tokens_neox': 1814, 'n_words': 349}