HyukjinKwon commented on code in PR #59246:
URL: https://github.com/apache/spark/pull/59246#discussion_r4223899927
##########
core/src/test/scala/org/apache/spark/api/python/BasePythonRunnerSuite.scala:
##########
@@ -121,41 +127,32 @@ class BasePythonRunnerSuite extends SparkFunSuite {
}
test("SPARK-59773: read worker metrics and leave the following sections
aligned") {
- val withExtra = metricsReport.dropRight(1) +
",\"futureMetric\":{\"value\":[1,2]}}"
+ val withExtra = metricsReport.dropRight(1) +
",\"futureMetric\":17,\"zeroMetric\":0}"
val stream = framedMetricsStream(withExtra)
- assert(BasePythonRunner.readWorkerMetrics(stream) == expectedMetrics)
+ val result = BasePythonRunner.readWorkerMetrics(stream)
+ assert(result == expectedMetrics ++ Map("futureMetric" -> 17L,
"zeroMetric" -> 0L))
checkEndMarkers(stream)
}
- test("SPARK-59773: worker metrics require all current fields") {
Review Comment:
**Nit (P3):** This test was the only coverage for the missing-field failure.
That check now lives in `requiredMetric` inside `handleMetricsData`
(PythonRunner.scala:895-903), and no test reaches it. The remaining tests call
`readWorkerMetrics`, which accepts any subset of fields. If `requiredMetric`
regressed to `getOrElse(name, 0L)`, every test would still pass, and
boot/init/total/processing times or spill bytes would silently become 0. Could
the six lookups move into a small `private[python]` helper on
`BasePythonRunner`? Then this test could drop each field in turn and assert the
`Missing Python worker metric: <name>` failure again.
**Verification:**
- **Regression:** For each of the six required fields, a report that omits
it is rejected with a SparkException whose message names the field.
- **Behavior:** A complete report still yields the same required values, so
boot, init, total and processing times and spill increments are unchanged.
##########
core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala:
##########
@@ -213,21 +213,12 @@ private[spark] object BasePythonRunner extends Logging {
}
}
- // Values used to update the existing Python SQL metrics and task spill
metrics.
- private[python] case class WorkerMetrics(
- bootTimestampMs: Long,
- initTimestampMs: Long,
- finishTimestampMs: Long,
- pythonExecutionDurationMs: Long,
- memoryBytesSpilled: Long,
- diskBytesSpilled: Long)
-
private lazy val workerMetricsMapper = new ObjectMapper()
.enable(JsonParser.Feature.STRICT_DUPLICATE_DETECTION)
.enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS)
- /** Read and validate worker metrics after METRICS_DATA; ignore additional
fields. */
- private[python] def readWorkerMetrics(stream: DataInputStream):
WorkerMetrics = {
+ /** Decode all numeric fields after METRICS_DATA without knowing their
metric names. */
Review Comment:
**Nit (P3):** This reads as if non-numeric or non-integral fields were
skipped, which is what the old 'ignore additional fields' doc promised. But the
loop below throws `Invalid Python worker metric: <key>` for any value that is
not an int64 integer, including `42.0` and `true`. Someone adding a string or
float diagnostic to the worker report would make every UDF task fail. Maybe
something like: `/** Decode every field after METRICS_DATA as an int64 by name;
any other value fails the report. */`
**Verification:**
- **Inspection:** The decoder's doc comment describes the same accepted
value domain and failure behavior that the decoder enforces.
##########
python/pyspark/sql/tests/arrow/test_arrow_python_udf.py:
##########
@@ -542,6 +548,103 @@ def tearDownClass(cls):
finally:
super().tearDownClass()
+ def test_phase_metrics(self):
+ @udf("long", useArrow=True)
+ def measured(value):
+ time.sleep(0.002)
+ return value + 1
+
+ constant = udf(lambda: 7, "long", useArrow=True)
+ phase_names = (
+ "pythonDataReadTime",
+ "pythonInputPreparationTime",
+ "pythonUDFExecutionTime",
+ "pythonOutputPreparationTime",
+ "pythonDataWriteTime",
+ )
+ with self.sql_conf(
+ {
+ "spark.sql.execution.arrow.maxRecordsPerBatch": "2",
+ "spark.sql.adaptive.enabled": "false",
+ }
+ ):
+ # Exercise sequential and concurrent evaluation, then an empty
worker task.
+ for num_rows, concurrency in ((5, 0), (3, 2), (0, 0)):
+ concurrency_conf = (
+ {"spark.sql.execution.pythonUDF.arrow.concurrency.level":
str(concurrency)}
+ if concurrency
+ else {}
+ )
+ with (
+ self.subTest(num_rows=num_rows, concurrency=concurrency),
+ self.sql_conf(concurrency_conf),
+ ):
+ if num_rows:
+ data = self.spark.range(num_rows, numPartitions=1)
+ else:
+ data =
self.spark.createDataFrame(self.sc.parallelize([], 1), "id long")
+ result = data.select(measured("id"), constant())
+ self.assertEqual(
+ [tuple(row) for row in result.collect()],
+ [(value + 1, 7) for value in range(num_rows)],
+ )
+ plan = result._jdf.queryExecution().executedPlan()
+ while plan.nodeName() != "ArrowEvalPython":
+ self.assertEqual(plan.children().size(), 1)
+ plan = plan.children().apply(0)
+ metrics = plan.metrics()
+
self.assertEqual(metrics.apply("pythonNumPipelinedTasks").value(), 0)
+
self.assertEqual(metrics.apply("pythonNumTimingReports").value(), 1)
+ self.assertEqual(
+ metrics.apply("pythonNumTimedBatches").value(),
(num_rows + 1) // 2
+ )
+ phases = [metrics.apply(name).value() for name in
phase_names]
+ self.assertTrue(all(value >= 0 for value in phases))
Review Comment:
**Nit (P3):** `value >= 0` also holds for a metric the worker never
reported, because a timing `SQLMetric` that was never updated reads 0. If
`collect_timing` were dropped from `read_udfs` (worker.py:1835-1838), or the
`input_timer`/`output_timer` scopes were removed, this test would still pass.
`ArrowStreamTimingTests` builds the serializer with `collect_timing=True`
itself, so it doesn't cover that wiring either. One deterministic option is to
assert `not metrics.apply(name).isZero()` for each phase on a non-empty task,
since any worker update clears `isZero`, even a value of 0. Then add a
patched-`perf_counter_ns` check that the input, UDF and output scopes each
accumulate time.
**Verification:**
- **Regression:** On the non-legacy Arrow-batched UDF path, each of the five
phase metrics is observed as reported by the worker for a non-empty task.
- **Behavior:** With a controlled clock, the input preparation, UDF
execution and output preparation phases each accumulate a positive duration and
exclude the work done between their scopes.
--
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]
