<commit_before>package com.asakusafw.spark.runtime

import scala.collection.JavaConversions._

import org.apache.spark.SparkConf
import org.slf4j.LoggerFactory

import com.asakusafw.bridge.launch.LaunchConfiguration
import com.asakusafw.bridge.stage.StageInfo

object Launcher {

  val Logger = LoggerFactory.getLogger(getClass)

  def main(args: Array[String]): Unit = {
    try {
      val conf = LaunchConfiguration.parse(Thread.currentThread.getContextClassLoader, args: _*)

      val sparkClient = conf.getStageClient.asSubclass(classOf[SparkClient]).newInstance()

      val sparkConf = new SparkConf()

      sparkConf.setHadoopConf(StageInfo.KEY_NAME, conf.getStageInfo.serialize)

      conf.getHadoopProperties.foreach {
        case (key, value) => sparkConf.setHadoopConf(key, value)
      }

      conf.getEngineProperties.foreach {
        case (key, value) => sparkConf.set(key, value)
      }

      if (!sparkConf.contains(Props.Parallelism) && !sparkConf.contains("spark.default.parallelism")) {
        if (Logger.isWarnEnabled) {
          Logger.warn(s"`${Props.Parallelism}` is not set, we set parallelism to ${Props.ParallelismFallback}.")
        }
      }

      sparkClient.execute(sparkConf)
    } catch {
      case t: Throwable =>
        Logger.error(s"SparkClient throws: ${t.getMessage}", t)
        throw t
    }
  }
}<commit_msg>Support Asakusa RuntimeContext facilities.<commit_after>package com.asakusafw.spark.runtime

import scala.collection.JavaConversions._

import org.apache.spark.SparkConf
import org.slf4j.LoggerFactory

import com.asakusafw.bridge.launch.LaunchConfiguration
import com.asakusafw.bridge.stage.StageInfo
import com.asakusafw.runtime.core.context.RuntimeContext

object Launcher {

  val Logger = LoggerFactory.getLogger(getClass)

  def main(args: Array[String]): Unit = {
    try {
      val cl = Thread.currentThread.getContextClassLoader

      RuntimeContext.set(RuntimeContext.DEFAULT.apply(System.getenv))
      RuntimeContext.get.verifyApplication(cl)
      val conf = LaunchConfiguration.parse(cl, args: _*)

      val sparkClient = conf.getStageClient.asSubclass(classOf[SparkClient]).newInstance()

      val sparkConf = new SparkConf()

      sparkConf.setHadoopConf(StageInfo.KEY_NAME, conf.getStageInfo.serialize)

      conf.getHadoopProperties.foreach {
        case (key, value) => sparkConf.setHadoopConf(key, value)
      }

      conf.getEngineProperties.foreach {
        case (key, value) => sparkConf.set(key, value)
      }

      if (!sparkConf.contains(Props.Parallelism) && !sparkConf.contains("spark.default.parallelism")) {
        if (Logger.isWarnEnabled) {
          Logger.warn(s"`${Props.Parallelism}` is not set, we set parallelism to ${Props.ParallelismFallback}.")
        }
      }

      if (!RuntimeContext.get.isSimulation) {
        sparkClient.execute(sparkConf)
      }
    } catch {
      case t: Throwable =>
        Logger.error(s"SparkClient throws: ${t.getMessage}", t)
        throw t
    }
  }
}
{'fraction_non_alphanumeric': 0.1111111111111111, 'fraction_numerical': 0.0006653359946773121, 'mean_word_length': 3.4416543574593796, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, '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': '678379', 'n_tokens_mistral': 942, 'n_tokens_neox': 909, 'n_words': 156}