This is an automated email from the ASF dual-hosted git repository.
zaynt4606 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new 5c116df97b [CELEBORN-2415] Add Celeborn Spark UI extension
5c116df97b is described below
commit 5c116df97be78538f8c98140f8b833ca72efda3f
Author: yew1eb <[email protected]>
AuthorDate: Fri Oct 9 22:06:17 2026 +0800
[CELEBORN-2415] Add Celeborn Spark UI extension
### What changes were proposed in this pull request?
Add a new module `celeborn-client-spark-3-ui` — a Spark UI extension that
renders a **"Celeborn" tab** in both the live Spark UI and the History Server
(enabled via `spark.plugins=org.apache.spark.shuffle.celeborn.CelebornPlugin`,
supports Spark 3.x and 4.x).
This is an **initial version** that lays the foundation; it intentionally
starts minimal. The page shows:
1. **Summary** — shuffle write/read bytes, time, speed, and shuffle-vs-task
duration ratio, aggregated from Spark's built-in `TaskMetrics`
2. **Celeborn Properties** — a collapsible table of `spark.celeborn.*`
configs
The skeleton is: `CelebornPlugin` (live UI) / `CelebornHistoryServerPlugin`
(SHS, via `AppHistoryServerPlugin` SPI) → listener aggregates into Spark
`KVStore` → `CelebornStatusStore` → `CelebornUITab` renders; a
`javax`/`jakarta` servlet bridge covers Spark 3.x and 4.x. This
plugin/store/render pipeline is designed so that fine-grained Celeborn client
metrics can be added as collapsible sections in follow-up PRs:
- **Shuffle Write Times** — serialize / copy / queue wait / compress /
queue stall / inflight wait / drain wait / max push RTT / slow push
- **Shuffle Read Times** — chunk wait / decompress / retry wait / max chunk
RTT / slow chunk
- **Shuffle Write Servers** — per-worker push bytes, push count, RTT,
soft/hard splits, congested counts, last failure reason
- **Shuffle Read Servers** — per-worker chunk count, read bytes, total/max
RTT
- **Shuffle Assignments** — partition-to-worker slot assignments
### Why are the changes needed?
Spark UI already shows stage/task-level shuffle metrics, but nothing
`Celeborn-specific`: which workers served the data, push/chunk latencies,
congestion and split events are invisible there, while Celeborn's own metrics
sit in cluster-level Grafana dashboards that cannot be attributed back to a
specific application. This tab closes that gap by putting per-application
Celeborn shuffle information inside the Spark UI and History Server, alongside
the job's stages and tasks.
### Does this PR introduce any user-facing change?
Yes.
Setting `spark.plugins=org.apache.spark.shuffle.celeborn.CelebornPlugin`
(with the extension jar on the classpath) adds a "Celeborn" tab to the Spark UI
and History Server. Disabled by default; no effect otherwise.
### Does this PR resolve a correctness bug?
No.
### How was this patch tested?
Manual end-to-end verification:
<img width="2584" height="1352" alt="image"
src="https://github.com/user-attachments/assets/202eb4b7-389c-4dab-a2e4-799971594b92";
/>
Closes #3793 from yew1eb/celeborn-ui-mvp.
Authored-by: yew1eb <[email protected]>
Signed-off-by: zhengtao <[email protected]>
AI-Contributed/Feature: 0/716
AI-Contributed/UT: 0/170
---
client-spark/spark-3-shaded/pom.xml | 5 +
client-spark/spark-3-ui/pom.xml | 91 +++++++++++
.../org.apache.spark.status.AppHistoryServerPlugin | 16 ++
.../shuffle/celeborn/ui/SparkServletBridge.scala | 24 +++
.../shuffle/celeborn/ui/SparkServletBridge.scala | 24 +++
.../celeborn/CelebornHistoryServerPlugin.scala | 50 ++++++
.../spark/shuffle/celeborn/CelebornListener.scala | 145 ++++++++++++++++++
.../spark/shuffle/celeborn/CelebornPlugin.scala | 65 ++++++++
.../shuffle/celeborn/CelebornStatusStore.scala | 86 +++++++++++
.../shuffle/celeborn/ui/CelebornShufflePage.scala | 130 ++++++++++++++++
.../spark/shuffle/celeborn/ui/CelebornUITab.scala | 30 ++++
.../shuffle/celeborn/CelebornListenerSuite.scala | 170 +++++++++++++++++++++
client-spark/spark-4-shaded/pom.xml | 5 +
pom.xml | 18 +++
project/CelebornBuild.scala | 27 +++-
15 files changed, 884 insertions(+), 2 deletions(-)
diff --git a/client-spark/spark-3-shaded/pom.xml
b/client-spark/spark-3-shaded/pom.xml
index bc8c2065e2..8ad454779d 100644
--- a/client-spark/spark-3-shaded/pom.xml
+++ b/client-spark/spark-3-shaded/pom.xml
@@ -34,6 +34,11 @@
<artifactId>celeborn-client-spark-3_${scala.binary.version}</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.celeborn</groupId>
+
<artifactId>celeborn-client-spark-3-ui_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
</dependencies>
<build>
diff --git a/client-spark/spark-3-ui/pom.xml b/client-spark/spark-3-ui/pom.xml
new file mode 100644
index 0000000000..d11f0a29a6
--- /dev/null
+++ b/client-spark/spark-3-ui/pom.xml
@@ -0,0 +1,91 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You under the Apache License, Version 2.0
+ ~ (the "License"); you may not use this file except in compliance with
+ ~ the License. You may obtain a copy of the License at
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing, software
+ ~ distributed under the License is distributed on an "AS IS" BASIS,
+ ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ ~ See the License for the specific language governing permissions and
+ ~ limitations under the License.
+ -->
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.celeborn</groupId>
+ <artifactId>celeborn-parent_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <relativePath>../../pom.xml</relativePath>
+ </parent>
+
+ <artifactId>celeborn-client-spark-3-ui_${scala.binary.version}</artifactId>
+ <packaging>jar</packaging>
+ <name>Celeborn Spark UI Plugin</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.celeborn</groupId>
+ <artifactId>celeborn-client-spark-3_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-core_${scala.binary.version}</artifactId>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>org.xerial.snappy</groupId>
+ <artifactId>snappy-java</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>javax.servlet</groupId>
+ <artifactId>javax.servlet-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>jakarta.servlet</groupId>
+ <artifactId>jakarta.servlet-api</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>net.alchim31.maven</groupId>
+ <artifactId>scala-maven-plugin</artifactId>
+ </plugin>
+ <plugin>
+ <groupId>com.diffplug.spotless</groupId>
+ <artifactId>spotless-maven-plugin</artifactId>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>build-helper-maven-plugin</artifactId>
+ <version>3.6.0</version>
+ <executions>
+ <execution>
+ <id>add-servlet-source</id>
+ <goals>
+ <goal>add-source</goal>
+ </goals>
+ <phase>generate-sources</phase>
+ <configuration>
+ <sources>
+ <source>src/main/${servlet.source.dir}</source>
+ </sources>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+</project>
diff --git
a/client-spark/spark-3-ui/src/main/resources/META-INF/services/org.apache.spark.status.AppHistoryServerPlugin
b/client-spark/spark-3-ui/src/main/resources/META-INF/services/org.apache.spark.status.AppHistoryServerPlugin
new file mode 100644
index 0000000000..9710c39a11
--- /dev/null
+++
b/client-spark/spark-3-ui/src/main/resources/META-INF/services/org.apache.spark.status.AppHistoryServerPlugin
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+org.apache.spark.shuffle.celeborn.CelebornHistoryServerPlugin
diff --git
a/client-spark/spark-3-ui/src/main/scala-spark3/org/apache/spark/shuffle/celeborn/ui/SparkServletBridge.scala
b/client-spark/spark-3-ui/src/main/scala-spark3/org/apache/spark/shuffle/celeborn/ui/SparkServletBridge.scala
new file mode 100644
index 0000000000..8b397d566c
--- /dev/null
+++
b/client-spark/spark-3-ui/src/main/scala-spark3/org/apache/spark/shuffle/celeborn/ui/SparkServletBridge.scala
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle.celeborn.ui
+
+// Spark 3.x uses javax.servlet. This type alias bridges the API difference
+// so a single CelebornShufflePage.scala compiles for both Spark 3 and Spark 4.
+object SparkServletBridge {
+ type HttpServletRequest = javax.servlet.http.HttpServletRequest
+}
diff --git
a/client-spark/spark-3-ui/src/main/scala-spark4/org/apache/spark/shuffle/celeborn/ui/SparkServletBridge.scala
b/client-spark/spark-3-ui/src/main/scala-spark4/org/apache/spark/shuffle/celeborn/ui/SparkServletBridge.scala
new file mode 100644
index 0000000000..8b52a5e1ed
--- /dev/null
+++
b/client-spark/spark-3-ui/src/main/scala-spark4/org/apache/spark/shuffle/celeborn/ui/SparkServletBridge.scala
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle.celeborn.ui
+
+// Spark 4.x uses jakarta.servlet. This type alias bridges the API difference
+// so a single CelebornShufflePage.scala compiles for both Spark 3 and Spark 4.
+object SparkServletBridge {
+ type HttpServletRequest = jakarta.servlet.http.HttpServletRequest
+}
diff --git
a/client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornHistoryServerPlugin.scala
b/client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornHistoryServerPlugin.scala
new file mode 100644
index 0000000000..3634f2701e
--- /dev/null
+++
b/client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornHistoryServerPlugin.scala
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle.celeborn
+
+import org.apache.spark.SparkConf
+import org.apache.spark.scheduler.SparkListener
+import org.apache.spark.shuffle.celeborn.ui.CelebornUITab
+import org.apache.spark.status.{AppHistoryServerPlugin, ElementTrackingStore}
+import org.apache.spark.ui.SparkUI
+
+/** Registered via SPI at
META-INF/services/org.apache.spark.status.AppHistoryServerPlugin. */
+class CelebornHistoryServerPlugin extends AppHistoryServerPlugin {
+
+ override def createListeners(
+ conf: SparkConf,
+ store: ElementTrackingStore): Seq[SparkListener] = {
+ val listener = new CelebornListener(store, conf, requirePluginOptIn = true)
+ // Persist the final accumulated values when replay finishes, covering logs
+ // whose last events fall inside the throttle interval and that may lack an
+ // ApplicationEnd event.
+ store.onFlush(listener.flush())
+ Seq(listener)
+ }
+
+ override def setupUI(ui: SparkUI): Unit = {
+ val statusStore = new CelebornStatusStore(ui.store.store)
+ // Only attach the tab for applications that opted in via spark.plugins or
+ // spark.plugins.defaultList.
+ if (statusStore.extensionEnabled()) {
+ new CelebornUITab(statusStore, ui)
+ }
+ }
+
+ override def displayOrder: Int = 1
+}
diff --git
a/client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornListener.scala
b/client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornListener.scala
new file mode 100644
index 0000000000..1348ee2f8f
--- /dev/null
+++
b/client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornListener.scala
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle.celeborn
+
+import java.util.concurrent.atomic.AtomicLong
+
+import org.apache.spark.SparkConf
+import org.apache.spark.internal.Logging
+import org.apache.spark.scheduler._
+import org.apache.spark.util.Utils
+import org.apache.spark.util.kvstore.KVStore
+
+/**
+ * Collects Celeborn shuffle metrics into the Spark KVStore for live UI and
+ * HistoryServer replay.
+ *
+ * When `requirePluginOptIn` is true (History Server replay), collection stays
+ * disabled until the application's recorded `spark.plugins` or
+ * `spark.plugins.defaultList` contains [[CelebornPlugin]], and an enable
marker
+ * is persisted so `setupUI` can decide whether to attach the Celeborn tab.
+ */
+private[celeborn] class CelebornListener(
+ val kvstore: KVStore,
+ val conf: SparkConf,
+ requirePluginOptIn: Boolean = false)
+ extends SparkListener with Logging {
+
+ @volatile private var pluginEnabled = !requirePluginOptIn
+
+ private val totalWriteBytes = new AtomicLong(0L)
+ private val totalWriteTimeMs = new AtomicLong(0L)
+ private val totalReadBytes = new AtomicLong(0L)
+ private val totalFetchWaitTimeMs = new AtomicLong(0L)
+ private val totalTaskDurationMs = new AtomicLong(0L)
+
+ private val lastUpdateTimestamp = new AtomicLong(-1L)
+ private val updateIntervalMillis = 5000L
+
+ def register(sc: org.apache.spark.SparkContext): Unit = {
+ sc.addSparkListener(this)
+ logInfo("CelebornListener registered successfully")
+ }
+
+ override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = {
+ if (!pluginEnabled) {
+ return
+ }
+ Option(taskEnd.taskMetrics).foreach { metrics =>
+ totalWriteBytes.addAndGet(metrics.shuffleWriteMetrics.bytesWritten)
+ // writeTime is in nanoseconds; normalize to ms.
+ totalWriteTimeMs.addAndGet(metrics.shuffleWriteMetrics.writeTime /
1000000L)
+ totalReadBytes.addAndGet(metrics.shuffleReadMetrics.totalBytesRead)
+ totalFetchWaitTimeMs.addAndGet(metrics.shuffleReadMetrics.fetchWaitTime)
+ totalTaskDurationMs.addAndGet(taskEnd.taskInfo.duration)
+ }
+ mayUpdate()
+ }
+
+ override def onJobEnd(jobEnd: SparkListenerJobEnd): Unit = {
+ // Flush per-job so that tasks finishing within the throttle interval of
the
+ // previous flush are not lost when the application stays alive but idle.
+ mayUpdate(force = true)
+ }
+
+ override def onEnvironmentUpdate(environmentUpdate:
SparkListenerEnvironmentUpdate): Unit = {
+ val sparkProps = environmentUpdate.environmentDetails
+ .getOrElse("Spark Properties", Seq.empty)
+ if (!pluginEnabled) {
+ val pluginClass = classOf[CelebornPlugin].getName
+ // Spark loads plugins from both keys: `spark.plugins.defaultList`
allows a
+ // default plugin list in the config file that `spark.plugins` does not
overwrite.
+ val optedIn = sparkProps.exists { case (k, v) =>
+ (k == "spark.plugins" || k == "spark.plugins.defaultList") &&
+ v.split(",").exists(_.trim == pluginClass)
+ }
+ if (optedIn) {
+ pluginEnabled = true
+ kvstore.write(new CelebornExtensionEnabledUIData())
+ } else {
+ return
+ }
+ }
+ val celebornProps = Utils.redact(
+ conf,
+ sparkProps.filter { case (k, _) => k.startsWith("spark.celeborn.") })
+ .sortBy(_._1)
+ if (celebornProps.nonEmpty) {
+ kvstore.write(new CelebornPropertiesUIData(celebornProps.toList))
+ }
+ }
+
+ override def onApplicationEnd(applicationEnd: SparkListenerApplicationEnd):
Unit = {
+ mayUpdate(force = true)
+ logInfo("CelebornListener: application ended, final flush completed")
+ }
+
+ /** Flushes the current aggregations immediately, bypassing the throttle. */
+ def flush(): Unit = {
+ lastUpdateTimestamp.set(System.currentTimeMillis())
+ flushAggregations()
+ }
+
+ private def mayUpdate(force: Boolean = false): Unit = {
+ val now = System.currentTimeMillis()
+ val last = lastUpdateTimestamp.get()
+ if (!force && (last != -1L && (now - last) < updateIntervalMillis)) {
+ return
+ }
+ if (lastUpdateTimestamp.compareAndSet(last, now) || force) {
+ flushAggregations()
+ }
+ }
+
+ private def flushAggregations(): Unit = {
+ if (!pluginEnabled) {
+ return
+ }
+ try {
+ kvstore.write(AggregatedTaskInfoUIData(
+ totalWriteBytes.get(),
+ totalWriteTimeMs.get(),
+ totalReadBytes.get(),
+ totalFetchWaitTimeMs.get(),
+ totalTaskDurationMs.get()))
+ } catch {
+ case e: Exception =>
+ logWarning("Failed to flush CelebornListener aggregations", e)
+ }
+ }
+}
diff --git
a/client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornPlugin.scala
b/client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornPlugin.scala
new file mode 100644
index 0000000000..d0e0caae50
--- /dev/null
+++
b/client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornPlugin.scala
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle.celeborn
+
+import java.util
+
+import org.apache.spark.SparkContext
+import org.apache.spark.api.plugin.{DriverPlugin, PluginContext, SparkPlugin}
+import org.apache.spark.internal.Logging
+import org.apache.spark.shuffle.celeborn.ui.CelebornUITab
+
+/**
+ * SparkPlugin entry point for the Celeborn UI extension.
+ *
+ * Configure via:
+ * {{{
+ * spark.plugins=org.apache.spark.shuffle.celeborn.CelebornPlugin
+ * }}}
+ */
+class CelebornPlugin extends SparkPlugin {
+
+ override def driverPlugin(): DriverPlugin = new CelebornDriverPlugin()
+
+ override def executorPlugin(): org.apache.spark.api.plugin.ExecutorPlugin =
null
+}
+
+private class CelebornDriverPlugin extends DriverPlugin with Logging {
+
+ private var sc: SparkContext = _
+
+ override def init(
+ sc: SparkContext,
+ ctx: PluginContext): util.Map[String, String] = {
+ logInfo("Initializing CelebornDriverPlugin...")
+ this.sc = sc
+ val kvStore = sc.statusStore.store
+ new CelebornListener(kvStore, sc.conf).register(sc)
+ java.util.Collections.emptyMap[String, String]()
+ }
+
+ override def registerMetrics(
+ appId: String,
+ ctx: PluginContext): Unit = {
+ sc.ui.foreach { ui =>
+ new CelebornUITab(new CelebornStatusStore(ui.store.store), ui)
+ }
+ }
+
+ override def shutdown(): Unit = {}
+}
diff --git
a/client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornStatusStore.scala
b/client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornStatusStore.scala
new file mode 100644
index 0000000000..9a8ffe5cf1
--- /dev/null
+++
b/client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornStatusStore.scala
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle.celeborn
+
+import com.fasterxml.jackson.annotation.JsonIgnore
+import org.apache.spark.util.kvstore.{KVIndex, KVStore}
+
+private[celeborn] case class AggregatedTaskInfoUIData(
+ shuffleWriteBytes: Long,
+ shuffleWriteTimeMs: Long,
+ shuffleReadBytes: Long,
+ shuffleFetchWaitTimeMs: Long,
+ taskDurationMs: Long) {
+
+ @JsonIgnore
+ @KVIndex
+ def id: String = classOf[AggregatedTaskInfoUIData].getName
+}
+
+private[celeborn] class CelebornPropertiesUIData(
+ val info: Seq[(String, String)]) {
+
+ @JsonIgnore
+ @KVIndex
+ def id: String = classOf[CelebornPropertiesUIData].getName
+}
+
+/**
+ * Marker persisted during History Server replay when the application's
recorded
+ * `spark.plugins` or `spark.plugins.defaultList` contains [[CelebornPlugin]].
+ * `setupUI` runs on a fresh plugin
+ * instance (FsHistoryProvider loads plugins again after replay), so the opt-in
+ * state has to be passed through the KVStore rather than in memory.
+ */
+private[celeborn] class CelebornExtensionEnabledUIData {
+
+ @JsonIgnore
+ @KVIndex
+ def id: String = classOf[CelebornExtensionEnabledUIData].getName
+}
+
+private[celeborn] class CelebornStatusStore(store: KVStore) {
+
+ def aggregatedTaskInfo(): AggregatedTaskInfoUIData = {
+ val kClass = classOf[AggregatedTaskInfoUIData]
+ try {
+ store.read(kClass, kClass.getName)
+ } catch {
+ case _: NoSuchElementException => AggregatedTaskInfoUIData(0L, 0L, 0L,
0L, 0L)
+ }
+ }
+
+ def celebornProperties(): CelebornPropertiesUIData = {
+ val kClass = classOf[CelebornPropertiesUIData]
+ try {
+ store.read(kClass, kClass.getName)
+ } catch {
+ case _: NoSuchElementException => new CelebornPropertiesUIData(Seq.empty)
+ }
+ }
+
+ def extensionEnabled(): Boolean = {
+ val kClass = classOf[CelebornExtensionEnabledUIData]
+ try {
+ store.read(kClass, kClass.getName)
+ true
+ } catch {
+ case _: NoSuchElementException => false
+ }
+ }
+}
diff --git
a/client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/ui/CelebornShufflePage.scala
b/client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/ui/CelebornShufflePage.scala
new file mode 100644
index 0000000000..4a57652fcf
--- /dev/null
+++
b/client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/ui/CelebornShufflePage.scala
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle.celeborn.ui
+
+import scala.xml.Node
+
+import org.apache.spark.internal.Logging
+import
org.apache.spark.shuffle.celeborn.ui.SparkServletBridge.HttpServletRequest
+import org.apache.spark.ui.{UIUtils, WebUIPage}
+import org.apache.spark.util.Utils
+
+private[celeborn] class CelebornShufflePage(parent: CelebornUITab)
+ extends WebUIPage("") with Logging {
+
+ private val store = parent.store
+
+ override def render(request: HttpServletRequest): Seq[Node] = {
+ try {
+ renderBody(request)
+ } catch {
+ case e: Throwable =>
+ logError("Failed to render Celeborn Shuffle page", e)
+ val errorContent =
+ <div class="row-fluid">
+ <div class="span12">
+ <h4>
+ <strong>Celeborn Shuffle</strong>
+ </h4>
+ <div class="alert alert-error">
+ <pre>Failed to render the Celeborn page: {e.getMessage}</pre>
+ </div>
+ </div>
+ </div>
+ UIUtils.headerSparkPage(request, "Celeborn Shuffle", errorContent,
parent)
+ }
+ }
+
+ private def renderBody(request: HttpServletRequest): Seq[Node] = {
+ val taskInfo = store.aggregatedTaskInfo()
+ val properties = store.celebornProperties()
+
+ val writeBytes = taskInfo.shuffleWriteBytes
+ val readBytes = taskInfo.shuffleReadBytes
+ val writeMs = taskInfo.shuffleWriteTimeMs
+ val readMs = taskInfo.shuffleFetchWaitTimeMs
+ val durationMs = taskInfo.taskDurationMs
+
+ def mbps(bytes: Long, ms: Long): String =
+ if (ms <= 0) "N/A" else f"${bytes.toDouble / 1000.0 / 1000.0 /
(ms.toDouble / 1000.0)}%.2f"
+ def pct(part: Long, total: Long): String =
+ if (total <= 0) "N/A" else f"${part.toDouble * 100.0 /
total.toDouble}%.1f%%"
+
+ val summary =
+ <div>
+ <ul class="list-unstyled">
+ <li>
+ <strong>Shuffle Write: </strong>
+ {
+ s"${Utils.bytesToString(writeBytes)} | Time: ${UIUtils.formatDuration(
+ writeMs)} | Speed: ${mbps(writeBytes, writeMs)} MB/s"
+ }
+ </li>
+ <li>
+ <strong>Shuffle Read: </strong>
+ {
+ s"${Utils.bytesToString(readBytes)} | Time: ${UIUtils.formatDuration(
+ readMs)} | Speed: ${mbps(readBytes, readMs)} MB/s"
+ }
+ </li>
+ <li>
+ <strong>Shuffle Duration (write+read) / Task Duration: </strong>
+ {
+ s"${pct(writeMs + readMs, durationMs)} (Write ${pct(
+ writeMs,
+ durationMs)}, Read ${pct(readMs, durationMs)})"
+ }
+ </li>
+ </ul>
+ </div>
+
+ val propertiesTable = UIUtils.listingTable(
+ propertyHeader,
+ propertyRow,
+ properties.info,
+ fixedWidth = true,
+ headerClasses = headerClasses)
+
+ val content =
+ <span>
+ {summary}
+ <span class="collapse-aggregated-celebornProperties collapse-table"
+ onClick="collapseTable('collapse-aggregated-celebornProperties',
+ 'aggregated-celebornProperties')">
+ <h4>
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a>Celeborn Properties</a>
+ </h4>
+ </span>
+ <div class="aggregated-celebornProperties collapsible-table">
+ {propertiesTable}
+ </div>
+ </span>
+
+ UIUtils.headerSparkPage(request, "Celeborn Shuffle", content, parent)
+ }
+
+ private def propertyHeader: Seq[String] = Seq("Name", "Value")
+ private def headerClasses: Seq[String] = Seq("sorttable_alpha",
"sorttable_alpha")
+
+ private def propertyRow(kv: (String, String)): Seq[Node] =
+ <tr>
+ <td>{kv._1}</td>
+ <td>{kv._2}</td>
+ </tr>
+}
diff --git
a/client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/ui/CelebornUITab.scala
b/client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/ui/CelebornUITab.scala
new file mode 100644
index 0000000000..d9a135181d
--- /dev/null
+++
b/client-spark/spark-3-ui/src/main/scala/org/apache/spark/shuffle/celeborn/ui/CelebornUITab.scala
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle.celeborn.ui
+
+import org.apache.spark.shuffle.celeborn.CelebornStatusStore
+import org.apache.spark.ui.{SparkUI, SparkUITab}
+
+private[celeborn] class CelebornUITab(val store: CelebornStatusStore, sparkUI:
SparkUI)
+ extends SparkUITab(sparkUI, "celeborn") {
+
+ override val name: String = "Celeborn"
+
+ attachPage(new CelebornShufflePage(this))
+ sparkUI.attachTab(this)
+}
diff --git
a/client-spark/spark-3-ui/src/test/scala/org/apache/spark/shuffle/celeborn/CelebornListenerSuite.scala
b/client-spark/spark-3-ui/src/test/scala/org/apache/spark/shuffle/celeborn/CelebornListenerSuite.scala
new file mode 100644
index 0000000000..ed9ff78ad4
--- /dev/null
+++
b/client-spark/spark-3-ui/src/test/scala/org/apache/spark/shuffle/celeborn/CelebornListenerSuite.scala
@@ -0,0 +1,170 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle.celeborn
+
+import org.apache.spark.{SparkConf, Success, TaskState}
+import org.apache.spark.executor.TaskMetrics
+import org.apache.spark.internal.config.Status.ASYNC_TRACKING_ENABLED
+import org.apache.spark.scheduler.{JobSucceeded,
SparkListenerEnvironmentUpdate, SparkListenerJobEnd, SparkListenerTaskEnd,
TaskInfo, TaskLocality}
+import org.apache.spark.status.ElementTrackingStore
+import org.apache.spark.util.Utils
+import org.apache.spark.util.kvstore.InMemoryStore
+import org.junit.Assert.{assertEquals, assertFalse, assertTrue}
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.junit.runners.JUnit4
+
+@RunWith(classOf[JUnit4])
+class CelebornListenerSuite {
+
+ private val pluginClass = classOf[CelebornPlugin].getName
+
+ private def newTaskEnd(
+ writeBytes: Long,
+ writeTimeMs: Long,
+ fetchWaitMs: Long,
+ durationMs: Long): SparkListenerTaskEnd = {
+ val metrics = TaskMetrics.empty
+ metrics.shuffleWriteMetrics.incBytesWritten(writeBytes)
+ metrics.shuffleWriteMetrics.incWriteTime(writeTimeMs * 1000000L)
+ metrics.shuffleReadMetrics.incFetchWaitTime(fetchWaitMs)
+ val launchTime = 1000L
+ val info = new TaskInfo(
+ 0L,
+ 0,
+ 0,
+ launchTime,
+ "exec-0",
+ "localhost",
+ TaskLocality.PROCESS_LOCAL,
+ false)
+ info.markFinished(TaskState.FINISHED, launchTime + durationMs)
+ SparkListenerTaskEnd(0, 0, "ShuffleMapTask", Success, info, null, metrics)
+ }
+
+ private def envUpdate(sparkProps: (String, String)*):
SparkListenerEnvironmentUpdate = {
+ SparkListenerEnvironmentUpdate(Map("Spark Properties" -> sparkProps.toSeq))
+ }
+
+ @Test
+ def flushOnJobEndWithinThrottleInterval(): Unit = {
+ val store = new InMemoryStore()
+ val statusStore = new CelebornStatusStore(store)
+ val listener = new CelebornListener(store, new SparkConf())
+
+ listener.onTaskEnd(newTaskEnd(100L, 10L, 1L, 5L))
+ listener.onTaskEnd(newTaskEnd(200L, 20L, 2L, 6L))
+ // The second task lands inside the throttle window: only the first is
persisted so far.
+ assertEquals(100L, statusStore.aggregatedTaskInfo().shuffleWriteBytes)
+
+ listener.onJobEnd(SparkListenerJobEnd(0, 2000L, JobSucceeded))
+ assertEquals(300L, statusStore.aggregatedTaskInfo().shuffleWriteBytes)
+ assertEquals(30L, statusStore.aggregatedTaskInfo().shuffleWriteTimeMs)
+ assertEquals(3L, statusStore.aggregatedTaskInfo().shuffleFetchWaitTimeMs)
+ assertEquals(11L, statusStore.aggregatedTaskInfo().taskDurationMs)
+ }
+
+ @Test
+ def flushTriggerPersistsFinalValuesOnReplayClose(): Unit = {
+ val conf = new SparkConf().set(ASYNC_TRACKING_ENABLED, false)
+ val store = new InMemoryStore()
+ val tracking = new ElementTrackingStore(store, conf)
+ val statusStore = new CelebornStatusStore(tracking)
+ val listener = new CelebornListener(tracking, conf, requirePluginOptIn =
true)
+ tracking.onFlush(listener.flush())
+
+ listener.onEnvironmentUpdate(envUpdate("spark.plugins" -> pluginClass))
+ listener.onTaskEnd(newTaskEnd(100L, 10L, 1L, 5L))
+ listener.onTaskEnd(newTaskEnd(200L, 20L, 2L, 6L))
+ // Replay finishes within the throttle window without an ApplicationEnd
event.
+ tracking.close(false)
+
+ assertEquals(300L, statusStore.aggregatedTaskInfo().shuffleWriteBytes)
+ assertEquals(11L, statusStore.aggregatedTaskInfo().taskDurationMs)
+ }
+
+ @Test
+ def collectionGatedOnPluginOptIn(): Unit = {
+ // Without the plugin in the recorded spark.plugins: nothing is collected
or persisted.
+ val store1 = new InMemoryStore()
+ val statusStore1 = new CelebornStatusStore(store1)
+ val listener1 = new CelebornListener(store1, new SparkConf(),
requirePluginOptIn = true)
+ listener1.onEnvironmentUpdate(envUpdate(
+ "spark.celeborn.master.endpoints" -> "host:9097"))
+ listener1.onTaskEnd(newTaskEnd(100L, 10L, 1L, 5L))
+ listener1.onJobEnd(SparkListenerJobEnd(0, 2000L, JobSucceeded))
+ listener1.flush()
+ assertFalse(statusStore1.extensionEnabled())
+ assertEquals(0L, statusStore1.aggregatedTaskInfo().shuffleWriteBytes)
+ assertTrue(statusStore1.celebornProperties().info.isEmpty)
+
+ // With the plugin in the recorded spark.plugins: collection and tab
marker are enabled.
+ val store2 = new InMemoryStore()
+ val statusStore2 = new CelebornStatusStore(store2)
+ val listener2 = new CelebornListener(store2, new SparkConf(),
requirePluginOptIn = true)
+ listener2.onEnvironmentUpdate(envUpdate(
+ "spark.plugins" -> s"com.example.OtherPlugin,$pluginClass",
+ "spark.celeborn.master.endpoints" -> "host:9097"))
+ listener2.onTaskEnd(newTaskEnd(100L, 10L, 1L, 5L))
+ assertTrue(statusStore2.extensionEnabled())
+ assertEquals(100L, statusStore2.aggregatedTaskInfo().shuffleWriteBytes)
+ assertTrue(statusStore2.celebornProperties().info.exists(
+ _._1 == "spark.celeborn.master.endpoints"))
+ }
+
+ @Test
+ def pluginOptInViaDefaultList(): Unit = {
+ // The plugin can also be loaded via spark.plugins.defaultList (e.g. from
the
+ // Spark default config file); replay must recognize that as opt-in too.
+ val store = new InMemoryStore()
+ val statusStore = new CelebornStatusStore(store)
+ val listener = new CelebornListener(store, new SparkConf(),
requirePluginOptIn = true)
+ listener.onEnvironmentUpdate(envUpdate(
+ "spark.plugins.defaultList" -> s"com.example.OtherPlugin, $pluginClass"))
+ listener.onTaskEnd(newTaskEnd(100L, 10L, 1L, 5L))
+ assertTrue(statusStore.extensionEnabled())
+ assertEquals(100L, statusStore.aggregatedTaskInfo().shuffleWriteBytes)
+ }
+
+ @Test
+ def sensitivePropertiesAreRedacted(): Unit = {
+ val store = new InMemoryStore()
+ val statusStore = new CelebornStatusStore(store)
+ val listener = new CelebornListener(store, new SparkConf())
+
+ listener.onEnvironmentUpdate(envUpdate(
+ "spark.celeborn.ssl.rpc_service.trustStorePassword" ->
"top-secret-password",
+ "spark.celeborn.storage.oss.secret.key" -> "oss-secret-key",
+ "spark.celeborn.master.endpoints" -> "host1:9097,host2:9097",
+ "spark.executor.memory" -> "1g"))
+
+ val props = statusStore.celebornProperties().info.toMap
+
+ // Sensitive values are redacted, same as Spark's Environment tab.
+ assertEquals(
+ Utils.REDACTION_REPLACEMENT_TEXT,
+ props("spark.celeborn.ssl.rpc_service.trustStorePassword"))
+ assertEquals(Utils.REDACTION_REPLACEMENT_TEXT,
props("spark.celeborn.storage.oss.secret.key"))
+
+ // Ordinary Celeborn property remains visible.
+ assertEquals("host1:9097,host2:9097",
props("spark.celeborn.master.endpoints"))
+
+ // Non-Celeborn properties are filtered out.
+ assertFalse(props.contains("spark.executor.memory"))
+ }
+}
diff --git a/client-spark/spark-4-shaded/pom.xml
b/client-spark/spark-4-shaded/pom.xml
index 5e741d6017..bf0dbcb4e0 100644
--- a/client-spark/spark-4-shaded/pom.xml
+++ b/client-spark/spark-4-shaded/pom.xml
@@ -34,6 +34,11 @@
<artifactId>celeborn-client-spark-3_${scala.binary.version}</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.celeborn</groupId>
+
<artifactId>celeborn-client-spark-3-ui_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ </dependency>
</dependencies>
<build>
diff --git a/pom.xml b/pom.xml
index 0bab4860e4..878646622d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1516,6 +1516,7 @@
<module>client-spark/spark-3</module>
<module>client-spark/spark-3-columnar-common</module>
<module>client-spark/spark-3-columnar-shuffle</module>
+ <module>client-spark/spark-3-ui</module>
<module>client-spark/spark-3-shaded</module>
<module>tests/spark-it</module>
</modules>
@@ -1527,6 +1528,7 @@
<spark.version>3.0.3</spark.version>
<zstd-jni.version>1.4.4-3</zstd-jni.version>
<commons-lang3.version>3.17.0</commons-lang3.version>
+ <servlet.source.dir>scala-spark3</servlet.source.dir>
</properties>
</profile>
@@ -1537,6 +1539,7 @@
<module>client-spark/spark-3</module>
<module>client-spark/spark-3-columnar-common</module>
<module>client-spark/spark-3-columnar-shuffle</module>
+ <module>client-spark/spark-3-ui</module>
<module>client-spark/spark-3-shaded</module>
<module>tests/spark-it</module>
</modules>
@@ -1548,6 +1551,7 @@
<spark.version>3.1.3</spark.version>
<zstd-jni.version>1.4.8-1</zstd-jni.version>
<commons-lang3.version>3.17.0</commons-lang3.version>
+ <servlet.source.dir>scala-spark3</servlet.source.dir>
</properties>
</profile>
@@ -1558,6 +1562,7 @@
<module>client-spark/spark-3</module>
<module>client-spark/spark-3-columnar-common</module>
<module>client-spark/spark-3-columnar-shuffle</module>
+ <module>client-spark/spark-3-ui</module>
<module>client-spark/spark-3-shaded</module>
<module>tests/spark-it</module>
</modules>
@@ -1569,6 +1574,7 @@
<spark.version>3.2.4</spark.version>
<zstd-jni.version>1.5.0-4</zstd-jni.version>
<commons-lang3.version>3.17.0</commons-lang3.version>
+ <servlet.source.dir>scala-spark3</servlet.source.dir>
</properties>
</profile>
@@ -1579,6 +1585,7 @@
<module>client-spark/spark-3</module>
<module>client-spark/spark-3-columnar-common</module>
<module>client-spark/spark-3-columnar-shuffle</module>
+ <module>client-spark/spark-3-ui</module>
<module>client-spark/spark-3-shaded</module>
<module>tests/spark-it</module>
</modules>
@@ -1590,6 +1597,7 @@
<spark.version>3.3.4</spark.version>
<zstd-jni.version>1.5.2-1</zstd-jni.version>
<commons-lang3.version>3.17.0</commons-lang3.version>
+ <servlet.source.dir>scala-spark3</servlet.source.dir>
</properties>
</profile>
@@ -1600,6 +1608,7 @@
<module>client-spark/spark-3</module>
<module>client-spark/spark-3-columnar-common</module>
<module>client-spark/spark-3-columnar-shuffle</module>
+ <module>client-spark/spark-3-ui</module>
<module>client-spark/spark-3-shaded</module>
<module>tests/spark-it</module>
</modules>
@@ -1611,6 +1620,7 @@
<spark.version>3.4.4</spark.version>
<zstd-jni.version>1.5.2-5</zstd-jni.version>
<commons-lang3.version>3.17.0</commons-lang3.version>
+ <servlet.source.dir>scala-spark3</servlet.source.dir>
</properties>
</profile>
@@ -1621,6 +1631,7 @@
<module>client-spark/spark-3</module>
<module>client-spark/spark-3-columnar-common</module>
<module>client-spark/spark-3.5-columnar-shuffle</module>
+ <module>client-spark/spark-3-ui</module>
<module>client-spark/spark-3-shaded</module>
<module>tests/spark-it</module>
</modules>
@@ -1632,6 +1643,7 @@
<spark.version>3.5.8</spark.version>
<zstd-jni.version>1.5.5-4</zstd-jni.version>
<commons-lang3.version>3.17.0</commons-lang3.version>
+ <servlet.source.dir>scala-spark3</servlet.source.dir>
</properties>
</profile>
@@ -1642,6 +1654,7 @@
<module>client-spark/spark-3</module>
<module>client-spark/spark-3-columnar-common</module>
<module>client-spark/spark-4-columnar-shuffle</module>
+ <module>client-spark/spark-3-ui</module>
<module>client-spark/spark-4-shaded</module>
<module>tests/spark-it</module>
</modules>
@@ -1652,6 +1665,7 @@
<scala.binary.version>2.13</scala.binary.version>
<spark.version>4.0.3</spark.version>
<zstd-jni.version>1.5.6-9</zstd-jni.version>
+ <servlet.source.dir>scala-spark4</servlet.source.dir>
</properties>
</profile>
@@ -1662,6 +1676,7 @@
<module>client-spark/spark-3</module>
<module>client-spark/spark-3-columnar-common</module>
<module>client-spark/spark-4-columnar-shuffle</module>
+ <module>client-spark/spark-3-ui</module>
<module>client-spark/spark-4-shaded</module>
<module>tests/spark-it</module>
</modules>
@@ -1672,6 +1687,7 @@
<scala.binary.version>2.13</scala.binary.version>
<spark.version>4.1.2</spark.version>
<zstd-jni.version>1.5.7-6</zstd-jni.version>
+ <servlet.source.dir>scala-spark4</servlet.source.dir>
</properties>
<dependencyManagement>
<dependencies>
@@ -1691,6 +1707,7 @@
<module>client-spark/spark-3</module>
<module>client-spark/spark-3-columnar-common</module>
<module>client-spark/spark-4-columnar-shuffle</module>
+ <module>client-spark/spark-3-ui</module>
<module>client-spark/spark-4-shaded</module>
<module>tests/spark-it</module>
</modules>
@@ -1701,6 +1718,7 @@
<scala.binary.version>2.13</scala.binary.version>
<spark.version>4.2.0</spark.version>
<zstd-jni.version>1.5.7-7</zstd-jni.version>
+ <servlet.source.dir>scala-spark4</servlet.source.dir>
</properties>
<dependencyManagement>
<dependencies>
diff --git a/project/CelebornBuild.scala b/project/CelebornBuild.scala
index 6c9abcd735..1a00e98626 100644
--- a/project/CelebornBuild.scala
+++ b/project/CelebornBuild.scala
@@ -1008,6 +1008,7 @@ object Spark40 extends SparkClientProjects {
val zstdJniVersion = "1.5.6-9"
val scalaBinaryVersion = "2.13"
+ override val servletSourceDir: String = "scala-spark4"
override val sparkColumnarShuffleVersion: String = "4"
}
@@ -1025,6 +1026,7 @@ object Spark41 extends SparkClientProjects {
val zstdJniVersion = "1.5.7-6"
val scalaBinaryVersion = "2.13"
+ override val servletSourceDir: String = "scala-spark4"
override val sparkColumnarShuffleVersion: String = "4"
override val paranamerVersionOverride: Option[String] = Some("2.8.3")
}
@@ -1044,6 +1046,7 @@ object Spark42 extends SparkClientProjects {
val scalaBinaryVersion = "2.13"
override val lz4JavaGroup = "at.yawk.lz4"
+ override val servletSourceDir: String = "scala-spark4"
override val sparkColumnarShuffleVersion: String = "4"
override val paranamerVersionOverride: Option[String] = Some("2.8.3")
}
@@ -1066,15 +1069,19 @@ trait SparkClientProjects {
val includeColumnarShuffle: Boolean = true
+ // Mirrors Maven's `servlet.source.dir`: Spark 3.x uses javax.servlet,
+ // Spark 4.x uses jakarta.servlet.
+ val servletSourceDir: String = "scala-spark3"
+
def modules: Seq[Project] = {
- val seq = Seq(sparkCommon, sparkClient, sparkIt, sparkGroup,
sparkClientShade)
+ val seq = Seq(sparkCommon, sparkClient, sparkClientUi, sparkIt,
sparkGroup, sparkClientShade)
if (includeColumnarShuffle) seq ++ Seq(sparkColumnarCommon,
sparkColumnarShuffle) else seq
}
// for test only, don't use this group for any other projects
lazy val sparkGroup = {
val p = (project withId "celeborn-spark-group")
- .aggregate(sparkCommon, sparkClient, sparkIt)
+ .aggregate(sparkCommon, sparkClient, sparkClientUi, sparkIt)
if (includeColumnarShuffle) {
p.aggregate(sparkColumnarCommon, sparkColumnarShuffle)
} else {
@@ -1116,6 +1123,21 @@ trait SparkClientProjects {
)
}
+ def sparkClientUi: Project = {
+ Project("celeborn-client-spark-3-ui", file("client-spark/spark-3-ui"))
+ .dependsOn(sparkClient)
+ .settings (
+ commonSettings,
+ libraryDependencies ++= Seq(
+ "org.apache.spark" %% "spark-core" % sparkVersion % "provided",
+ Dependencies.javaxServletApi % "provided",
+ Dependencies.jakartaServletApi % "provided"
+ ) ++ commonUnitTestDependencies,
+ // Mirrors Maven's build-helper-maven-plugin `add-servlet-source`
execution.
+ Compile / unmanagedSourceDirectories += baseDirectory.value / "src" /
"main" / servletSourceDir
+ )
+ }
+
def sparkColumnarCommon: Project = {
Project("celeborn-spark-3-columnar-common",
file("client-spark/spark-3-columnar-common"))
// ref:
https://www.scala-sbt.org/1.x/docs/Multi-Project.html#Classpath+dependencies
@@ -1190,6 +1212,7 @@ trait SparkClientProjects {
def sparkClientShade: Project = {
var p = Project(sparkClientShadedProjectName,
file(sparkClientShadedProjectPath))
.dependsOn(sparkClient)
+ .dependsOn(sparkClientUi)
if (includeColumnarShuffle) {
p = p.dependsOn(sparkColumnarShuffle)
