Spenserrrr commented on code in PR #59246:
URL: https://github.com/apache/spark/pull/59246#discussion_r4251432018
##########
python/pyspark/worker.py:
##########
@@ -2382,45 +2384,72 @@ def _evaluate_batch_udf(udf_func, rows):
return list(pool.map(lambda row: udf_func(*row), rows))
def func(split_index: int, data: Iterator[pa.RecordBatch]) ->
Iterator[pa.RecordBatch]:
- for input_batch in data:
- num_rows = input_batch.num_rows
-
- # --- Input: Arrow -> Python columns ---
- columns = [
- (
- [conv(v) for v in
ArrowTableToRowsConversion._to_pylist(col)]
- if conv is not None
- else ArrowTableToRowsConversion._to_pylist(col)
- )
- for col, conv in zip(input_batch.itercolumns(),
arrow_to_py_converters)
- ]
- if not columns:
- columns = [[_NoValue] * num_rows]
-
- # --- Process: evaluate each UDF row-by-row ---
- output_arrays = []
- for udf_func, offsets, zero_arg, arrow_return_type,
result_conv in udf_infos:
- rows = (
- [() for _ in range(num_rows)]
- if zero_arg
- else list(zip(*[columns[o] for o in offsets]))
- )
- results = _evaluate_batch_udf(udf_func, rows)
- verify_result_row_count(len(results), num_rows)
-
- # --- Output: Python -> Arrow ---
- converted = (
- [result_conv(r) for r in results] if result_conv is
not None else results
- )
- try:
- arr = pa.array(converted, type=arrow_return_type)
- except pa.lib.ArrowInvalid:
- arr = pa.array(converted).cast(
- target_type=arrow_return_type,
safe=runner_conf.safecheck
+ # Keep phase totals local so each metric is updated once instead
of per batch.
+ input_duration_ns = 0
+ udf_duration_ns = 0
+ output_duration_ns = 0
+ clock = time.perf_counter_ns
+ try:
+ for input_batch in data:
+ input_start_ns = clock()
+ # --- Input: Arrow -> Python columns ---
Review Comment:
Moved the input-section comment before clock() so it labels the interval
being timed. Thanks for catching this!
--
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]
