You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "tdcmeehan (via GitHub)" <gi...@apache.org> on 2026/09/28 21:35:38 UTC
[PR] [SPARK-57324][SQL] Surface external UDF terminal metrics [spark]
tdcmeehan opened a new pull request, #59111:
URL: https://github.com/apache/spark/pull/59111
### What changes were proposed in this pull request?
Create SQL metrics for the portable external UDF worker counters, byte totals, and nanosecond timings. When a worker session closes, aggregate terminal metrics returned by both FinishResponse and acknowledged CancelResponse.
Unknown metric names and malformed, negative, or overflowing values are ignored so workers can evolve without breaking query completion.
### Why are the changes needed?
Worker terminal responses already preserve per-execution metrics, but ExternalUDFExec discarded the returned Termination. This connects that existing client boundary to Spark SQL metrics.
### Does this PR introduce any user-facing change?
Yes. External UDF physical operators expose worker-provided execution metrics in the SQL UI and event data.
### How was this patch tested?
Added ExternalUDFExecSuite coverage for finish and cancellation aggregation and for unknown, malformed, negative, and overflowing values.
The focused suite passed with 2 tests, and the 19-module reactor compiled successfully.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57324][SQL] Surface external UDF terminal metrics [spark]
Posted by "tdcmeehan (via GitHub)" <gi...@apache.org>.
tdcmeehan commented on code in PR #59111:
URL: https://github.com/apache/spark/pull/59111#discussion_r4211239314
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExternalUDFExec.scala:
##########
@@ -99,11 +96,84 @@ 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")
+
+ 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.map { case (name, description) =>
+ name -> SQLMetrics.createNanoTimingMetric(sc, description)
+ }
+ }
+
+ def update(target: Map[String, SQLMetric], reported: ExecutionMetrics): Unit = {
+ def updateIfPresent(name: String, present: Boolean, value: => Long): Unit = {
+ if (present) {
+ target(name) += value
+ }
Review Comment:
You're right for the timing fields -- Spark sees one cumulative snapshot per task, never per-unit worker time, so a within-stream max can only come from the worker. Added two in e74992a: `max_work_wall_nanos` and `max_work_cpu_nanos`, the slowest and most CPU-intensive data unit. They're timing metrics, so the figure to read is the per-task max in the UI's (min, med, max) breakdown; the summed total isn't meaningful, like `peakMemory`. (Peak batch *size* stays Spark-side -- it sees every request and response -- as the follow-up with the Spark-side byte/row counts.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57324][SQL] Surface external UDF terminal metrics [spark]
Posted by "haiyangsun-db (via GitHub)" <gi...@apache.org>.
haiyangsun-db commented on code in PR #59111:
URL: https://github.com/apache/spark/pull/59111#discussion_r4146352881
##########
udf/worker/proto/src/main/protobuf/org/apache/spark/udf/worker/udf_message.proto:
##########
@@ -482,6 +479,55 @@ message CancelResponse {
optional ExecutionError error = 2;
}
+// Per-execution metrics accumulated by one Execute stream, as measured by the
+// worker. Each field is cumulative for the stream. 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.
+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;
Review Comment:
would it make sense to track max batch size?
##########
udf/worker/proto/src/main/protobuf/org/apache/spark/udf/worker/udf_message.proto:
##########
@@ -482,6 +479,55 @@ message CancelResponse {
optional ExecutionError error = 2;
}
+// Per-execution metrics accumulated by one Execute stream, as measured by the
+// worker. Each field is cumulative for the stream. 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.
+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 the stream's first
+ // terminal event: finish, cancellation, or an execution error.
+ optional uint64 processing_wall_nanos = 8;
Review Comment:
Something not very generic, but could be helpful is the first result batch sent wall time since processing starts, that could somehow indicate the latency of the processing. In a good streaming-style execution, the first batch could be streamed soon for lower latency.
But this is not a very standard metric, we can also add later if you prefer.
##########
udf/worker/proto/src/main/protobuf/org/apache/spark/udf/worker/udf_message.proto:
##########
@@ -482,6 +479,55 @@ message CancelResponse {
optional ExecutionError error = 2;
}
+// Per-execution metrics accumulated by one Execute stream, as measured by the
+// worker. Each field is cumulative for the stream. 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.
+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 the stream's first
+ // terminal event: finish, cancellation, or an execution error.
+ 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;
Review Comment:
There is also a wall time between last result batch sent - the data processing phase ends, and finish response sent.
Usually in spark udf case, the data processing ends would already trigger task completes, while the finish response is handled in the task completion listener.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57324][SQL] Surface external UDF terminal metrics [spark]
Posted by "haiyangsun-db (via GitHub)" <gi...@apache.org>.
haiyangsun-db commented on code in PR #59111:
URL: https://github.com/apache/spark/pull/59111#discussion_r4146295602
##########
udf/worker/proto/src/main/protobuf/org/apache/spark/udf/worker/udf_message.proto:
##########
@@ -407,9 +407,8 @@ message Finish {}
// introduced as a separate streaming message in a future revision
// without changing this contract. TODO(SPARK-56922).
message FinishResponse {
- // Per-execution metrics accumulated over the session. Free-form;
- // names are worker-defined.
- map<string, string> metrics = 1;
+ // (Optional) Final metrics accumulated over this Execute stream.
+ optional ExecutionMetrics metrics = 1;
Review Comment:
can we append this field at the end of the message, it doesn't have to replace the generic metrics
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57324][SQL] Surface external UDF terminal metrics [spark]
Posted by "tdcmeehan (via GitHub)" <gi...@apache.org>.
tdcmeehan commented on code in PR #59111:
URL: https://github.com/apache/spark/pull/59111#discussion_r4213144955
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExternalUDFExec.scala:
##########
@@ -99,11 +96,84 @@ 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")
+
+ 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.map { case (name, description) =>
+ name -> SQLMetrics.createNanoTimingMetric(sc, description)
+ }
+ }
+
+ def update(target: Map[String, SQLMetric], reported: ExecutionMetrics): Unit = {
+ def updateIfPresent(name: String, present: Boolean, value: => Long): Unit = {
+ if (present) {
+ target(name) += value
+ }
Review Comment:
You're right. `+=` summed the per-stream maxima whenever a task runs more than one session -- your `coalesce(1)` case: `CoalescedRDD` folds several parent partitions into one task, and each session's completion listener calls `update` on the same metrics map. Fixed in 9134b57: the two peak fields now go through a `maxIfPresent` that keeps the per-task max (`if (metric.isZero || v > metric.value) metric.set(v)`), mirroring `pythonPeakPickledBatchBytes` in `BatchEvalPythonExec`; the cumulative fields keep summing via `addIfPresent`. Added a regression test that records two snapshots in one task and asserts the peaks take the max (10, 8) while `workWallNanos` sums (30) -- it fails on the old `+=`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExternalUDFExec.scala:
##########
@@ -99,11 +96,96 @@ 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 the worker reports rather than totals. Like peakMemory,
+ // these are timing metrics so the figure to read is the per-task max in the
+ // UI's min/med/max breakdown; the aggregated total (a sum of per-task maxima)
+ // is not meaningful on its own.
+ 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 updateIfPresent(name: String, present: Boolean, value: => Long): Unit = {
+ if (present) {
+ target(name) += value
+ }
+ }
+
+ updateIfPresent("bytesIn", reported.hasBytesIn, reported.getBytesIn)
+ updateIfPresent("bytesOut", reported.hasBytesOut, reported.getBytesOut)
+ updateIfPresent("rowsIn", reported.hasRowsIn, reported.getRowsIn)
+ updateIfPresent("rowsOut", reported.hasRowsOut, reported.getRowsOut)
+ updateIfPresent("batchesIn", reported.hasBatchesIn, reported.getBatchesIn)
+ updateIfPresent("batchesOut", reported.hasBatchesOut, reported.getBatchesOut)
+ updateIfPresent("initWallNanos", reported.hasInitWallNanos, reported.getInitWallNanos)
+ updateIfPresent(
+ "processingWallNanos", reported.hasProcessingWallNanos, reported.getProcessingWallNanos)
+ updateIfPresent("receiveWallNanos", reported.hasReceiveWallNanos, reported.getReceiveWallNanos)
+ updateIfPresent("sendWallNanos", reported.hasSendWallNanos, reported.getSendWallNanos)
+ updateIfPresent("workWallNanos", reported.hasWorkWallNanos, reported.getWorkWallNanos)
+ updateIfPresent("workCpuNanos", reported.hasWorkCpuNanos, reported.getWorkCpuNanos)
+ updateIfPresent("finishWallNanos", reported.hasFinishWallNanos, reported.getFinishWallNanos)
+ updateIfPresent(
+ "maxWorkWallNanos", reported.hasMaxWorkWallNanos, reported.getMaxWorkWallNanos)
Review Comment:
Fixed in 9134b57 -- the peak fields now keep the per-task max via `maxIfPresent` instead of `+=` (mirrors `pythonPeakPickledBatchBytes`), with a regression test for the multi-session-per-task case. More detail on the min/max thread.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExternalUDFExec.scala:
##########
@@ -99,11 +96,96 @@ 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 the worker reports rather than totals. Like peakMemory,
Review Comment:
Reworded in 9134b57; dropped "these are timing metrics" since peakMemory is a size metric. It now reads: per-stream maxima rather than totals; like peakMemory, the figure to read is the per-task max in the (min, med, max) breakdown; formatted as nano timing metrics.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/ExecuteExternalUDFExecSuite.scala:
##########
@@ -396,6 +434,69 @@ 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 preserves missing and measured zero terminal metrics") {
+ val reported = ExecutionMetrics.newBuilder()
+ .setBytesIn(0)
+ .build()
+ val execution = testExecution(
+ EchoResponses,
+ rowCount = 1L,
+ terminalMetrics = Some(reported))
+ val plan = execution.plan.asInstanceOf[TestExecuteExternalUDFExec]
+
+ plan.executeCollect()
+
+ assert(plan.metrics("bytesIn").value === 0L)
+ assert(!plan.metrics("bytesIn").isZero)
+ assert(plan.metrics("bytesOut").isZero)
Review Comment:
Done in 9134b57 -- the test now asserts `isZero` for every size/timing/peak metric except the reported `bytesIn`, so a miswired `hasX` guard on any of them fails here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57324][SQL] Surface external UDF terminal metrics [spark]
Posted by "haiyangsun-db (via GitHub)" <gi...@apache.org>.
haiyangsun-db commented on code in PR #59111:
URL: https://github.com/apache/spark/pull/59111#discussion_r4147944845
##########
udf/worker/proto/src/main/protobuf/org/apache/spark/udf/worker/udf_message.proto:
##########
@@ -480,6 +488,59 @@ 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. 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.
+message ExecutionMetrics {
+ // Bytes of [[DataRequest.data]] payload received.
+ optional uint64 bytes_in = 1;
Review Comment:
non-blocking:
Long.MaxValue is the maximum value Spark accepts for ExecutionMetrics. The wire fields accept larger uint64 values. Maybe worth a one-line note, Long.MaxValue is already a too large value for the metrics to reach.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57324][SQL][UDF] Surface external UDF terminal metrics [spark]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on PR #59111:
URL: https://github.com/apache/spark/pull/59111#issuecomment-6086063990
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/202b4d6e6eb11bf75ec8f539e460734f9c073c79
- merged into branch-4.x https://github.com/apache/spark/commit/d88acf5f482a4930c5e51b663caf56f44a4146c9
*Posted by `merge_spark_pr.py`*
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57324][SQL] Surface external UDF terminal metrics [spark]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on code in PR #59111:
URL: https://github.com/apache/spark/pull/59111#discussion_r4149439049
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExternalUDFExec.scala:
##########
@@ -99,11 +96,84 @@ 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")
+
+ 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.map { case (name, description) =>
+ name -> SQLMetrics.createNanoTimingMetric(sc, description)
+ }
+ }
+
+ def update(target: Map[String, SQLMetric], reported: ExecutionMetrics): Unit = {
+ def updateIfPresent(name: String, present: Boolean, value: => Long): Unit = {
+ if (present) {
+ target(name) += value
+ }
Review Comment:
hmm, if we are only taking the sum, we will lose the ability for min/max etc. would it be a bad idea to add them, if space is not a concern?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExternalUDFExec.scala:
##########
@@ -99,11 +96,84 @@ 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")
Review Comment:
can some of them overlap?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/ExecuteExternalUDFExecSuite.scala:
##########
@@ -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()
+ }
Review Comment:
would it make sense to also record count of cancels?
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExternalUDFExec.scala:
##########
@@ -99,11 +96,84 @@ 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")
Review Comment:
I think we need a comment or declaraition of each wall clock time and their relationships.
for instance, I am not sure if it is fair to say "receive + process + send = work time"?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57324][SQL] Surface external UDF terminal metrics [spark]
Posted by "tdcmeehan (via GitHub)" <gi...@apache.org>.
tdcmeehan commented on code in PR #59111:
URL: https://github.com/apache/spark/pull/59111#discussion_r4150510927
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExternalUDFExec.scala:
##########
@@ -99,11 +96,84 @@ 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")
Review Comment:
Good point: they don't add up. Init, processing and finish are back-to-back windows of the stream. Receive, send and work are totals rather than windows; they can overlap, and a worker using several threads sums each one across its threads, so work can exceed processing. Spelled this out in the `ExecutionMetrics` comment in 0913081d.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExternalUDFExec.scala:
##########
@@ -99,11 +96,84 @@ 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")
+
+ 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.map { case (name, description) =>
+ name -> SQLMetrics.createNanoTimingMetric(sc, description)
+ }
+ }
+
+ def update(target: Map[String, SQLMetric], reported: ExecutionMetrics): Unit = {
+ def updateIfPresent(name: String, present: Boolean, value: => Long): Unit = {
+ if (present) {
+ target(name) += value
+ }
Review Comment:
The SQL UI already shows min, median and max across tasks for every size and timing metric: each task keeps its own value, and `+=` only adds that session's total to it. What we don't keep is the spread inside one task, like the slowest batch, which would need per-batch fields. I'd add those where they're needed, starting with peak batch size in the Spark-side follow-up.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/ExecuteExternalUDFExecSuite.scala:
##########
@@ -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()
+ }
Review Comment:
Spark already knows how each session ended, so this wouldn't need a protocol change. The catch is that SQL metrics from failed or killed tasks are dropped, so it would mostly count early stops such as a `LIMIT`. I'd leave it out for now unless there's a case for counting those.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57324][SQL] Surface external UDF terminal metrics [spark]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on PR #59111:
URL: https://github.com/apache/spark/pull/59111#issuecomment-6016750853
Let's wait for @Yicong-Huang's re-review here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57324][SQL] Surface external UDF terminal metrics [spark]
Posted by "tdcmeehan (via GitHub)" <gi...@apache.org>.
tdcmeehan commented on code in PR #59111:
URL: https://github.com/apache/spark/pull/59111#discussion_r4148471431
##########
udf/worker/proto/src/main/protobuf/org/apache/spark/udf/worker/udf_message.proto:
##########
@@ -407,9 +407,8 @@ message Finish {}
// introduced as a separate streaming message in a future revision
// without changing this contract. TODO(SPARK-56922).
message FinishResponse {
- // Per-execution metrics accumulated over the session. Free-form;
- // names are worker-defined.
- map<string, string> metrics = 1;
+ // (Optional) Final metrics accumulated over this Execute stream.
+ optional ExecutionMetrics metrics = 1;
Review Comment:
Done in 8b59fd58: the `metrics` map is back at field 1 unchanged, and the typed message is appended as `execution_metrics` (field 4 in `FinishResponse`, field 3 in `CancelResponse`).
##########
udf/worker/proto/src/main/protobuf/org/apache/spark/udf/worker/udf_message.proto:
##########
@@ -482,6 +479,55 @@ message CancelResponse {
optional ExecutionError error = 2;
}
+// Per-execution metrics accumulated by one Execute stream, as measured by the
+// worker. Each field is cumulative for the stream. 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.
+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;
Review Comment:
Good idea. Spark sees every request and response, so I'll measure peak batch size on the Spark side, like `pythonPeakPickledBatchBytes`, in the follow-up with the Spark-side byte and row counts.
##########
udf/worker/proto/src/main/protobuf/org/apache/spark/udf/worker/udf_message.proto:
##########
@@ -482,6 +479,55 @@ message CancelResponse {
optional ExecutionError error = 2;
}
+// Per-execution metrics accumulated by one Execute stream, as measured by the
+// worker. Each field is cumulative for the stream. 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.
+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 the stream's first
+ // terminal event: finish, cancellation, or an execution error.
+ 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;
Review Comment:
Added in feecacb4. `processing_wall_nanos` now ends once the worker has received `Finish` and sent all of its output, and `finish_wall_nanos` covers the time from there to the terminal response. I started it there rather than at the last result batch: a worker that answers each batch right away sends its last batch before `Finish` arrives, and that wait is Spark's time.
##########
udf/worker/proto/src/main/protobuf/org/apache/spark/udf/worker/udf_message.proto:
##########
@@ -482,6 +479,55 @@ message CancelResponse {
optional ExecutionError error = 2;
}
+// Per-execution metrics accumulated by one Execute stream, as measured by the
+// worker. Each field is cumulative for the stream. 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.
+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 the stream's first
+ // terminal event: finish, cancellation, or an execution error.
+ optional uint64 processing_wall_nanos = 8;
Review Comment:
Agreed, let's add it later. Spark can measure it from the first batch sent to the first result received, so it'll go with the Spark-side metrics.
##########
udf/worker/proto/src/main/protobuf/org/apache/spark/udf/worker/udf_message.proto:
##########
@@ -480,6 +488,59 @@ 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. 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.
+message ExecutionMetrics {
+ // Bytes of [[DataRequest.data]] payload received.
+ optional uint64 bytes_in = 1;
Review Comment:
Added in feecacb4: the `ExecutionMetrics` comment now says Spark reads each field as a signed 64-bit value and ignores values above 2^63 - 1.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57324][SQL] Surface external UDF terminal metrics [spark]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on code in PR #59111:
URL: https://github.com/apache/spark/pull/59111#discussion_r4212507857
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExternalUDFExec.scala:
##########
@@ -99,11 +96,96 @@ 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 the worker reports rather than totals. Like peakMemory,
+ // these are timing metrics so the figure to read is the per-task max in the
+ // UI's min/med/max breakdown; the aggregated total (a sum of per-task maxima)
+ // is not meaningful on its own.
+ 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 updateIfPresent(name: String, present: Boolean, value: => Long): Unit = {
+ if (present) {
+ target(name) += value
+ }
+ }
+
+ updateIfPresent("bytesIn", reported.hasBytesIn, reported.getBytesIn)
+ updateIfPresent("bytesOut", reported.hasBytesOut, reported.getBytesOut)
+ updateIfPresent("rowsIn", reported.hasRowsIn, reported.getRowsIn)
+ updateIfPresent("rowsOut", reported.hasRowsOut, reported.getRowsOut)
+ updateIfPresent("batchesIn", reported.hasBatchesIn, reported.getBatchesIn)
+ updateIfPresent("batchesOut", reported.hasBatchesOut, reported.getBatchesOut)
+ updateIfPresent("initWallNanos", reported.hasInitWallNanos, reported.getInitWallNanos)
+ updateIfPresent(
+ "processingWallNanos", reported.hasProcessingWallNanos, reported.getProcessingWallNanos)
+ updateIfPresent("receiveWallNanos", reported.hasReceiveWallNanos, reported.getReceiveWallNanos)
+ updateIfPresent("sendWallNanos", reported.hasSendWallNanos, reported.getSendWallNanos)
+ updateIfPresent("workWallNanos", reported.hasWorkWallNanos, reported.getWorkWallNanos)
+ updateIfPresent("workCpuNanos", reported.hasWorkCpuNanos, reported.getWorkCpuNanos)
+ updateIfPresent("finishWallNanos", reported.hasFinishWallNanos, reported.getFinishWallNanos)
+ updateIfPresent(
+ "maxWorkWallNanos", reported.hasMaxWorkWallNanos, reported.getMaxWorkWallNanos)
Review Comment:
**Nit (P3):** These two are per-stream maxima, but `update` merges them with `+=`. When one task runs several sessions (e.g. `df.select(udf(x)).coalesce(1)`, where `CoalescedRDD` computes every parent partition in one task), the per-task value becomes the sum of the per-stream maxima, not the max the comment above says to read. Could these keep the per-task max instead, like `pythonPeakPickledBatchBytes` in `BatchEvalPythonExec` (`if (m.isZero || v > m.value) m.set(v)`)?
**Verification:**
- **Regression:** Two terminal snapshots recorded into the same metrics in one task produce the max of their max_work_* values and the sum of their cumulative values.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExternalUDFExec.scala:
##########
@@ -99,11 +96,96 @@ 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 the worker reports rather than totals. Like peakMemory,
Review Comment:
**Nit (P3):** nit: `peakMemory` is a size metric (`createSizeMetric` in `SortExec`/`HashAggregateExec`), so "Like peakMemory, these are timing metrics" reads wrong. Maybe: "Like peakMemory, the figure to read is the per-task max in the UI's min/med/max breakdown; they use nano timing metrics for formatting."
**Verification:**
- **Inspection:** The peak metric comment describes peakMemory only as an analogue for reading the per-task max.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/ExecuteExternalUDFExecSuite.scala:
##########
@@ -396,6 +434,69 @@ 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 preserves missing and measured zero terminal metrics") {
+ val reported = ExecutionMetrics.newBuilder()
+ .setBytesIn(0)
+ .build()
+ val execution = testExecution(
+ EchoResponses,
+ rowCount = 1L,
+ terminalMetrics = Some(reported))
+ val plan = execution.plan.asInstanceOf[TestExecuteExternalUDFExec]
+
+ plan.executeCollect()
+
+ assert(plan.metrics("bytesIn").value === 0L)
+ assert(!plan.metrics("bytesIn").isZero)
+ assert(plan.metrics("bytesOut").isZero)
Review Comment:
**Nit (P3):** This only checks `bytesOut` for absence. A miswired `hasX` guard on any timing/peak field (e.g. `sendWallNanos` using `hasReceiveWallNanos`) would still pass this and the grid test. Could we assert `isZero` for every non-count metric other than `bytesIn`?
**Verification:**
- **Behavior:** With only bytes_in reported, every other size, timing and peak metric stays unreported while counts report raw zero.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/externalUDF/ExternalUDFExec.scala:
##########
@@ -99,11 +96,84 @@ 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")
+
+ 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.map { case (name, description) =>
+ name -> SQLMetrics.createNanoTimingMetric(sc, description)
+ }
+ }
+
+ def update(target: Map[String, SQLMetric], reported: ExecutionMetrics): Unit = {
+ def updateIfPresent(name: String, present: Boolean, value: => Long): Unit = {
+ if (present) {
+ target(name) += value
+ }
Review Comment:
Thanks, the two max fields look good. One catch: `update` merges them with `+=`, so when one task runs several sessions (e.g. `coalesce(1)` after the UDF, where CoalescedRDD computes all parent partitions in one task), the per-task value becomes the sum of the per-stream maxima rather than the max. Could these two keep the per-task max instead, like `pythonPeakPickledBatchBytes` does (`if (m.isZero || v > m.value) m.set(v)`)?
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"inline:4149439049","thread_id":"inline:4149439049","verdict_sha256":"87477ce65a50a26d03c88ebb991591a37555f280ae823d9e3898c044afa9c70c"} -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57324][SQL] Surface external UDF terminal metrics [spark]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on PR #59111:
URL: https://github.com/apache/spark/pull/59111#issuecomment-6047832124
Thanks for waiting. I've re-reviewed the latest revision (e74992a), including the new max_work_* fields; my comments are in this review.
<!-- SPARK_DEV_REVIEW_REPLY {"feedback_id":"conversation:6016750853","thread_id":"conversation:6016750853","verdict_sha256":"87477ce65a50a26d03c88ebb991591a37555f280ae823d9e3898c044afa9c70c"} -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57324][SQL] Surface external UDF terminal metrics [spark]
Posted by "haiyangsun-db (via GitHub)" <gi...@apache.org>.
haiyangsun-db commented on code in PR #59111:
URL: https://github.com/apache/spark/pull/59111#discussion_r4146352881
##########
udf/worker/proto/src/main/protobuf/org/apache/spark/udf/worker/udf_message.proto:
##########
@@ -482,6 +479,55 @@ message CancelResponse {
optional ExecutionError error = 2;
}
+// Per-execution metrics accumulated by one Execute stream, as measured by the
+// worker. Each field is cumulative for the stream. 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.
+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;
Review Comment:
non-blocking: would it make sense to track max batch size?
##########
udf/worker/proto/src/main/protobuf/org/apache/spark/udf/worker/udf_message.proto:
##########
@@ -482,6 +479,55 @@ message CancelResponse {
optional ExecutionError error = 2;
}
+// Per-execution metrics accumulated by one Execute stream, as measured by the
+// worker. Each field is cumulative for the stream. 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.
+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 the stream's first
+ // terminal event: finish, cancellation, or an execution error.
+ optional uint64 processing_wall_nanos = 8;
Review Comment:
non-blocking: Something not very generic, but could be helpful is the first result batch sent wall time since processing starts, that could somehow indicate the latency of the processing. In a good streaming-style execution, the first batch could be streamed soon for lower latency.
But this is not a very standard metric, we can also add later if you prefer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57324][SQL] Surface external UDF terminal metrics [spark]
Posted by "tdcmeehan (via GitHub)" <gi...@apache.org>.
tdcmeehan commented on code in PR #59111:
URL: https://github.com/apache/spark/pull/59111#discussion_r4223124129
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/ExecuteExternalUDFExecSuite.scala:
##########
@@ -396,6 +434,91 @@ 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") {
+ val reported = ExecutionMetrics.newBuilder()
+ .setBytesIn(0)
Review Comment:
Good catch -- that branch was untested. Covered in 8107481: the measured-zero test now also reports `max_work_cpu_nanos = 0` and asserts the metric is `!isZero` with value 0, while `max_work_wall_nanos` stays unset for the keep-max absence case. Mutation-checked: dropping the `metric.isZero ||` disjunct makes it fail with `maxWorkCpuNanos.isZero was true` -- exactly your verification.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57324][SQL] Surface external UDF terminal metrics [spark]
Posted by "haiyangsun-db (via GitHub)" <gi...@apache.org>.
haiyangsun-db commented on code in PR #59111:
URL: https://github.com/apache/spark/pull/59111#discussion_r4146379297
##########
udf/worker/proto/src/main/protobuf/org/apache/spark/udf/worker/udf_message.proto:
##########
@@ -482,6 +479,55 @@ message CancelResponse {
optional ExecutionError error = 2;
}
+// Per-execution metrics accumulated by one Execute stream, as measured by the
+// worker. Each field is cumulative for the stream. 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.
+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 the stream's first
+ // terminal event: finish, cancellation, or an execution error.
+ 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;
Review Comment:
There is also a wall time between last result batch sent - the data processing phase ends, and finish response sent.
Usually in spark udf case, the data processing ends would already trigger task completes, while the finish response is handled in the task completion listener - for metrics, and properly worker session cleanup / reuse.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57324][SQL] Surface external UDF terminal metrics [spark]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on PR #59111:
URL: https://github.com/apache/spark/pull/59111#issuecomment-6016746657
Thank you @tdcmeehan and @haiyangsun-db @Yicong-Huang!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57324][SQL] Surface external UDF terminal metrics [spark]
Posted by "haiyangsun-db (via GitHub)" <gi...@apache.org>.
haiyangsun-db commented on code in PR #59111:
URL: https://github.com/apache/spark/pull/59111#discussion_r4146379297
##########
udf/worker/proto/src/main/protobuf/org/apache/spark/udf/worker/udf_message.proto:
##########
@@ -482,6 +479,55 @@ message CancelResponse {
optional ExecutionError error = 2;
}
+// Per-execution metrics accumulated by one Execute stream, as measured by the
+// worker. Each field is cumulative for the stream. 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.
+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 the stream's first
+ // terminal event: finish, cancellation, or an execution error.
+ 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;
Review Comment:
There is also a wall time between last result batch sent - the data processing phase ends, and finish response sent.
Usually in spark udf case, the data processing ends would already trigger task completes, while the finish response is handled in the task completion listener (e.g., to collect metrics, and properly worker session cleanup / reuse).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57324][SQL] Surface external UDF terminal metrics [spark]
Posted by "haiyangsun-db (via GitHub)" <gi...@apache.org>.
haiyangsun-db commented on code in PR #59111:
URL: https://github.com/apache/spark/pull/59111#discussion_r4146379297
##########
udf/worker/proto/src/main/protobuf/org/apache/spark/udf/worker/udf_message.proto:
##########
@@ -482,6 +479,55 @@ message CancelResponse {
optional ExecutionError error = 2;
}
+// Per-execution metrics accumulated by one Execute stream, as measured by the
+// worker. Each field is cumulative for the stream. 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.
+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 the stream's first
+ // terminal event: finish, cancellation, or an execution error.
+ 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;
Review Comment:
non-blocking: There is also a wall time between last result batch sent - the data processing phase ends, and finish response sent.
Usually in spark udf case, the data processing ends would already trigger task completes, while the finish response is handled in the task completion listener (e.g., to collect metrics, and properly worker session cleanup / reuse).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57324][SQL] Surface external UDF terminal metrics [spark]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang commented on code in PR #59111:
URL: https://github.com/apache/spark/pull/59111#discussion_r4222514831
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/externalUDF/ExecuteExternalUDFExecSuite.scala:
##########
@@ -396,6 +434,91 @@ 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") {
+ val reported = ExecutionMetrics.newBuilder()
+ .setBytesIn(0)
Review Comment:
**Nit (P3):** This test checks measured-zero only for `bytesIn`, which goes through `addIfPresent`. The two peak metrics go through `maxIfPresent`, and there the `metric.isZero` branch is the only path that records a reported 0. The nano-timing metric starts at -1 and `value` reads 0 while unset, so `0 > metric.value` alone is false. No test reports `max_work_wall_nanos` or `max_work_cpu_nanos` as 0. If that branch were dropped, a measured-zero peak would show as unreported and every assertion here would still pass. Could this snapshot also set `.setMaxWorkCpuNanos(0)` (and/or the wall peak), with an assertion that the metric is `!isZero` and has value 0?
**Verification:**
- **Regression:** A snapshot whose max_work_* field is set to 0 leaves the peak metric non-isZero with value 0, and the assertion fails if the keep-max logic records a value only when it exceeds the current value.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57324][SQL][UDF] Surface external UDF terminal metrics [spark]
Posted by "Yicong-Huang (via GitHub)" <gi...@apache.org>.
Yicong-Huang closed pull request #59111: [SPARK-57324][SQL][UDF] Surface external UDF terminal metrics
URL: https://github.com/apache/spark/pull/59111
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org