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-35083][CORE] Support remote scheduler pool files #32184

Closed
wants to merge 4 commits into from
Closed

[SPARK-35083][CORE] Support remote scheduler pool files #32184

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8bd973d
Support remote scheduler pool file
ulysses-you Apr 15, 2021
18e9af3
Merge branch 'master' of https://github.com/apache/spark into SPARK-3…
ulysses-you Apr 15, 2021
4eec581
docs
ulysses-you Apr 15, 2021
bfd2a55
add test
ulysses-you Apr 16, 2021
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
13 changes: 8 additions & 5 deletions core/src/main/scala/org/apache/spark/scheduler/SchedulableBuilder.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,13 +17,15 @@

package org.apache.spark.scheduler

import java.io.{FileInputStream, InputStream}
import java.io.InputStream
import java.util.{Locale, NoSuchElementException, Properties}

import scala.util.control.NonFatal
import scala.xml.{Node, XML}

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.hadoop.fs.Path

import org.apache.spark.SparkContext
import org.apache.spark.internal.Logging
import org.apache.spark.internal.config.SCHEDULER_ALLOCATION_FILE
import org.apache.spark.scheduler.SchedulingMode.SchedulingMode
Expand Down Expand Up @@ -54,10 +56,10 @@ private[spark] class FIFOSchedulableBuilder(val rootPool: Pool)
}
}

