Yicong-Huang commented on code in PR #55552:
URL: https://github.com/apache/spark/pull/55552#discussion_r3170740296
##########
python/pyspark/worker.py:
##########
@@ -3609,12 +3588,93 @@ def process():
if hasattr(out_iter, "close"):
out_iter.close()
+ def pipelined_process():
+ """
+ Pipelined variant of process() that pre-fetches input batches in a
background
+ reader thread while the main thread computes the UDF and writes
output.
+ This allows input deserialization to overlap with UDF computation.
+ """
+ # Mark that pipelined mode is active so UDFs can verify the code
path.
+ os.environ["SPARK_PIPELINED_UDF_ACTIVE"] = "1"
+ import queue
+ import threading
+
+ queue_depth =
int(os.environ.get("SPARK_PIPELINED_UDF_QUEUE_DEPTH", "2"))
+ _SENTINEL = object()
+ input_queue = queue.Queue(maxsize=queue_depth)
+ reader_error = [None]
+ stop_event = threading.Event()
+
+ def _reader_thread():
+ try:
+ for batch in deserializer.load_stream(infile):
+ # Some serializers (e.g., ArrowStreamGroupSerializer,
+ # ArrowStreamAggPandasUDFSerializer) yield lazy
iterators
+ # that still read from infile. Materialize them here
so the
+ # main thread can consume them without touching infile.
+ if hasattr(batch, "__next__"):
+ batch = list(batch)
+ # Use timeout put so we can check stop_event
periodically.
+ # This prevents the reader from blocking forever if
the main
+ # thread stops consuming (e.g., due to UDF exception).
+ while not stop_event.is_set():
+ try:
+ input_queue.put(batch, timeout=1)
+ break
+ except queue.Full:
+ continue
+ if stop_event.is_set():
+ return
Review Comment:
when a UDF takes a long time to process, this loop will try to put a batch
into the input queue every second. I am not a fan of this design. why can't we
let the consumer thread notify the producer thread that you can put now?
--
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]
