You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2026/09/28 01:24:38 UTC
(spark) branch branch-4.x updated: [SPARK-59487][CORE] Add task attempt spans to executor profiling
This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon 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 87afa934866d [SPARK-59487][CORE] Add task attempt spans to executor profiling
87afa934866d is described below
commit 87afa934866d97bbc3c6d1bacd91a4217c7587b7
Author: Zhen Wang <wa...@apache.org>
AuthorDate: Mon Sep 28 10:24:11 2026 +0900
[SPARK-59487][CORE] Add task attempt spans to executor profiling
### What changes were proposed in this pull request?
Add task attempt spans to `ProfilerExecutorPlugin` using the Span API introduced by [async-profiler PR #1755](https://github.com/async-profiler/async-profiler/pull/1755).
Each span records a task attempt’s execution interval on its executor thread in the same JFR file as the profiling samples. Comma-separated tags include stage ID, stage attempt number, partition ID, task attempt ID, attempt number, and completion status.
Spans start in `onTaskStart` and end in the success or failure callback. Task metadata is retained in thread-local state because `TaskContext` is cleared before the end callbacks run.
### Why are the changes needed?
The profiling output currently lacks an explicit mapping between Spark task attempts and executor threads over time. Since an executor thread runs multiple tasks sequentially, thread identity alone cannot determine which task produced a profiling sample.
Task spans provide this mapping through their thread, execution interval, and task identifiers. This allows users to extract profiling samples for selected task attempts from the JFR file using the task-to-thread mapping and execution intervals recorded in the spans.
### Does this PR introduce _any_ user-facing change?
Yes. This PR adds `spark.profiler.executor.taskSpan.enabled`, disabled by default.
### How was this patch tested?
Test manually.
Spark conf:
```
spark.plugins=org.apache.spark.profiler.ProfilerPlugin;
spark.profiler.executor.enabled=true;
spark.profiler.executor.fraction=1;
spark.profiler.asyncProfiler.args=event=wall,interval=10ms,alloc=2m,lock=10ms,chunktime=10s;
spark.profiler.executor.taskSpan.enabled=true;
```
Test query:
```
SELECT
java_method('java.lang.Thread', 'sleep', CAST(60000 AS BIGINT)),
java_method('java.lang.Thread', 'sleep', CAST(30000 AS BIGINT)),
java_method('java.lang.Thread', 'sleep', CAST(10000 AS BIGINT))
FROM range(0, 1, 1, 1);
```
print span event `jfr print --events profiler.Span profile-exec-1.jfr`:
```
profiler.Span {
startTime = 07:28:50.478
duration = 1 m 41 s
tag = "SparkTask:stageId=0,stageAttemptNumber=0,partitionId=0,taskAttemptId=0,attemptNumber=0,status=succeeded"
eventThread = "Executor task launch worker for task 0.0 in stage 0.0 (TID 0)" (javaThreadId = 53)
}
```
get flame graph for task 0:
```
bin/jfrconv --wall --bci \
--tag 'SparkTask:stageId=0,stageAttemptNumber=0,partitionId=0,taskAttemptId=0,attemptNumber=0,status=succeeded' \
--title 'Task 0' \
profile-exec-1.jfr task-0.html
```
<img width="1900" height="808" alt="image" src="https://github.com/user-attachments/assets/35d0f8e4-85bc-4ce7-b778-7174b2533d62" />
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-6)
Closes #58770 from wForget/dev-profiler.
Authored-by: Zhen Wang <wa...@apache.org>
Signed-off-by: Hyukjin Kwon <hy...@databricks.com>
(cherry picked from commit eb6c7056ecbeb5442ae875a341dacad7d990cfd8)
Signed-off-by: Hyukjin Kwon <hy...@databricks.com>
---
connector/profiler/README.md | 27 ++++++++++++++
.../org/apache/spark/profiler/ProfilerPlugin.scala | 42 +++++++++++++++++++++-
.../scala/org/apache/spark/profiler/package.scala | 7 ++++
3 files changed, 75 insertions(+), 1 deletion(-)
diff --git a/connector/profiler/README.md b/connector/profiler/README.md
index 742b1a8bb5da..9e74e589220e 100644
--- a/connector/profiler/README.md
+++ b/connector/profiler/README.md
@@ -44,6 +44,23 @@ spark.plugins=org.apache.spark.profiler.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
@@ -73,6 +90,16 @@ Then enable the profiling in the configuration.
</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 cb341aef3a0d..4198b846aab2 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,7 +21,9 @@ import java.util.{Map => JMap}
import scala.jdk.CollectionConverters._
import scala.util.Random
-import org.apache.spark.{SparkConf, SparkContext}
+import one.profiler.Span
+
+import org.apache.spark.{SparkConf, SparkContext, TaskContext, TaskFailedReason}
import org.apache.spark.api.plugin.{DriverPlugin, ExecutorPlugin, PluginContext, SparkPlugin}
import org.apache.spark.internal.Logging
import org.apache.spark.internal.LogKeys.EXECUTOR_ID
@@ -70,12 +72,16 @@ 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) {
@@ -88,6 +94,40 @@ 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 30c2808b3d58..63bb080adc1e 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,6 +45,13 @@ 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: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org