Skip to content

[SPARK-2608] fix executor backend launch commond over mesos mode #1986

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
scwf wants to merge 18 commits into from
Closed

[SPARK-2608] fix executor backend launch commond over mesos mode #1986

Show file tree
Hide file tree
Changes from 14 commits
Commits
Show all changes
18 commits
Select commit Hold shift + click to select a range
5b150a4
fix executor backend launch commond over mesos mode
Jul 21, 2014
fdc3cb1
fix code format
Jul 21, 2014
4290c71
fix compile error
scwf Aug 6, 2014
bc92bdb
fix compile error
scwf Aug 6, 2014
bb5ada5
delete empty lines
scwf Aug 6, 2014
a954c5c
delete empty lines
scwf Aug 6, 2014
14578b2
fix import
scwf Aug 7, 2014
7192f6f
fix import
scwf Aug 7, 2014
d80e455
fix import
scwf Aug 7, 2014
4f701b5
delete no use import
scwf Aug 7, 2014
e515fc8
Update MesosSchedulerBackend.scala
scwf Aug 7, 2014
80d99b0
Update CoarseMesosSchedulerBackend.scala
scwf Aug 7, 2014
4f694c0
Update MesosSchedulerBackend.scala
scwf Aug 7, 2014
0e65083
Merge branch 'master' into mesosfix
scwf Aug 16, 2014
3e73513
fix conflict
scwf Aug 26, 2014
3b8a276
propagate env to the target executor process
scwf Aug 26, 2014
2fe459a
add PYTHONPATH to env
scwf Aug 26, 2014
c2fe5b0
Update spark-env.sh.template
scwf Aug 26, 2014
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 32 additions & 35 deletions ...src/main/scala/org/apache/spark/scheduler/cluster/mesos/CoarseMesosSchedulerBackend.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@

package org.apache.spark.scheduler.cluster.mesos

import java.io.File
import java.util.{List => JList}
import java.util.Collections

Expand All @@ -28,9 +27,12 @@ import org.apache.mesos.{Scheduler => MScheduler}
import org.apache.mesos._
import org.apache.mesos.Protos.{TaskInfo => MesosTaskInfo, TaskState => MesosTaskState, _}

import org.apache.spark.{Logging, SparkContext, SparkException}
import org.apache.spark.{SparkConf, Logging, SparkContext, SparkException}
import org.apache.spark.scheduler.TaskSchedulerImpl
import org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend
import org.apache.spark.deploy.Command
import org.apache.spark.deploy.worker.CommandUtils
import org.apache.spark.util.Utils

