This is an automated email from the ASF dual-hosted git repository.
Yicong-Huang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 202b4d6e6eb1 [SPARK-57324][SQL][UDF] Surface external UDF terminal
metrics
202b4d6e6eb1 is described below
commit 202b4d6e6eb11bf75ec8f539e460734f9c073c79
Author: Timothy Meehan <[email protected]>
AuthorDate: Fri Oct 9 17:38:05 2026 +0000
[SPARK-57324][SQL][UDF] Surface external UDF terminal metrics
### What changes were proposed in this pull request?
Add a typed `ExecutionMetrics` protobuf to the external UDF worker
protocol's `FinishResponse` and `CancelResponse`, as a new `execution_metrics`
field at the end of each message. The existing free-form `metrics` map stays at
field 1, so workers and engines built against either version keep working
together.
The message defines optional fields for row, batch, and payload-byte counts
plus wall-clock and CPU timings, and documents what each field measures: its
direction, unit, and interval, including which timers are wait times. It also
carries two per-stream maxima -- the worker's slowest and most CPU-intensive
data unit -- which the engine cannot recover from the summed timings. Each
terminal response carries one cumulative snapshot for its `Execute` stream:
final totals on finish and par [...]
Create corresponding SQL metrics and aggregate the fields present in the
terminal snapshot when a worker session closes. Values outside Spark's signed
`Long` metric range are ignored. Count metrics start at zero, so a count that a
worker omits adds nothing to the SQL UI total. Worker session creation is a
separate `createUDFWorkerSession` hook, so tests can replace the worker while
running the production close-and-record lifecycle.
### Why are the changes needed?
The protocol currently exposes terminal metrics as string keys and values,
and `ExternalUDFExec` discards them. A typed protobuf gives worker
implementations generated, type-safe accessors and gives the engine a stable
contract for exposing equivalent measurements consistently.
### Does this PR introduce _any_ user-facing change?
Yes, within unreleased `master`. External UDF physical operators expose
typed worker execution metrics in the SQL UI and event data. Spark reads these
metrics from the new `execution_metrics` field; the `metrics` map is unchanged
and remains worker-defined.
### How was this patch tested?
Extended `ExecuteExternalUDFExecSuite` to cover:
- typed snapshots from finish and acknowledged cancellation, recorded
through `ExternalUDFExec`'s own session-close listener; and
- missing versus explicitly measured zero values, including that an omitted
count reports no negative value for the SQL UI to sum.
Ran:
```text
build/sbt 'udf-worker-core/test' 'udf-worker-grpc/test' \
'sql/testOnly
org.apache.spark.sql.execution.externalUDF.ExecuteExternalUDFExecSuite'
```
All tests passed: 59 in `udf-worker-core`, 60 in `udf-worker-grpc`, and 18
in `ExecuteExternalUDFExecSuite`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex 0.147.0
Generated-by: Isaac (Claude Opus 5.5)
This pull request and its description were written by Isaac.
Closes #59111 from tdcmeehan/tdcmeehan/external-udf-terminal-metrics.
Authored-by: Timothy Meehan <[email protected]>
Signed-off-by: Yicong-Huang <[email protected]>
---
.../execution/externalUDF/ExternalUDFExec.scala | 132 ++++++++++++++---
.../externalUDF/ExecuteExternalUDFExecSuite.scala | 161 +++++++++++++++++++--
.../spark/udf/worker/core/WorkerSession.scala | 4 +-
.../org/apache/spark/udf/worker/udf_message.proto | 96 +++++++++++-
4 files changed, 352 insertions(+), 41 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExternalUDFExec.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExternalUDFExec.scala
index db88ef90d62b..bbfa3e331493 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExternalUDFExec.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExternalUDFExec.scala
@@ -17,13 +17,13 @@
package org.apache.spark.sql.execution.externalUDF
-import org.apache.spark.{SparkEnv, TaskContext}
+import org.apache.spark.{SparkContext, SparkEnv, TaskContext}
import org.apache.spark.annotation.Experimental
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.execution.UnaryExecNode
-import org.apache.spark.sql.execution.metric.SQLMetric
-import org.apache.spark.udf.worker.UDFWorkerSpecification
-import org.apache.spark.udf.worker.core.{WorkerSecurityScope, WorkerSession}
+import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
+import org.apache.spark.udf.worker.{ExecutionMetrics, UDFWorkerSpecification}
+import org.apache.spark.udf.worker.core.{Termination, WorkerSecurityScope,
WorkerSession}
/**
* :: Experimental ::
@@ -48,9 +48,8 @@ trait ExternalUDFExec extends UnaryExecNode {
// Metrics
//
---------------------------------------------------------------------------
- protected def externalUdfMetrics: Map[String, SQLMetric] = Map(
- // TODO [SPARK-57324]: Emit the correct metrics here
- )
+ protected val externalUdfMetrics: Map[String, SQLMetric] =
+ ExternalUDFMetrics.create(sparkContext)
override lazy val metrics: Map[String, SQLMetric] = externalUdfMetrics
@@ -59,25 +58,23 @@ trait ExternalUDFExec extends UnaryExecNode {
//
---------------------------------------------------------------------------
/**
- * Creates a [[WorkerSession]] via [[SparkEnv#getExternalUDFDispatcher]].
- * Finalizes the session on task completion (which fires on both success and
- * failure). [[WorkerSession#close]] is the single finalizer: it fetches the
+ * Creates a [[WorkerSession]] with [[createUDFWorkerSession]] and finalizes
it
+ * on task completion (which fires on both success and failure).
+ * [[WorkerSession#close]] is the single finalizer: it fetches the
* `FinishResponse` if processing completed, or cancels anything still in
* flight and waits for the `CancelResponse`. For the UDF sessions used here,
* exhausting the data iterator completes execution and surfaces execution or
- * finish errors. Therefore, `close` only cleans up the session and its
returned
- * termination does not determine whether the Spark task succeeds. The
provided
- * function receives the session and must return the result iterator. It may
use
- * the session but MUST NOT close it.
+ * finish errors. Therefore, `close` cleans up the session, and the
termination
+ * it returns is used only for metrics; it does not determine whether the
Spark
+ * task succeeds. The provided function receives the session and must return
the
+ * result iterator. It may use the session but MUST NOT close it.
*/
protected def withUDFWorkerSession(
taskContext: TaskContext,
securityScope: Option[WorkerSecurityScope] = None)(
f: WorkerSession => Iterator[InternalRow]
): Iterator[InternalRow] = {
- val dispatcher = SparkEnv.get.getExternalUDFDispatcher(
- workerSpec)
- val session = dispatcher.createSession(securityScope)
+ val session = createUDFWorkerSession(securityScope)
// Finalize the session when the task ends. The completion listener fires
on
// both success and failure, and close() is the single finalizer that
@@ -99,11 +96,106 @@ trait ExternalUDFExec extends UnaryExecNode {
// failure has already surfaced through the result iterator.
//
// For these UDF sessions, exhausting the data iterator covers execution
and
- // surfaces its errors. close() only cleans up protocol state and releases
or
- // invalidates the worker handle, so its return value is intentionally
ignored.
+ // surfaces its errors. close() cleans up protocol state and releases or
+ // invalidates the worker handle; a clean terminal response also carries
the
+ // execution's final or partial metrics.
//
- taskContext.addTaskCompletionListener[Unit](_ => session.close())
+ taskContext.addTaskCompletionListener[Unit] { _ =>
+ recordTerminalMetrics(session.close())
+ }
f(session)
}
+
+ protected def createUDFWorkerSession(
+ securityScope: Option[WorkerSecurityScope]): WorkerSession = {
+
SparkEnv.get.getExternalUDFDispatcher(workerSpec).createSession(securityScope)
+ }
+
+ private def recordTerminalMetrics(termination: Termination): Unit = {
+ val reported = termination match {
+ case Termination.Finished(response) if response.hasExecutionMetrics =>
+ Some(response.getExecutionMetrics)
+ case Termination.Cancelled(response) if response.hasExecutionMetrics =>
+ Some(response.getExecutionMetrics)
+ case _ => None
+ }
+ reported.foreach(ExternalUDFMetrics.update(metrics, _))
+ }
+}
+
+private[externalUDF] object ExternalUDFMetrics {
+ private val sizeMetrics = Map(
+ "bytesIn" -> "data received by the external UDF worker",
+ "bytesOut" -> "data returned by the external UDF worker")
+
+ private val countMetrics = Map(
+ "rowsIn" -> "rows received by the external UDF worker",
+ "rowsOut" -> "rows returned by the external UDF worker",
+ "batchesIn" -> "batches received by the external UDF worker",
+ "batchesOut" -> "batches returned by the external UDF worker")
+
+ private val timingMetrics = Map(
+ "initWallNanos" -> "external UDF worker initialization time",
+ "processingWallNanos" -> "external UDF worker processing time",
+ "receiveWallNanos" -> "time the external UDF worker waited for requests",
+ "sendWallNanos" -> "time the external UDF worker was blocked sending
responses",
+ "workWallNanos" -> "external UDF worker execution time",
+ "workCpuNanos" -> "external UDF worker CPU time",
+ "finishWallNanos" -> "external UDF worker finish time")
+
+ // Per-stream maxima rather than totals. Like peakMemory, the figure to read
is the
+ // per-task max in the UI's (min, med, max) breakdown, not the aggregated
total (a sum
+ // of per-task maxima). Formatted as nano timing metrics.
+ private val peakMetrics = Map(
+ "maxWorkWallNanos" -> "longest data unit executed by the external UDF
worker",
+ "maxWorkCpuNanos" -> "most CPU-intensive data unit executed by the
external UDF worker")
+
+ def create(sc: SparkContext): Map[String, SQLMetric] = {
+ sizeMetrics.map { case (name, description) =>
+ name -> SQLMetrics.createSizeMetric(sc, description)
+ } ++ countMetrics.map { case (name, description) =>
+ name -> SQLMetrics.createMetric(sc, description)
+ } ++ (timingMetrics ++ peakMetrics).map { case (name, description) =>
+ name -> SQLMetrics.createNanoTimingMetric(sc, description)
+ }
+ }
+
+ def update(target: Map[String, SQLMetric], reported: ExecutionMetrics): Unit
= {
+ def addIfPresent(name: String, present: Boolean, value: => Long): Unit = {
+ if (present) {
+ target(name) += value
+ }
+ }
+
+ // Peak metrics are per-stream maxima, so a task running several worker
sessions keeps
+ // the largest rather than summing. Mirrors pythonPeakPickledBatchBytes in
+ // BatchEvalPythonExec; set() overwrites and SQLMetric surfaces the
cross-task max.
+ def maxIfPresent(name: String, present: Boolean, value: => Long): Unit = {
+ if (present) {
+ val metric = target(name)
+ val reportedValue = value
+ if (metric.isZero || reportedValue > metric.value) {
+ metric.set(reportedValue)
+ }
+ }
+ }
+
+ addIfPresent("bytesIn", reported.hasBytesIn, reported.getBytesIn)
+ addIfPresent("bytesOut", reported.hasBytesOut, reported.getBytesOut)
+ addIfPresent("rowsIn", reported.hasRowsIn, reported.getRowsIn)
+ addIfPresent("rowsOut", reported.hasRowsOut, reported.getRowsOut)
+ addIfPresent("batchesIn", reported.hasBatchesIn, reported.getBatchesIn)
+ addIfPresent("batchesOut", reported.hasBatchesOut, reported.getBatchesOut)
+ addIfPresent("initWallNanos", reported.hasInitWallNanos,
reported.getInitWallNanos)
+ addIfPresent(
+ "processingWallNanos", reported.hasProcessingWallNanos,
reported.getProcessingWallNanos)
+ addIfPresent("receiveWallNanos", reported.hasReceiveWallNanos,
reported.getReceiveWallNanos)
+ addIfPresent("sendWallNanos", reported.hasSendWallNanos,
reported.getSendWallNanos)
+ addIfPresent("workWallNanos", reported.hasWorkWallNanos,
reported.getWorkWallNanos)
+ addIfPresent("workCpuNanos", reported.hasWorkCpuNanos,
reported.getWorkCpuNanos)
+ addIfPresent("finishWallNanos", reported.hasFinishWallNanos,
reported.getFinishWallNanos)
+ maxIfPresent("maxWorkWallNanos", reported.hasMaxWorkWallNanos,
reported.getMaxWorkWallNanos)
+ maxIfPresent("maxWorkCpuNanos", reported.hasMaxWorkCpuNanos,
reported.getMaxWorkCpuNanos)
+ }
}
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/ExecuteExternalUDFExecSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/ExecuteExternalUDFExecSuite.scala
index 17ebde5ffbbb..29727eaa8ce0 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/ExecuteExternalUDFExecSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/ExecuteExternalUDFExecSuite.scala
@@ -46,8 +46,9 @@ import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types.{DataType, LongType, StructField, StructType}
import org.apache.spark.sql.util.ArrowUtils
-import org.apache.spark.udf.worker.{Cancel, DataRequest, DataResponse, Finish,
FinishResponse, Init,
- InitResponse, UDFWorkerDataFormat, UDFWorkerSpecification}
+import org.apache.spark.udf.worker.{Cancel, CancelResponse, DataRequest,
DataResponse,
+ ExecutionMetrics, Finish, FinishResponse, Init, InitResponse,
UDFWorkerDataFormat,
+ UDFWorkerSpecification}
import org.apache.spark.udf.worker.core.{Termination, WorkerHandle,
WorkerLogger,
WorkerSecurityScope, WorkerSession}
import org.apache.spark.util.{LongAccumulator, Utils}
@@ -86,7 +87,9 @@ object ExecuteExternalUDFExecSuite {
expectedTimeZone: String,
expectedLargeVarTypes: Boolean,
requestCount: LongAccumulator,
- closeCount: LongAccumulator)
+ closeCount: LongAccumulator,
+ terminalMetrics: Option[ExecutionMetrics],
+ cancelOnClose: Boolean)
extends WorkerSession(new TestWorkerHandle, WorkerLogger.NoOp) {
override protected def doInit(message: Init): InitResponse = {
@@ -144,7 +147,18 @@ object ExecuteExternalUDFExecSuite {
override protected def doClose(cancel: () => Cancel): Termination = {
closeCount.add(1L)
- completeTerminal(Termination.Finished(FinishResponse.getDefaultInstance))
+ val termination = if (cancelOnClose) {
+ val response = terminalMetrics.fold(CancelResponse.getDefaultInstance)
{ metrics =>
+ CancelResponse.newBuilder().setExecutionMetrics(metrics).build()
+ }
+ Termination.Cancelled(response)
+ } else {
+ val response = terminalMetrics.fold(FinishResponse.getDefaultInstance)
{ metrics =>
+ FinishResponse.newBuilder().setExecutionMetrics(metrics).build()
+ }
+ Termination.Finished(response)
+ }
+ completeTerminal(termination)
settledTermination
}
@@ -262,24 +276,24 @@ object ExecuteExternalUDFExecSuite {
expectedTimeZone: String,
expectedLargeVarTypes: Boolean,
requestCount: LongAccumulator,
- closeCount: LongAccumulator)
+ closeCount: LongAccumulator,
+ terminalMetrics: Option[ExecutionMetrics],
+ cancelOnClose: Boolean)
extends ExecuteExternalUDFExec(udf, resultAttr, child) {
- override protected def withUDFWorkerSession(
- taskContext: TaskContext,
- securityScope: Option[WorkerSecurityScope])(
- f: WorkerSession => Iterator[InternalRow]): Iterator[InternalRow] = {
+ override protected def createUDFWorkerSession(
+ securityScope: Option[WorkerSecurityScope]): WorkerSession = {
require(securityScope.isEmpty, "scalar external UDF execution must not
request a sandbox")
- val session = new TestWorkerSession(
+ new TestWorkerSession(
behavior,
expectedInputSchema,
expectedOutputSchema,
expectedTimeZone,
expectedLargeVarTypes,
requestCount,
- closeCount)
- taskContext.addTaskCompletionListener[Unit](_ => session.close())
- f(session)
+ closeCount,
+ terminalMetrics,
+ cancelOnClose)
}
}
}
@@ -289,14 +303,18 @@ class ExecuteExternalUDFExecSuite extends QueryTest with
SharedSparkSession {
private def testExecution(
behavior: ResponseBehavior,
- rowCount: Long): TestExecution = {
+ rowCount: Long,
+ terminalMetrics: Option[ExecutionMetrics] = None,
+ cancelOnClose: Boolean = false): TestExecution = {
val child = spark.range(0L, rowCount, 1L, 1).queryExecution.executedPlan
testExecution(
behavior,
child,
Seq(child.output.head),
LongType,
- udfNullable = false)
+ udfNullable = false,
+ terminalMetrics = terminalMetrics,
+ cancelOnClose = cancelOnClose)
}
private def testExecution(
@@ -305,6 +323,24 @@ class ExecuteExternalUDFExecSuite extends QueryTest with
SharedSparkSession {
udfChildren: Seq[Expression],
udfDataType: DataType,
udfNullable: Boolean): TestExecution = {
+ testExecution(
+ behavior,
+ child,
+ udfChildren,
+ udfDataType,
+ udfNullable,
+ terminalMetrics = None,
+ cancelOnClose = false)
+ }
+
+ private def testExecution(
+ behavior: ResponseBehavior,
+ child: SparkPlan,
+ udfChildren: Seq[Expression],
+ udfDataType: DataType,
+ udfNullable: Boolean,
+ terminalMetrics: Option[ExecutionMetrics],
+ cancelOnClose: Boolean): TestExecution = {
val udf = ExternalUserDefinedFunction(
name = Some("identity"),
workerSpec = UDFWorkerSpecification.getDefaultInstance,
@@ -337,7 +373,9 @@ class ExecuteExternalUDFExecSuite extends QueryTest with
SharedSparkSession {
conf.sessionLocalTimeZone,
conf.arrowUseLargeVarTypes,
requestCount,
- closeCount),
+ closeCount,
+ terminalMetrics,
+ cancelOnClose),
requestCount,
closeCount)
}
@@ -396,6 +434,97 @@ class ExecuteExternalUDFExecSuite extends QueryTest with
SharedSparkSession {
parameters = Map.empty)
}
+ gridTest("external UDF records typed terminal metrics from")(
+ Seq("finish", "cancel")) { terminal =>
+ val reported = ExecutionMetrics.newBuilder()
+ .setBytesIn(1)
+ .setBytesOut(2)
+ .setRowsIn(3)
+ .setRowsOut(4)
+ .setBatchesIn(5)
+ .setBatchesOut(6)
+ .setInitWallNanos(7)
+ .setProcessingWallNanos(8)
+ .setReceiveWallNanos(9)
+ .setSendWallNanos(10)
+ .setWorkWallNanos(11)
+ .setWorkCpuNanos(12)
+ .setFinishWallNanos(13)
+ .setMaxWorkWallNanos(14)
+ .setMaxWorkCpuNanos(15)
+ .build()
+ val execution = testExecution(
+ EchoResponses,
+ rowCount = 5L,
+ terminalMetrics = Some(reported),
+ cancelOnClose = terminal == "cancel")
+ val plan = execution.plan.asInstanceOf[TestExecuteExternalUDFExec]
+
+ terminal match {
+ case "finish" => plan.executeCollect()
+ case "cancel" => plan.executeTake(1)
+ case other => fail(s"unexpected terminal: $other")
+ }
+
+ Seq(
+ "bytesIn", "bytesOut", "rowsIn", "rowsOut", "batchesIn", "batchesOut",
+ "initWallNanos", "processingWallNanos", "receiveWallNanos",
"sendWallNanos",
+ "workWallNanos", "workCpuNanos", "finishWallNanos", "maxWorkWallNanos",
+ "maxWorkCpuNanos"
+ ).zipWithIndex.foreach { case (name, index) =>
+ assert(plan.metrics(name).value === index + 1L)
+ }
+ }
+
+ test("external UDF keeps the max, not the sum, of peak metrics across
sessions in a task") {
+ val metrics = ExternalUDFMetrics.create(spark.sparkContext)
+ def snapshot(workWallNanos: Long, maxWall: Long, maxCpu: Long):
ExecutionMetrics =
+ ExecutionMetrics.newBuilder()
+ .setWorkWallNanos(workWallNanos)
+ .setMaxWorkWallNanos(maxWall)
+ .setMaxWorkCpuNanos(maxCpu)
+ .build()
+
+ ExternalUDFMetrics.update(metrics, snapshot(workWallNanos = 10, maxWall =
10, maxCpu = 4))
+ ExternalUDFMetrics.update(metrics, snapshot(workWallNanos = 20, maxWall =
5, maxCpu = 8))
+
+ assert(metrics("workWallNanos").value === 30L)
+ assert(metrics("maxWorkWallNanos").value === 10L)
+ assert(metrics("maxWorkCpuNanos").value === 8L)
+ }
+
+ test("external UDF preserves missing and measured zero terminal metrics") {
+ // bytesIn exercises a measured zero on the additive path; maxWorkCpuNanos
exercises one on the
+ // keep-max path, where only the metric.isZero branch records it (0 never
exceeds the unset
+ // value, which reads 0). maxWorkWallNanos stays unset to cover keep-max
absence.
+ val reported = ExecutionMetrics.newBuilder()
+ .setBytesIn(0)
+ .setMaxWorkCpuNanos(0)
+ .build()
+ val execution = testExecution(
+ EchoResponses,
+ rowCount = 1L,
+ terminalMetrics = Some(reported))
+ val plan = execution.plan.asInstanceOf[TestExecuteExternalUDFExec]
+
+ plan.executeCollect()
+
+ Seq("bytesIn", "maxWorkCpuNanos").foreach { name =>
+ assert(plan.metrics(name).value === 0L)
+ assert(!plan.metrics(name).isZero, s"$name should report measured zero")
+ }
+ Seq(
+ "bytesOut", "initWallNanos", "processingWallNanos", "receiveWallNanos",
"sendWallNanos",
+ "workWallNanos", "workCpuNanos", "finishWallNanos", "maxWorkWallNanos"
+ ).foreach { name =>
+ assert(plan.metrics(name).isZero, s"$name should be unreported")
+ }
+ // toInfoUpdate is the raw value the live SQL UI sums; value hides a
negative initial value.
+ Seq("rowsIn", "rowsOut", "batchesIn", "batchesOut").foreach { name =>
+ assert(plan.metrics(name).toInfoUpdate.update === Some(0L))
+ }
+ }
+
test("scalar external UDF exchanges multiple Arrow batches through a worker
session") {
withSQLConf(
SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC",
diff --git
a/udf/worker/core/src/main/scala/org/apache/spark/udf/worker/core/WorkerSession.scala
b/udf/worker/core/src/main/scala/org/apache/spark/udf/worker/core/WorkerSession.scala
index 0afde972bc29..7cef64c1da33 100644
---
a/udf/worker/core/src/main/scala/org/apache/spark/udf/worker/core/WorkerSession.scala
+++
b/udf/worker/core/src/main/scala/org/apache/spark/udf/worker/core/WorkerSession.scala
@@ -50,7 +50,7 @@ import org.apache.spark.udf.worker.{Cancel, DataRequest,
DataResponse, Finish, I
*
* '''Lifecycle:'''
* - [[init]] must be called exactly once before [[process]]; it returns the
- * worker's [[InitResponse]] (init-callback result + metrics).
+ * worker's [[InitResponse]] (init-callback result).
* - [[process]] must be called at most once per session. It returns the
* result iterator; consuming that iterator drives the data exchange and
* surfaces any UDF / data-phase error (as an exception from `next()` /
@@ -148,7 +148,7 @@ abstract class WorkerSession(
/**
* Initializes the UDF execution. Must be called exactly once before
* [[process]]. Returns the worker's [[InitResponse]] (carrying the
- * init-callback result and metrics) on success; throws if the worker
+ * init-callback result) on success; throws if the worker
* reports an init error or the stream fails before init completes.
*
* @param message the [[Init]] message carrying the UDF body, data
diff --git
a/udf/worker/proto/src/main/protobuf/org/apache/spark/udf/worker/udf_message.proto
b/udf/worker/proto/src/main/protobuf/org/apache/spark/udf/worker/udf_message.proto
index e1be17b15d9f..d1bf2b5dda85 100644
---
a/udf/worker/proto/src/main/protobuf/org/apache/spark/udf/worker/udf_message.proto
+++
b/udf/worker/proto/src/main/protobuf/org/apache/spark/udf/worker/udf_message.proto
@@ -395,10 +395,13 @@ message Finish {}
// Worker -> Engine completion message. Carries per-execution summary metrics.
//
// Metrics design:
-// - [[FinishResponse.metrics]] / [[CancelResponse.metrics]]: per-execution
-// metrics accumulated up to the point of stream termination (e.g. rows
-// processed, time per phase). Emitted once per [[Execute]] stream,
+// - [[FinishResponse.execution_metrics]] /
+// [[CancelResponse.execution_metrics]]: per-execution metrics with
+// defined semantics (see [[ExecutionMetrics]]), accumulated up to the
+// point of stream termination. Emitted once per [[Execute]] stream,
// regardless of whether the stream ended cleanly or was cancelled.
+// - [[FinishResponse.metrics]] / [[CancelResponse.metrics]]: free-form,
+// worker-defined per-execution metrics, sent alongside them.
// - [[HeartbeatResponse.metrics]]: worker-global metrics aggregated across
// all sessions (e.g. total rows processed, memory usage). Emitted
// periodically via the [[Manage]] RPC.
@@ -426,6 +429,11 @@ message FinishResponse {
// data processing, the terminator is [[CancelResponse]], not
// [[FinishResponse]]. The engine SHOULD surface this as an exception.
optional ExecutionError error = 3;
+
+ // (Optional) Final metrics accumulated over this Execute stream, in
+ // the typed fields the engine reads. Measurements that have no field
+ // there belong in [[metrics]].
+ optional ExecutionMetrics execution_metrics = 4;
}
// Engine -> Worker explicit cancel. Distinct from a gRPC stream error
@@ -480,6 +488,88 @@ message CancelResponse {
// alongside any prior [[ExecutionError]] (e.g. as a suppressed
// exception).
optional ExecutionError error = 2;
+
+ // (Optional) Partial metrics accumulated over this Execute stream up
+ // to cancellation. See [[FinishResponse.execution_metrics]].
+ optional ExecutionMetrics execution_metrics = 3;
+}
+
+// Per-execution metrics accumulated by one Execute stream, as measured by the
+// worker. Each field is cumulative for the stream, except
+// [[max_work_wall_nanos]] and [[max_work_cpu_nanos]], which are per-stream
+// maxima. Absence means unavailable;
+// zero means measured zero. "In" counts data the worker received from the
+// engine and "out" counts data it returned. Byte counts are data payload
+// bytes and exclude gRPC, HTTP/2, and TLS framing. Timing values are
+// nanoseconds; the wait times [[receive_wall_nanos]] and [[send_wall_nanos]]
+// are not transfer times and MUST NOT be used as rate denominators. Spark
+// reads each field as a signed 64-bit value and ignores values above
+// 2^63 - 1.
+//
+// [[init_wall_nanos]], [[processing_wall_nanos]], and [[finish_wall_nanos]]
+// are consecutive windows of the stream. [[receive_wall_nanos]],
+// [[send_wall_nanos]], [[work_wall_nanos]], and [[work_cpu_nanos]] are
+// totals, summed across the worker's threads. They can overlap and do not
+// add up to any window; [[work_wall_nanos]] can exceed
+// [[processing_wall_nanos]].
+message ExecutionMetrics {
+ // Bytes of [[DataRequest.data]] payload received.
+ optional uint64 bytes_in = 1;
+
+ // Bytes of [[DataResponse.data]] payload returned.
+ optional uint64 bytes_out = 2;
+
+ // Rows in the received data payloads. Absent unless the rows of every
+ // received payload could be counted.
+ optional uint64 rows_in = 3;
+
+ // Rows in the returned data payloads. Absent unless the rows of every
+ // returned payload could be counted.
+ optional uint64 rows_out = 4;
+
+ // Number of [[DataRequest]] messages received.
+ optional uint64 batches_in = 5;
+
+ // Number of [[DataResponse]] messages returned.
+ optional uint64 batches_out = 6;
+
+ // Wall time spent handling [[Init]] before sending [[InitResponse]].
+ optional uint64 init_wall_nanos = 7;
+
+ // Wall time from sending [[InitResponse]] until data processing ends,
+ // when the worker has received [[Finish]] and sent all of its output.
+ // Cancellation or an execution error also ends it.
+ optional uint64 processing_wall_nanos = 8;
+
+ // Wall time spent waiting for the next request from the engine.
+ optional uint64 receive_wall_nanos = 9;
+
+ // Wall time spent blocked sending responses to the engine.
+ optional uint64 send_wall_nanos = 10;
+
+ // Wall time spent executing received data: decoding input, running the
+ // UDF, and encoding output. Excludes the wait times.
+ optional uint64 work_wall_nanos = 11;
+
+ // CPU time consumed executing received data.
+ optional uint64 work_cpu_nanos = 12;
+
+ // Wall time from the end of data processing until sending
+ // [[FinishResponse]] or [[CancelResponse]], such as running the finish
+ // callback. Absent if cancellation or an error ended data processing.
+ optional uint64 finish_wall_nanos = 13;
+
+ // Wall time of the single longest data unit the worker executed: the
+ // maximum over the stream of the per-unit work time that
+ // [[work_wall_nanos]] totals. Lets the engine surface the slowest unit,
+ // which the summed total cannot. Not cumulative. Absent means unavailable.
+ optional uint64 max_work_wall_nanos = 14;
+
+ // CPU time of the single most CPU-intensive data unit the worker executed:
+ // the maximum over the stream of the per-unit CPU time that
+ // [[work_cpu_nanos]] totals. The CPU counterpart of
[[max_work_wall_nanos]].
+ // Not cumulative. Absent means unavailable.
+ optional uint64 max_work_cpu_nanos = 15;
}
// Worker -> Engine application-level error raised during the
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
