This is an automated email from the ASF dual-hosted git repository.
pan3793 pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new b4bdb0c4fd73 Revert "[SPARK-59487][CORE] Add task attempt spans to
executor profiling"
b4bdb0c4fd73 is described below
commit b4bdb0c4fd73f79fc4a1a90e71a3b19ca0ad6b9e
Author: Zhen Wang <[email protected]>
AuthorDate: Fri Oct 9 11:42:25 2026 +0800
Revert "[SPARK-59487][CORE] Add task attempt spans to executor profiling"
### What changes were proposed in this pull request?
This reverts commit 87afa934866d97bbc3c6d1bacd91a4217c7587b7.
### Why are the changes needed?
As mentioned in
https://github.com/apache/spark/pull/58770#issuecomment-6003701164, this PR
broke branch-4.x because the version of ap-loader used in branch-4.x does not
yet support the span feature.
### Does this PR introduce _any_ user-facing change?
### How was this patch tested?
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #59284 from wForget/branch-4.x.
Authored-by: Zhen Wang <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>
---
connector/profiler/README.md | 27 --------------
.../org/apache/spark/profiler/ProfilerPlugin.scala | 42 +---------------------
.../scala/org/apache/spark/profiler/package.scala | 7 ----
3 files changed, 1 insertion(+), 75 deletions(-)
diff --git a/connector/profiler/README.md b/connector/profiler/README.md
index ff6fef20bcbd..04dc00f3985e 100644
--- a/connector/profiler/README.md
+++ b/connector/profiler/README.md
@@ -44,23 +44,6 @@ spark.plugins=ProfilerPlugin
Then enable the profiling in the configuration.
-### Task spans
-
-Set `spark.profiler.executor.taskSpan.enabled=true` to record each task
attempt as a
-`profiler.Span` event on profiled executors. Task spans are disabled by
default and
-require executor profiling to be enabled; executor sampling still applies.
-
-Spans are written to the same JFR file as the profiling samples.
-The span starts in the task start callback
-and ends in the success or failure callback, including result serialization on
-successful tasks. Task deserialization before the start callback is not
included.
-
-The span tag starts with `SparkTask`, followed by comma-separated `stageId`,
-`stageAttemptNumber`, `partitionId`, `taskAttemptId`, `attemptNumber`, and
`status`
-(`succeeded` or `failed`) key-value pairs. These tags allow profiling
-samples to be associated with individual task attempts, including retries and
-speculative attempts. Spans are only emitted while a JFR recording is active.
-
### Code profiling configuration
@@ -90,16 +73,6 @@ speculative attempts. Spans are only emitted while a JFR
recording is active.
</td>
<td>4.0.0</td>
</tr>
-<tr>
- <td><code>spark.profiler.executor.taskSpan.enabled</code></td>
- <td><code>false</code></td>
- <td>
- If true, record task attempt spans in the JFR output of executors selected
for profiling.
- Requires <code>spark.profiler.executor.enabled=true</code> and respects
- <code>spark.profiler.executor.fraction</code>.
- </td>
- <td>4.4.0</td>
-</tr>
<tr>
<td><code>spark.profiler.dfsDir</code></td>
<td>(none)</td>
diff --git
a/connector/profiler/src/main/scala/org/apache/spark/profiler/ProfilerPlugin.scala
b/connector/profiler/src/main/scala/org/apache/spark/profiler/ProfilerPlugin.scala
index 4198b846aab2..cb341aef3a0d 100644
---
a/connector/profiler/src/main/scala/org/apache/spark/profiler/ProfilerPlugin.scala
+++
b/connector/profiler/src/main/scala/org/apache/spark/profiler/ProfilerPlugin.scala
@@ -21,9 +21,7 @@ import java.util.{Map => JMap}
import scala.jdk.CollectionConverters._
import scala.util.Random
-import one.profiler.Span
-
-import org.apache.spark.{SparkConf, SparkContext, TaskContext,
TaskFailedReason}
+import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.api.plugin.{DriverPlugin, ExecutorPlugin,
PluginContext, SparkPlugin}
import org.apache.spark.internal.Logging
import org.apache.spark.internal.LogKeys.EXECUTOR_ID
@@ -72,16 +70,12 @@ class ProfilerExecutorPlugin extends ExecutorPlugin with
Logging {
private var profiler: SparkAsyncProfiler = _
private var executorProfilerEnabled: Boolean = _
private var executorProfilerFraction: Double = _
- private var taskSpanEnabled: Boolean = _
private val rand: Random = new Random(System.currentTimeMillis())
- // TaskContext is cleared before the task success and failure callbacks run.
- private val taskSpan = new ThreadLocal[(Long, String)]
override def init(ctx: PluginContext, extraConf: JMap[String, String]): Unit
= {
pluginCtx = ctx
sparkConf = ctx.conf()
executorProfilerEnabled = sparkConf.get(PROFILER_EXECUTOR_ENABLED)
- taskSpanEnabled = sparkConf.get(PROFILER_EXECUTOR_TASK_SPAN_ENABLED)
if (executorProfilerEnabled) {
executorProfilerFraction = sparkConf.get(PROFILER_EXECUTOR_FRACTION)
if (rand.nextInt(100) * 0.01 < executorProfilerFraction) {
@@ -94,40 +88,6 @@ class ProfilerExecutorPlugin extends ExecutorPlugin with
Logging {
Map.empty[String, String].asJava
}
- override def onTaskStart(): Unit = {
- if (taskSpanEnabled && profiler != null) {
- taskSpan.remove()
- val startTime = Span.start()
- if (startTime != 0) {
- val context = TaskContext.get()
- val tag = s"SparkTask:stageId=${context.stageId()}," +
- s"stageAttemptNumber=${context.stageAttemptNumber()}," +
- s"partitionId=${context.partitionId()}," +
- s"taskAttemptId=${context.taskAttemptId()}," +
- s"attemptNumber=${context.attemptNumber()}"
- taskSpan.set((startTime, tag))
- }
- }
- }
-
- override def onTaskSucceeded(): Unit = {
- endTaskSpan("succeeded")
- }
-
- override def onTaskFailed(failureReason: TaskFailedReason): Unit = {
- endTaskSpan("failed")
- }
-
- private def endTaskSpan(status: String): Unit = {
- if (taskSpanEnabled && profiler != null) {
- val span = taskSpan.get()
- taskSpan.remove()
- if (span != null) {
- Span.end(span._1, s"${span._2},status=$status")
- }
- }
- }
-
override def shutdown(): Unit = {
logInfo("Executor profiler shutting down")
if (profiler != null) {
diff --git
a/connector/profiler/src/main/scala/org/apache/spark/profiler/package.scala
b/connector/profiler/src/main/scala/org/apache/spark/profiler/package.scala
index 63bb080adc1e..30c2808b3d58 100644
--- a/connector/profiler/src/main/scala/org/apache/spark/profiler/package.scala
+++ b/connector/profiler/src/main/scala/org/apache/spark/profiler/package.scala
@@ -45,13 +45,6 @@ package object profiler {
"Fraction of executors to profile must be in [0,1]")
.createWithDefault(0.1)
- private[profiler] val PROFILER_EXECUTOR_TASK_SPAN_ENABLED =
- ConfigBuilder("spark.profiler.executor.taskSpan.enabled")
- .doc("Record task attempt spans in the JFR output of executors selected
for profiling.")
- .version("4.4.0")
- .booleanConf
- .createWithDefault(false)
-
private[profiler] val PROFILER_DFS_DIR =
ConfigBuilder("spark.profiler.dfsDir")
.doc("HDFS compatible file-system path to where the profiler will write
output jfr files.")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
