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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