private[spark] class FairSchedulableBuilder(val rootPool: Pool, conf: SparkConf)
private[spark] class FairSchedulableBuilder(val rootPool: Pool, sc: SparkContext)
extends SchedulableBuilder with Logging {

val schedulerAllocFile = conf.get(SCHEDULER_ALLOCATION_FILE)
val schedulerAllocFile = sc.conf.get(SCHEDULER_ALLOCATION_FILE)
val DEFAULT_SCHEDULER_FILE = "fairscheduler.xml"
val FAIR_SCHEDULER_PROPERTIES = SparkContext.SPARK_SCHEDULER_POOL
val DEFAULT_POOL_NAME = "default"
Expand All @@ -74,7 +76,8 @@ private[spark] class FairSchedulableBuilder(val rootPool: Pool, conf: SparkConf)
var fileData: Option[(InputStream, String)] = None
try {
fileData = schedulerAllocFile.map { f =>
val fis = new FileInputStream(f)
val filePath = new Path(f)
val fis = filePath.getFileSystem(sc.hadoopConfiguration).open(filePath)
Copy link
Member

Choose a reason for hiding this comment

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

This code can do many things. Please see my above comment. (#32184 (comment))

logInfo(s"Creating Fair Scheduler pools from $f")
Some((fis, f))
}.getOrElse {
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -205,7 +205,7 @@ private[spark] class TaskSchedulerImpl(
case SchedulingMode.FIFO =>
new FIFOSchedulableBuilder(rootPool)
case SchedulingMode.FAIR =>
new FairSchedulableBuilder(rootPool, conf)
new FairSchedulableBuilder(rootPool, sc)
case _ =>
throw new IllegalArgumentException(s"Unsupported $SCHEDULER_MODE_PROPERTY: " +
s"$schedulingMode")
Expand Down
15 changes: 8 additions & 7 deletions core/src/test/scala/org/apache/spark/scheduler/PoolSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,7 @@ class PoolSuite extends SparkFunSuite with LocalSparkContext {
val taskScheduler = new TaskSchedulerImpl(sc)

val rootPool = new Pool("", FAIR, 0, 0)
val schedulableBuilder = new FairSchedulableBuilder(rootPool, sc.conf)
val schedulableBuilder = new FairSchedulableBuilder(rootPool, sc)
schedulableBuilder.buildPools()

// Ensure that the XML file was read in correctly.
Expand Down Expand Up @@ -185,9 +185,10 @@ class PoolSuite extends SparkFunSuite with LocalSparkContext {
val xmlPath = getClass.getClassLoader.getResource("fairscheduler-with-invalid-data.xml")
.getFile()
val conf = new SparkConf().set(SCHEDULER_ALLOCATION_FILE, xmlPath)
sc = new SparkContext(LOCAL, APP_NAME, conf)

val rootPool = new Pool("", FAIR, 0, 0)
val schedulableBuilder = new FairSchedulableBuilder(rootPool, conf)
val schedulableBuilder = new FairSchedulableBuilder(rootPool, sc)
schedulableBuilder.buildPools()

verifyPool(rootPool, schedulableBuilder.DEFAULT_POOL_NAME, 0, 1, FIFO)
Expand Down Expand Up @@ -239,7 +240,7 @@ class PoolSuite extends SparkFunSuite with LocalSparkContext {
val taskScheduler = new TaskSchedulerImpl(sc)

val rootPool = new Pool("", SchedulingMode.FAIR, initMinShare = 0, initWeight = 0)
val schedulableBuilder = new FairSchedulableBuilder(rootPool, sc.conf)
val schedulableBuilder = new FairSchedulableBuilder(rootPool, sc)
schedulableBuilder.buildPools()

// Submit a new task set manager with pool properties set to null. This should result
Expand Down Expand Up @@ -267,7 +268,7 @@ class PoolSuite extends SparkFunSuite with LocalSparkContext {
val taskScheduler = new TaskSchedulerImpl(sc)

val rootPool = new Pool("", SchedulingMode.FAIR, initMinShare = 0, initWeight = 0)
val schedulableBuilder = new FairSchedulableBuilder(rootPool, sc.conf)
val schedulableBuilder = new FairSchedulableBuilder(rootPool, sc)
schedulableBuilder.buildPools()

assert(rootPool.getSchedulableByName(TEST_POOL) === null)
Expand Down Expand Up @@ -302,7 +303,7 @@ class PoolSuite extends SparkFunSuite with LocalSparkContext {
sc = new SparkContext(LOCAL, APP_NAME, conf)

val rootPool = new Pool("", SchedulingMode.FAIR, 0, 0)
val schedulableBuilder = new FairSchedulableBuilder(rootPool, sc.conf)
val schedulableBuilder = new FairSchedulableBuilder(rootPool, sc)
schedulableBuilder.buildPools()

verifyPool(rootPool, schedulableBuilder.DEFAULT_POOL_NAME, 0, 1, FIFO)
Expand All @@ -317,7 +318,7 @@ class PoolSuite extends SparkFunSuite with LocalSparkContext {
sc = new SparkContext(LOCAL, APP_NAME, conf)

val rootPool = new Pool("", SchedulingMode.FAIR, 0, 0)
val schedulableBuilder = new FairSchedulableBuilder(rootPool, sc.conf)
val schedulableBuilder = new FairSchedulableBuilder(rootPool, sc)
schedulableBuilder.buildPools()

verifyPool(rootPool, schedulableBuilder.DEFAULT_POOL_NAME, 0, 1, FIFO)
Expand All @@ -332,7 +333,7 @@ class PoolSuite extends SparkFunSuite with LocalSparkContext {
sc = new SparkContext(LOCAL, APP_NAME, conf)

val rootPool = new Pool("", SchedulingMode.FAIR, 0, 0)
val schedulableBuilder = new FairSchedulableBuilder(rootPool, sc.conf)
val schedulableBuilder = new FairSchedulableBuilder(rootPool, sc)
intercept[FileNotFoundException] {
schedulableBuilder.buildPools()
}
Expand Down
1 change: 1 addition & 0 deletions docs/job-scheduling.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -256,6 +256,7 @@ and either putting a file named `fairscheduler.xml` on the classpath, or setting

{% highlight scala %}
conf.set("spark.scheduler.allocation.file", "/path/to/file")
conf.set("spark.scheduler.allocation.file", "hdfs:///path/to/file")
Copy link
Member

Choose a reason for hiding this comment

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

I think HDFS also supports reading the local file. Could you mention it in the above document? e.g., "The file path can either be a local file path or HDFS file path."

{% endhighlight %}

The format of the XML file is simply a `<pool>` element for each pool, with different elements
Expand Down