/**
* A SchedulerBackend that runs tasks on Mesos, but uses "coarse-grained" tasks, where it holds
Expand Down Expand Up @@ -111,48 +113,43 @@ private[spark] class CoarseMesosSchedulerBackend(

def createCommand(offer: Offer, numCores: Int): CommandInfo = {
val environment = Environment.newBuilder()
val extraClassPath = conf.getOption("spark.executor.extraClassPath")
extraClassPath.foreach { cp =>
environment.addVariables(
Environment.Variable.newBuilder().setName("SPARK_CLASSPATH").setValue(cp).build())
}
val mesosCommand = CommandInfo.newBuilder()
.setEnvironment(environment)

val driverUrl = "akka.tcp://spark@%s:%s/user/%s".format(
conf.get("spark.driver.host"), conf.get("spark.driver.port"),
CoarseGrainedSchedulerBackend.ACTOR_NAME)
val args = Seq(driverUrl, offer.getSlaveId.getValue, offer.getHostname, numCores.toString)
val extraJavaOpts = conf.getOption("spark.executor.extraJavaOptions")
.map(Utils.splitCommandString).getOrElse(Seq.empty)

val libraryPathOption = "spark.executor.extraLibraryPath"
val extraLibraryPath = conf.getOption(libraryPathOption).map(p => s"-Djava.library.path=$p")
val extraOpts = Seq(extraJavaOpts, extraLibraryPath).flatten.mkString(" ")
// Start executors with a few necessary configs for registering with the scheduler
val sparkJavaOpts = Utils.sparkJavaOpts(conf, SparkConf.isExecutorStartupConf)
val javaOpts = sparkJavaOpts ++ extraJavaOpts

sc.executorEnvs.foreach { case (key, value) =>
environment.addVariables(Environment.Variable.newBuilder()
.setName(key)
.setValue(value)
.build())
val classPathEntries = conf.getOption("spark.executor.extraClassPath").toSeq.flatMap { cp =>
cp.split(java.io.File.pathSeparator)
}
val command = CommandInfo.newBuilder()
.setEnvironment(environment)
val driverUrl = "akka.tcp://spark@%s:%s/user/%s".format(
conf.get("spark.driver.host"),
conf.get("spark.driver.port"),
CoarseGrainedSchedulerBackend.ACTOR_NAME)
val libraryPathEntries =
conf.getOption("spark.executor.extraLibraryPath").toSeq.flatMap { cp =>
cp.split(java.io.File.pathSeparator)
}

val command = Command(
"org.apache.spark.executor.CoarseGrainedExecutorBackend", args, sc.executorEnvs,
Copy link
Contributor

@liancheng liancheng Aug 26, 2014

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Only passing sc.executorEnvs here to the command object here is not enough. Because we only use command with CommandUtils.buildCommandSeq below to generate the command line string, and in this case, command.environment is only used to run bin/compute-classpath (see here), not propagated to the target executor process.

Copy link
Contributor Author

@scwf scwf Aug 26, 2014

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes, here i should set env to CommandInfo to propagate to the target executor process.

classPathEntries, libraryPathEntries, javaOpts)

val uri = conf.get("spark.executor.uri", null)
if (uri == null) {
val runScript = new File(sparkHome, "./bin/spark-class").getCanonicalPath
command.setValue(
"\"%s\" org.apache.spark.executor.CoarseGrainedExecutorBackend %s %s %s %s %d".format(
runScript, extraOpts, driverUrl, offer.getSlaveId.getValue, offer.getHostname, numCores))
if ( uri == null ) {
mesosCommand.setValue(CommandUtils.buildCommandSeq(command, sc.executorMemory,
sparkHome).mkString("\"", "\" \"", "\""))
} else {
// Grab everything to the first '.'. We'll use that and '*' to
// glob the directory "correctly".
val basename = uri.split('/').last.split('.').head
command.setValue(
("cd %s*; " +
"./bin/spark-class org.apache.spark.executor.CoarseGrainedExecutorBackend %s %s %s %s %d")
.format(basename, extraOpts, driverUrl, offer.getSlaveId.getValue,
offer.getHostname, numCores))
command.addUris(CommandInfo.URI.newBuilder().setValue(uri))
mesosCommand.setValue(CommandUtils.buildCommandSeq(command, sc.executorMemory,
basename).mkString("\"", "\" \"", "\""))
mesosCommand.addUris(CommandInfo.URI.newBuilder().setValue(uri))
}
command.build()
mesosCommand.build()
}

override def offerRescinded(d: SchedulerDriver, o: OfferID) {}
Expand Down
46 changes: 35 additions & 11 deletions core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/MesosSchedulerBackend.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@

package org.apache.spark.scheduler.cluster.mesos

import java.io.File
import java.util.{ArrayList => JArrayList, List => JList}
import java.util.Collections

Expand All @@ -29,9 +28,12 @@ import org.apache.mesos.{Scheduler => MScheduler}
import org.apache.mesos._
import org.apache.mesos.Protos.{TaskInfo => MesosTaskInfo, TaskState => MesosTaskState, _}

import org.apache.spark.{Logging, SparkContext, SparkException, TaskState}
import org.apache.spark.{SparkConf, Logging, SparkContext, TaskState, SparkException}
import org.apache.spark.scheduler.{ExecutorExited, ExecutorLossReason, SchedulerBackend, SlaveLost, TaskDescription, TaskSchedulerImpl, WorkerOffer}
import org.apache.spark.util.Utils
import org.apache.spark.deploy.Command
import org.apache.spark.deploy.worker.CommandUtils
import org.apache.spark.util.Utils

/**
* A SchedulerBackend for running fine-grained tasks on Mesos. Each Spark task is mapped to a
Expand All @@ -56,6 +58,7 @@ private[spark] class MesosSchedulerBackend(
// Which slave IDs we have executors on
val slaveIdsWithExecutors = new HashSet[String]
val taskIdToSlaveId = new HashMap[Long, String]
val conf = sc.conf

// An ExecutorInfo for our tasks
var execArgs: Array[Byte] = null
Expand Down Expand Up @@ -96,26 +99,47 @@ private[spark] class MesosSchedulerBackend(
.setValue(value)
.build())
}
val command = CommandInfo.newBuilder()
val mesosCommand = CommandInfo.newBuilder()
.setEnvironment(environment)
val uri = sc.conf.get("spark.executor.uri", null)
if (uri == null) {
command.setValue(new File(sparkHome, "/sbin/spark-executor").getCanonicalPath)

val extraJavaOpts = conf.getOption("spark.executor.extraJavaOptions")
.map(Utils.splitCommandString).getOrElse(Seq.empty)

// Start executors with a few necessary configs for registering with the scheduler
val sparkJavaOpts = Utils.sparkJavaOpts(conf, SparkConf.isExecutorStartupConf)
val javaOpts = sparkJavaOpts ++ extraJavaOpts

val classPathEntries = conf.getOption("spark.executor.extraClassPath").toSeq.flatMap { cp =>
cp.split(java.io.File.pathSeparator)
}
val libraryPathEntries =
conf.getOption("spark.executor.extraLibraryPath").toSeq.flatMap { cp =>
cp.split(java.io.File.pathSeparator)
}

val command = Command(
"org.apache.spark.executor.MesosExecutorBackend", Nil, sc.executorEnvs,
classPathEntries, libraryPathEntries, javaOpts)
Copy link
Contributor

@liancheng liancheng Aug 26, 2014

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

PYTHONPATH set by sbin/spark-executor is ignored here.

Copy link
Contributor

@liancheng liancheng Aug 26, 2014

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should set PYTHONPATH as what sbin/spark-executor does somehow.

Copy link
Contributor Author

@scwf scwf Aug 26, 2014

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i am wondering it will be ok if set PYTHONPATH to environment as flowing
1 add PYTHONPATH to system env in driver process
2

    val environment = Environment.newBuilder()
    sc.executorEnvs.foreach { case (key, value) =>
      environment.addVariables(Environment.Variable.newBuilder()
        .setName(key)
        .setValue(value)
        .build())
    }
    environment.addVariables(Environment.Variable.newBuilder()
      .setName("PYTHONPATH")
      .setValue(sys.env.getOrElse("PYTHONPATH", ""))
      .build())

in this way maybe we can not get the PYTHONPATH by system env in executor process~


val uri = conf.get("spark.executor.uri", null)
if ( uri == null ) {
mesosCommand.setValue(CommandUtils.buildCommandSeq(command, sc.executorMemory,
sparkHome).mkString("\"", "\" \"", "\""))
} else {
// Grab everything to the first '.'. We'll use that and '*' to
// glob the directory "correctly".
val basename = uri.split('/').last.split('.').head
command.setValue("cd %s*; ./sbin/spark-executor".format(basename))
command.addUris(CommandInfo.URI.newBuilder().setValue(uri))
mesosCommand.setValue(CommandUtils.buildCommandSeq(command, sc.executorMemory,
basename).mkString("\"", "\" \"", "\""))
mesosCommand.addUris(CommandInfo.URI.newBuilder().setValue(uri))
}

val memory = Resource.newBuilder()
.setName("mem")
.setType(Value.Type.SCALAR)
.setScalar(Value.Scalar.newBuilder().setValue(sc.executorMemory).build())
.build()
ExecutorInfo.newBuilder()
.setExecutorId(ExecutorID.newBuilder().setValue(execId).build())
.setCommand(command)
.setCommand(mesosCommand)
.setData(ByteString.copyFrom(createExecArg()))
.addResources(memory)
.build()
Expand Down