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