Skip to content

Commit

Permalink
[SPARK-45753][CORE] Support spark.deploy.driverIdPattern
Browse files Browse the repository at this point in the history
### What changes were proposed in this pull request?

This PR aims to support `spark.deploy.driverIdPattern` for Apache Spark 4.0.0.

### Why are the changes needed?

This allows the users to be able to control driver ID pattern.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

Pass the CIs with the newly added test case.

### Was this patch authored or co-authored using generative AI tooling?

No.

Closes apache#43615 from dongjoon-hyun/SPARK-45753.

Authored-by: Dongjoon Hyun <dhyun@apple.com>
Signed-off-by: Dongjoon Hyun <dhyun@apple.com>
  • Loading branch information
dongjoon-hyun committed Nov 1, 2023
1 parent 253f358 commit 540d8d7
Show file tree
Hide file tree
Showing 3 changed files with 26 additions and 1 deletion.
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,8 @@ private[deploy] class Master(
private val forwardMessageThread =
ThreadUtils.newDaemonSingleThreadScheduledExecutor("master-forward-message-thread")

private val driverIdPattern = conf.get(DRIVER_ID_PATTERN)

// For application IDs
private def createDateFormat = new SimpleDateFormat("yyyyMMddHHmmss", Locale.US)

Expand Down Expand Up @@ -1174,7 +1176,7 @@ private[deploy] class Master(
}

private def newDriverId(submitDate: Date): String = {
val appId = "driver-%s-%04d".format(createDateFormat.format(submitDate), nextDriverNumber)
val appId = driverIdPattern.format(createDateFormat.format(submitDate), nextDriverNumber)
nextDriverNumber += 1
appId
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -82,4 +82,12 @@ private[spark] object Deploy {
.checkValue(_ > 0, "The maximum number of running drivers should be positive.")
.createWithDefault(Int.MaxValue)

val DRIVER_ID_PATTERN = ConfigBuilder("spark.deploy.driverIdPattern")
.doc("The pattern for driver ID generation based on Java `String.format` method. " +
"The default value is `driver-%s-%04d` which represents the existing driver id string " +
", e.g., `driver-20231031224459-0019`. Please be careful to generate unique IDs")
.version("4.0.0")
.stringConf
.checkValue(!_.format("20231101000000", 0).exists(_.isWhitespace), "Whitespace is not allowed.")
.createWithDefault("driver-%s-%04d")
}
Original file line number Diff line number Diff line change
Expand Up @@ -802,6 +802,7 @@ class MasterSuite extends SparkFunSuite
private val _waitingDrivers =
PrivateMethod[mutable.ArrayBuffer[DriverInfo]](Symbol("waitingDrivers"))
private val _state = PrivateMethod[RecoveryState.Value](Symbol("state"))
private val _newDriverId = PrivateMethod[String](Symbol("newDriverId"))

private val workerInfo = makeWorkerInfo(4096, 10)
private val workerInfos = Array(workerInfo, workerInfo, workerInfo)
Expand Down Expand Up @@ -1236,6 +1237,20 @@ class MasterSuite extends SparkFunSuite
private def getState(master: Master): RecoveryState.Value = {
master.invokePrivate(_state())
}

test("SPARK-45753: Support driver id pattern") {
val master = makeMaster(new SparkConf().set(DRIVER_ID_PATTERN, "my-driver-%2$05d"))
val submitDate = new Date()
assert(master.invokePrivate(_newDriverId(submitDate)) === "my-driver-00000")
assert(master.invokePrivate(_newDriverId(submitDate)) === "my-driver-00001")
}

test("SPARK-45753: Prevent invalid driver id patterns") {
val m = intercept[IllegalArgumentException] {
makeMaster(new SparkConf().set(DRIVER_ID_PATTERN, "my driver"))
}.getMessage
assert(m.contains("Whitespace is not allowed"))
}
}

private class FakeRecoveryModeFactory(conf: SparkConf, ser: serializer.Serializer)
Expand Down

0 comments on commit 540d8d7

Please sign in to comment.