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: [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]
