<commit_before>package spark.scheduler.cluster

import spark.{Utils, Logging, SparkContext}
import spark.deploy.client.{Client, ClientListener}
import spark.deploy.{Command, JobDescription}
import scala.collection.mutable.HashMap
import java.io.File

private[spark] class SparkDeploySchedulerBackend(
    scheduler: ClusterScheduler,
    sc: SparkContext,
    master: String,
    jobName: String)
  extends StandaloneSchedulerBackend(scheduler, sc.env.actorSystem)
  with ClientListener
  with Logging {

  var client: Client = null
  var stopping = false
  var shutdownCallback : (SparkDeploySchedulerBackend) => Unit = _

  val maxCores = System.getProperty("spark.cores.max", Int.MaxValue.toString).toInt
  val executorIdToSlaveId = new HashMap[String, String]

  // Memory used by each executor (in megabytes)
  val executorMemory = {
    if (System.getenv("SPARK_MEM") != null) {
      Utils.memoryStringToMb(System.getenv("SPARK_MEM"))
      // TODO: Might need to add some extra memory for the non-heap parts of the JVM
    } else {
      512
    }
  }

  override def start() {
    super.start()

    val masterUrl = "akka://spark@%s:%s/user/%s".format(
      System.getProperty("spark.master.host"), System.getProperty("spark.master.port"),
      StandaloneSchedulerBackend.ACTOR_NAME)
    val args = Seq(masterUrl, "{{SLAVEID}}", "{{HOSTNAME}}", "{{CORES}}")
    val command = Command("spark.executor.StandaloneExecutorBackend", args, sc.executorEnvs)
    val jobDesc = new JobDescription(jobName, maxCores, executorMemory, command, new File(sc.sparkHome))

    client = new Client(sc.env.actorSystem, master, jobDesc, this)
    client.start()
  }

  override def stop() {
    stopping = true;
    super.stop()
    client.stop()
    if (shutdownCallback != null) {
      shutdownCallback(this)
    }
  }

  def connected(jobId: String) {
    logInfo("Connected to Spark cluster with job ID " + jobId)
  }

  def disconnected() {
    if (!stopping) {
      logError("Disconnected from Spark cluster!")
      scheduler.error("Disconnected from Spark cluster")
    }
  }

  def executorAdded(id: String, workerId: String, host: String, cores: Int, memory: Int) {
    executorIdToSlaveId += id -> workerId
    logInfo("Granted executor ID %s on host %s with %d cores, %s RAM".format(
       id, host, cores, Utils.memoryMegabytesToString(memory)))
  }

  def executorRemoved(id: String, message: String, exitStatus: Option[Int]) {
    val reason: ExecutorLossReason = exitStatus match {
      case Some(code) => ExecutorExited(code)
      case None => SlaveLost(message)
    }
    logInfo("Executor %s removed: %s".format(id, message))
    executorIdToSlaveId.get(id) match {
      case Some(slaveId) => 
        executorIdToSlaveId.remove(id)
        scheduler.slaveLost(slaveId, reason)
      case None =>
        logInfo("No slave ID known for executor %s".format(id))
    }
  }
}
<commit_msg>get sparkHome the correct way<commit_after>package spark.scheduler.cluster

import spark.{Utils, Logging, SparkContext}
import spark.deploy.client.{Client, ClientListener}
import spark.deploy.{Command, JobDescription}
import scala.collection.mutable.HashMap
import java.io.File

private[spark] class SparkDeploySchedulerBackend(
    scheduler: ClusterScheduler,
    sc: SparkContext,
    master: String,
    jobName: String)
  extends StandaloneSchedulerBackend(scheduler, sc.env.actorSystem)
  with ClientListener
  with Logging {

  var client: Client = null
  var stopping = false
  var shutdownCallback : (SparkDeploySchedulerBackend) => Unit = _

  val maxCores = System.getProperty("spark.cores.max", Int.MaxValue.toString).toInt
  val executorIdToSlaveId = new HashMap[String, String]

  // Memory used by each executor (in megabytes)
  val executorMemory = {
    if (System.getenv("SPARK_MEM") != null) {
      Utils.memoryStringToMb(System.getenv("SPARK_MEM"))
      // TODO: Might need to add some extra memory for the non-heap parts of the JVM
    } else {
      512
    }
  }

  override def start() {
    super.start()

    val masterUrl = "akka://spark@%s:%s/user/%s".format(
      System.getProperty("spark.master.host"), System.getProperty("spark.master.port"),
      StandaloneSchedulerBackend.ACTOR_NAME)
    val args = Seq(masterUrl, "{{SLAVEID}}", "{{HOSTNAME}}", "{{CORES}}")
    val command = Command("spark.executor.StandaloneExecutorBackend", args, sc.executorEnvs)
    val jobDesc = new JobDescription(jobName, maxCores, executorMemory, command, new File(sc.getSparkHome()))

    client = new Client(sc.env.actorSystem, master, jobDesc, this)
    client.start()
  }

  override def stop() {
    stopping = true;
    super.stop()
    client.stop()
    if (shutdownCallback != null) {
      shutdownCallback(this)
    }
  }

  def connected(jobId: String) {
    logInfo("Connected to Spark cluster with job ID " + jobId)
  }

  def disconnected() {
    if (!stopping) {
      logError("Disconnected from Spark cluster!")
      scheduler.error("Disconnected from Spark cluster")
    }
  }

  def executorAdded(id: String, workerId: String, host: String, cores: Int, memory: Int) {
    executorIdToSlaveId += id -> workerId
    logInfo("Granted executor ID %s on host %s with %d cores, %s RAM".format(
       id, host, cores, Utils.memoryMegabytesToString(memory)))
  }

  def executorRemoved(id: String, message: String, exitStatus: Option[Int]) {
    val reason: ExecutorLossReason = exitStatus match {
      case Some(code) => ExecutorExited(code)
      case None => SlaveLost(message)
    }
    logInfo("Executor %s removed: %s".format(id, message))
    executorIdToSlaveId.get(id) match {
      case Some(slaveId) => 
        executorIdToSlaveId.remove(id)
        scheduler.slaveLost(slaveId, reason)
      case None =>
        logInfo("No slave ID known for executor %s".format(id))
    }
  }
}

{'fraction_non_alphanumeric': 0.11776824034334764, 'fraction_numerical': 0.0010300429184549357, 'mean_word_length': 3.687047465808528, 'pattern_counts': {'":': 0, '<': 3, '<?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': '4672752', 'n_tokens_mistral': 1887, 'n_tokens_neox': 1831, 'n_words': 452}