<commit_before>package org.sparkpipe.test.spark

import scala.io.Source
import scala.util.{Try, Success, Failure}
import org.apache.log4j.Level
import org.apache.spark.{SparkContext, SparkConf}
import org.sparkpipe.test.util.Base

trait SparkCluster extends SparkLocal with Base {
    /** Loading Spark configuration for cluster. */
    final protected def clusterConf(): SparkConf = {
        // OS check
        val system = System.getProperty("os.name").toLowerCase()
        require(system.startsWith("mac") || system.startsWith("linux"),
            "Spark Cluster runs only on Unix-like OS")

        // throw exception if SPARK_HOME is not set
        val SPARK_HOME = Option(System.getenv("SPARK_HOME")) match {
            case Some(opt) => opt
            case None => throw new UnsupportedOperationException(
                "Spark Cluster mode requires SPARK_HOME to be set"
            )
        }

        // throw exception if SPARK_MASTER_URL is not set
        val SPARK_MASTER_URL = Option(System.getenv("SPARK_MASTER_URL")) match {
            case Some(opt) => opt
            case None => throw new UnsupportedOperationException(
                "Spark Cluster mode requires SPARK_MASTER_URL to be set"
            )
        }

        // read internal classes to pass to executor JVMs
        val SPARK_PREPEND_CLASSES = Seq(
            targetDirectory() + / + "streams" + / + "runtime" + / + "exportedProducts" +
            / + "$global" + / + "streams" + / + "export",
            targetDirectory() + / + "streams" + / + "test" + / + "exportedProducts" +
            / + "$global" + / + "streams" + / + "export"
        ).flatMap(filepath =>
            Source.fromFile(filepath).getLines()
        ).mkString(`:`)

        // in case of dependencies pass them to executor JVMs as libs
        val SPARK_MANAGED_DEPENDENCIES = Seq(
            targetDirectory() + / + "streams" + / + "compile" + / + "dependencyClasspath" +
            / + "$global" + / + "streams" + / + "export"
        ).flatMap(filepath =>
            Source.fromFile(filepath).getLines()
        ).mkString(`:`)

        new SparkConf().
           setMaster(SPARK_MASTER_URL).
           setAppName("spark-local-cluster-test").
           set("spark.driver.memory", "1g").
           set("spark.executor.memory", "1g").
           set("spark.executorEnv.SPARK_HOME", SPARK_HOME).
           set("spark.executor.extraClassPath", SPARK_PREPEND_CLASSES).
           set("spark.executor.extraLibraryPath", SPARK_MANAGED_DEPENDENCIES)
    }

    /**
     * Tries loading configuration for cluster.
     * `fallback` option allows to load local mode instead in case of failure.
     */
    private def tryConf(fallback: Boolean = true): SparkConf = Try(clusterConf()) match {
        case Success(conf) => conf
        case Failure(e) => {
            if (fallback) {
                println("[info] Spark cluster failed to load: " + e.getMessage())
                println("[info] Using Spark local context instead")
                localConf()
            } else {
                throw e
            }
        }
    }

    override def startSparkContext() {
        setLoggingLevel(Level.ERROR)
        val conf = tryConf(fallback=true)
        _sc = new SparkContext(conf)
    }
}
<commit_msg>spark master address replace, minor fixes<commit_after>package org.sparkpipe.test.spark

import scala.io.Source
import scala.util.{Try, Success, Failure}
import org.apache.log4j.Level
import org.apache.spark.{SparkContext, SparkConf}
import org.sparkpipe.test.util.Base

trait SparkCluster extends SparkLocal with Base {
    /** Loading Spark configuration for cluster. */
    final protected def clusterConf(): SparkConf = {
        // OS check
        val system = System.getProperty("os.name").toLowerCase()
        require(system.startsWith("mac") || system.startsWith("linux"),
            "Spark Cluster runs only on Unix-like OS")

        // throw exception if SPARK_HOME is not set
        val SPARK_HOME = Option(System.getenv("SPARK_HOME")) match {
            case Some(opt) => opt
            case None => throw new UnsupportedOperationException(
                "Spark Cluster mode requires SPARK_HOME to be set"
            )
        }

        // throw exception if SPARK_MASTER_ADDRESS is not set
        val SPARK_MASTER_ADDRESS = Option(System.getenv("SPARK_MASTER_ADDRESS")) match {
            case Some(opt) => opt
            case None => throw new UnsupportedOperationException(
                "Spark Cluster mode requires SPARK_MASTER_ADDRESS to be set"
            )
        }

        // read internal classes to pass to executor JVMs
        val SPARK_PREPEND_CLASSES = Seq(
            targetDirectory() + / + "streams" + / + "runtime" + / + "exportedProducts" +
            / + "$global" + / + "streams" + / + "export",
            targetDirectory() + / + "streams" + / + "test" + / + "exportedProducts" +
            / + "$global" + / + "streams" + / + "export"
        ).flatMap(filepath =>
            Source.fromFile(filepath).getLines()
        ).mkString(`:`)

        // in case of dependencies pass them to executor JVMs as libs
        val SPARK_MANAGED_DEPENDENCIES = Seq(
            targetDirectory() + / + "streams" + / + "compile" + / + "dependencyClasspath" +
            / + "$global" + / + "streams" + / + "export"
        ).flatMap(filepath =>
            Source.fromFile(filepath).getLines()
        ).mkString(`:`)

        new SparkConf().
           setMaster(SPARK_MASTER_ADDRESS).
           setAppName("spark-cluster-test").
           set("spark.driver.memory", "1g").
           set("spark.executor.memory", "1g").
           set("spark.executorEnv.SPARK_HOME", SPARK_HOME).
           set("spark.executor.extraClassPath", SPARK_PREPEND_CLASSES).
           set("spark.executor.extraLibraryPath", SPARK_MANAGED_DEPENDENCIES)
    }

    /**
     * Tries loading configuration for cluster.
     * `fallback` option allows to load local mode instead in case of failure.
     */
    private def tryConf(fallback: Boolean = true): SparkConf = Try(clusterConf()) match {
        case Success(conf) => conf
        case Failure(e) => {
            if (fallback) {
                println("[info] Spark cluster failed to load: " + e.getMessage())
                println("[info] Using Spark local context instead")
                localConf()
            } else {
                throw e
            }
        }
    }

    override def startSparkContext() {
        setLoggingLevel(Level.ERROR)
        val conf = tryConf(fallback=true)
        _sc = new SparkContext(conf)
    }
}

{'fraction_non_alphanumeric': 0.11718514044095439, 'fraction_numerical': 0.000906070673512534, 'mean_word_length': 2.1255309108069844, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 19, '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': '2493933', 'n_tokens_mistral': 1971, 'n_tokens_neox': 1874, 'n_words': 453}