vishalmore90 commented on PR #40283: URL: https://github.com/apache/beam/pull/40283#issuecomment-6031039539
To safely fix this without breaking the default path, I've pushed an update with the following minimal changes: 1. **Made it Opt-In**: I reverted the default behavior back to `list(self.bag)`. For users who have large side inputs and suffer from OOMs, I introduced a new pipeline option in `DaskOptions` (`--dask_lazy_side_inputs`). 2. **Mitigated Deadlocks for the Opt-In Path**: When the lazy path is enabled, I now use Dask's `worker_client()` context manager (when `dask.distributed` is available). This explicitly tells Dask that this task will launch sub-tasks, preventing deadlocks by yielding the worker thread. 3. **Sequential Yielding**: The opt-in path retains the partition-by-partition sequential evaluation to keep the high-water memory mark low. This ensures that 99% of pipelines (which have small side inputs) keep their parallelism and performance, while providing a safe escape hatch for users hitting OOMs. Could you take another look? Let me know if you have any further feedback! -- 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]
