Spenserrrr commented on code in PR #59246:
URL: https://github.com/apache/spark/pull/59246#discussion_r4252023581
##########
python/pyspark/eval_handlers/_pandas.py:
##########
@@ -66,6 +68,21 @@ def __init__(
self._return_schema = StructType(
[StructField("_%d" % i, rt) for i, (_, _, _, rt) in
enumerate(udfs)]
)
+ self._input_timer: ContextManager[None] = nullcontext()
+ self._udf_timer: ContextManager[None] = nullcontext()
+ self._output_timer: ContextManager[None] = nullcontext()
+
+ def set_worker_metrics(self, metrics: "WorkerMetrics") -> None:
+ super().set_worker_metrics(metrics)
+ # Bind scopes once per task; each batch reuses them without creating
new timers.
+ # These phases run on the main thread even when a reader thread
prefetches input.
+ # The callback timer includes profiling work when a profiler wraps the
UDF.
+ self._input_timer = metrics.measure("pythonInputConversionTime")
+ self._udf_timer = metrics.measure("pythonUDFExecutionTime")
+ self._output_timer = metrics.measure("pythonOutputConversionTime")
+ # Advertise timing support even when the task has no input batches.
+ metrics.set("pythonNumTimingReports", 1)
+ metrics.set("pythonNumTimedBatches", 0)
Review Comment:
Agreed. The handler-local metric collection was removed; the shared
`WorkerMetrics` collector is reset once per task in the worker, and modern
Arrow UDF phase timing is collected there. Resolving this older placement
thread.
##########
python/pyspark/eval_handlers/_pandas.py:
##########
@@ -75,22 +92,24 @@ def run(self, split_index: int, data:
"Iterator[pa.RecordBatch]") -> "Iterator[p
num_rows = input_batch.num_rows
# Input: Arrow -> pandas Series (struct columns become DataFrames).
- pandas_columns = ArrowToPandasConversion.to_pandas(
- input_batch,
- timezone=runner_conf.timezone,
- struct_in_pandas="dict",
- ndarray_as_list=False,
- prefer_int_ext_dtype=runner_conf.prefer_int_ext_dtype,
- df_for_struct=True,
- )
+ with self._input_timer:
+ pandas_columns = ArrowToPandasConversion.to_pandas(
+ input_batch,
+ timezone=runner_conf.timezone,
+ struct_in_pandas="dict",
+ ndarray_as_list=False,
+ prefer_int_ext_dtype=runner_conf.prefer_int_ext_dtype,
+ df_for_struct=True,
+ )
# Process: evaluate each UDF column-wise on pandas Series.
results = []
for udf_func, args_offsets, kwargs_offsets, return_type in
self._udfs:
- result = udf_func(
- *[pandas_columns[o] for o in args_offsets],
- **{k: pandas_columns[v] for k, v in
kwargs_offsets.items()},
- )
+ # Argument selection and return validation are outside the
callback timer.
+ args = [pandas_columns[o] for o in args_offsets]
+ kwargs = {k: pandas_columns[v] for k, v in
kwargs_offsets.items()}
Review Comment:
Agreed that UDF argument preparation belongs to input time. In the current
Arrow path, `rows` is built before `udf_start_ns`, so that work is counted as
input preparation. Arrow read/write are separate intervals, and inter-UDF loop
transitions are left out, so the three in-loop phases are not defined to sum to
the full worker wall time. Resolving this older handler-scope thread.
--
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]
