vishalmore90 opened a new issue, #40282:
URL: https://github.com/apache/beam/issues/40282
### What happened?
In the experimental Python Dask Runner, the `DaskBagWindowedIterator` (which
handles iterators for `apache_beam.transforms.sideinputs.SideInputMap`)
iterates over a Dask `Bag` by wrapping it in a Python `list()`.
Calling `list(self.bag)` implicitly triggers a full `compute()` on the Dask
dataset. This blocking operation materializes the entirety of the side input
data into the client's local memory. For large side inputs, this completely
bypasses Dask's distributed memory management and results in an Out-Of-Memory
(OOM) crash, effectively bottlenecking the scalability of pipelines running on
Dask.
The code currently includes an explicit `FIXME` acknowledging this
proof-of-concept behavior, but it remains a silent, critical scalability flaw.
## Code Pointers / Steps to Reproduce
The issue is located in
`sdks/python/apache_beam/runners/dask/transform_evaluator.py` within the
`__iter__` method of the `DaskBagWindowedIterator` class (lines 91-96):
```python
class DaskBagWindowedIterator:
"""Iterator for `apache_beam.transforms.sideinputs.SideInputMap`"""
bag: db.Bag
window_fn: WindowFn
def __iter__(self):
# FIXME(cisaacstern): list() is likely inefficient, since it presumably
# materializes the full result before iterating over it. doing this for
# now as a proof-of-concept. can we can generate results incrementally?
for result in list(self.bag):
yield get_windowed_value(result, self.window_fn)
```
## Impact
Any Apache Beam pipeline using `DaskRunner` that relies on substantial side
inputs will crash with OOM errors as soon as the side input data surpasses the
available local RAM on the node where the iterator is evaluated. This severely
limits the `DaskRunner`'s ability to process real-world distributed datasets
and creates a harsh scalability ceiling.
## Proposed Solution
The evaluation should generate results incrementally rather than performing
a monolithic evaluation. Potential approaches:
1. **Partition-based iteration:** Utilize Dask's `.map_partitions` or
`.to_delayed()` to fetch and yield the underlying data partition-by-partition.
2. **Generators:** Instead of eager computation via `list()`, retrieve
delayed results asynchronously and yield them to allow the Python garbage
collector to free memory between partition iterations.
### Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
### Issue Components
- [x] Component: Python SDK
- [ ] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [ ] Component: IO connector
- [ ] Component: Beam YAML
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Infrastructure
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Prism Runner
- [ ] Component: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner
--
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]
