Skip to content
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

SPARK-1099:Spark's local mode should probably respect spark.cores.max by default #110

Closed
wants to merge 3 commits into from
Closed

SPARK-1099:Spark's local mode should probably respect spark.cores.max by default #110

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6ae1ee8
Add a static function in LocalBackEnd to let it use spark.cores.max s…
Mar 9, 2014
78b9c60
1 SparkContext MASTER=local pattern use default cores instead of 1 to…
Mar 9, 2014
731aefa
1 LocalBackend not change 2 In SparkContext do some process to the co…
Mar 12, 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
5 changes: 4 additions & 1 deletion core/src/main/scala/org/apache/spark/SparkContext.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1204,7 +1204,10 @@ object SparkContext extends Logging {
master match {
case "local" =>
val scheduler = new TaskSchedulerImpl(sc, MAX_LOCAL_TASK_FAILURES, isLocal = true)
val backend = new LocalBackend(scheduler, 1)
// Use user specified in config, up to all available cores
val realCores = Runtime.getRuntime.availableProcessors()
val toUseCores = math.min(sc.conf.getInt("spark.cores.max", realCores), realCores)
val backend = new LocalBackend(scheduler, toUseCores)
scheduler.initialize(backend)
scheduler

Expand Down
4 changes: 2 additions & 2 deletions core/src/test/scala/org/apache/spark/FileSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat
class FileSuite extends FunSuite with LocalSparkContext {

test("text files") {
sc = new SparkContext("local", "test")
sc = new SparkContext("local[1]", "test")
val tempDir = Files.createTempDir()
val outputDir = new File(tempDir, "output").getAbsolutePath
val nums = sc.makeRDD(1 to 4)
Expand Down Expand Up @@ -175,7 +175,7 @@ class FileSuite extends FunSuite with LocalSparkContext {

test("write SequenceFile using new Hadoop API") {
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat
sc = new SparkContext("local", "test")
sc = new SparkContext("local[1]", "test")
val tempDir = Files.createTempDir()
val outputDir = new File(tempDir, "output").getAbsolutePath
val nums = sc.makeRDD(1 to 3).map(x => (new IntWritable(x), new Text("a" * x)))
Expand Down
19 changes: 16 additions & 3 deletions core/src/test/scala/org/apache/spark/SparkContextSchedulerCreationSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,10 +27,10 @@ import org.apache.spark.scheduler.local.LocalBackend
class SparkContextSchedulerCreationSuite
extends FunSuite with PrivateMethodTester with LocalSparkContext with Logging {

def createTaskScheduler(master: String): TaskSchedulerImpl = {
def createTaskScheduler(master: String, conf: SparkConf = new SparkConf()): TaskSchedulerImpl = {
// Create local SparkContext to setup a SparkEnv. We don't actually want to start() the
// real schedulers, so we don't want to create a full SparkContext with the desired scheduler.
sc = new SparkContext("local", "test")
sc = new SparkContext("local", "test", conf)
val createTaskSchedulerMethod = PrivateMethod[TaskScheduler]('createTaskScheduler)
val sched = SparkContext invokePrivate createTaskSchedulerMethod(sc, master, "test")
sched.asInstanceOf[TaskSchedulerImpl]
Expand All @@ -44,13 +44,26 @@ class SparkContextSchedulerCreationSuite
}

test("local") {
val sched = createTaskScheduler("local")
var conf = new SparkConf()
conf.set("spark.cores.max", "1")
val sched = createTaskScheduler("local", conf)
sched.backend match {
case s: LocalBackend => assert(s.totalCores === 1)
case _ => fail()
}
}

test("local-cores-exceed") {
val cores = Runtime.getRuntime.availableProcessors() + 1
var conf = new SparkConf()
conf.set("spark.cores.max", cores.toString)
val sched = createTaskScheduler("local", conf)
sched.backend match {
case s: LocalBackend => assert(s.totalCores === Runtime.getRuntime.availableProcessors())
case _ => fail()
}
}

test("local-n") {
val sched = createTaskScheduler("local[5]")
assert(sched.maxTaskFailures === 1)
Expand Down