Skip to content

[SPARK-8675] Executors created by LocalBackend won't get the same classpath as other executor backends #7091

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
coderplay wants to merge 6 commits into from
Closed

[SPARK-8675] Executors created by LocalBackend won't get the same classpath as other executor backends #7091

Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
45bf62c
SPARK-8675 Executors created by LocalBackend won't get the same class…
coderplay Jun 29, 2015
e528be7
Merge branch 'master' of https://github.com/apache/spark
coderplay Jun 29, 2015
01f5d1a
Merge branch 'master' of https://github.com/apache/spark
coderplay Jul 9, 2015
84ad2cd
Use system specific path separator instead of ','
coderplay Jul 9, 2015
d215b7f
Follows spark standard scala style, make the auto testing happy
coderplay Jul 9, 2015
365838f
Fixed java.net.MalformedURLException, add default scheme, support rel…
coderplay Jul 9, 2015
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
19 changes: 17 additions & 2 deletions core/src/main/scala/org/apache/spark/scheduler/local/LocalBackend.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@

package org.apache.spark.scheduler.local

import java.io.File
import java.net.URL
import java.nio.ByteBuffer

import org.apache.spark.{Logging, SparkConf, SparkContext, SparkEnv, TaskState}
Expand All @@ -40,6 +42,7 @@ private case class StopExecutor()
*/
private[spark] class LocalEndpoint(
override val rpcEnv: RpcEnv,
userClassPath: Seq[URL],
scheduler: TaskSchedulerImpl,
executorBackend: LocalBackend,
private val totalCores: Int)
Expand All @@ -51,7 +54,7 @@ private[spark] class LocalEndpoint(
private val localExecutorHostname = "localhost"

private val executor = new Executor(
localExecutorId, localExecutorHostname, SparkEnv.get, isLocal = true)
localExecutorId, localExecutorHostname, SparkEnv.get, userClassPath, isLocal = true)

override def receive: PartialFunction[Any, Unit] = {
case ReviveOffers =>
Expand Down Expand Up @@ -97,10 +100,22 @@ private[spark] class LocalBackend(

private val appId = "local-" + System.currentTimeMillis
var localEndpoint: RpcEndpointRef = null
private val userClassPath = getUserClasspath(conf)

/**
* Returns a list of URLs representing the user classpath.
*
* @param conf Spark configuration.
*/
def getUserClasspath(conf: SparkConf): Seq[URL] = {
val userClassPathStr = conf.getOption("spark.executor.extraClassPath")
userClassPathStr.map(_.split(File.pathSeparator)).toSeq.flatten.map(new File(_).toURI.toURL)
}

override def start() {
localEndpoint = SparkEnv.get.rpcEnv.setupEndpoint(
"LocalBackendEndpoint", new LocalEndpoint(SparkEnv.get.rpcEnv, scheduler, this, totalCores))
"LocalBackendEndpoint",
new LocalEndpoint(SparkEnv.get.rpcEnv, userClassPath, scheduler, this, totalCores))
}

override def stop() {
Expand Down