GitHub user jrt1899 created a discussion: Parallelism bottleneck in multi-UDF
workflows
**The problem**
In multiUDF Python workflows - UDF1 (n workers) → UDF2 (m workers), the most
performance gains come from increasing worker parallelism. But increasing
parallelism on both stages can leave downstream workers idle for a long time,
limiting the gains.
The cause: each UDF1 worker creates one independent buffer per UDF2 worker,
filled round-robin. With n UDF1 workers and m UDF2 workers, that's n × m
independent buffers total. Each buffer only ships once it reaches the
configured Data Transfer Batch (DTB) size, so in the worst case, the link needs
roughly n × m × DTB tuples produced before any data reaches UDF2.
Since this threshold scales with n × m, adding more workers to either stage
makes idle time worse, not better — parallelism ends up widening the gap before
data starts flowing downstream instead of shortening the job.
```
UDF1 worker A ──┬─[DTB]──> UDF2 worker 1
├─[DTB]──> UDF2 worker 2
└─ … m separate buffers
UDF1 worker B ──┬─[DTB]──> UDF2 worker 1 ← a different buffer
├─[DTB]──> UDF2 worker 2
└─ … m separate buffers
...
UDF1 worker n ──┬─[DTB]──> UDF2 worker 1 ← a different buffer
├─[DTB]──> UDF2 worker 2
└─ … m separate buffers
```
In our test workflow (CSV scan → UDF → UDF, 8 workers per stage), n = m = 8 and
DTB = 5,000 — meaning ~320,000 rows had to accumulate before anything moved
downstream.
**Design 1: Threshold-triggered early flush**
Texera already polls each worker roughly twice a second for live execution
stats. We extended that poll to also track unshipped data per buffer and force
an early release once it exceeds the configured threshold. This caused the two
UDF stages to overlap, cutting our test workload's (two-filter credit card
fraud detection) latency by 39.5% in the CPU-only environment.
This is a simple change, but it has a flaw: forcing an early release means
shipped batches are smaller than the configured DTB value, so what's configured
no longer matches what actually happens.
**Design 2: Least-loaded routing**
This addresses the root cause instead of patching around it, and preserves the
configured DTB. Instead of one buffer per (UDF1 worker, UDF2 worker) pair, each
UDF1 worker holds a single buffer. Once it fills to the configured DTB size, it
ships to whichever UDF2 worker currently has the smallest backlog (tuples
delivered but not yet processed, from the same poll).
This removes the n × m multiplier entirely — a UDF1 worker only needs to fill
one DTB-sized batch before its first shipment goes out, and routing adapts
automatically as the two stages' relative speeds shift during the run. This
brought the latency improvement of 34.3% compared to the baseline in the same
environment.
Code: https://github.com/jrt1899/texera/tree/least-loaded-routing
GitHub link: https://github.com/apache/texera/discussions/8519
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
