Mola-maker opened a new issue, #40519:
URL: https://github.com/apache/beam/issues/40519
### What happened?
The experimental Python DaskRunner silently drops values emitted by a valid
`DoFn.finish_bundle()`. A pipeline that buffers its input in `process()` and
yields `WindowedValue` objects in `finish_bundle()` completes successfully but
produces an empty collection. When the DoFn emits from both methods, only the
`process()` output survives.
Reproduced on current master `d71149317ebf46376968e16367d8c4f8f6e5b0ea` with
Python 3.12.14, Beam 2.78.0.dev, Dask/distributed 2026.8.0 and two local worker
processes on Ubuntu.
### Minimal reproduction
Save this as a Python script and run it with an installed Beam Python SDK
and the `dask` extra:
```python
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.runners.dask.dask_runner import DaskRunner
from apache_beam.transforms.window import GlobalWindow
from apache_beam.utils.windowed_value import WindowedValue
from distributed import LocalCluster
class BufferElements(beam.DoFn):
def start_bundle(self):
self.elements = []
def process(self, element):
self.elements.append(element)
return []
def finish_bundle(self):
for element in self.elements:
yield WindowedValue(element, 0, [GlobalWindow()])
if __name__ == "__main__":
with LocalCluster(n_workers=2, threads_per_worker=2,
processes=True, dashboard_address=None) as cluster:
options = PipelineOptions([
"--dask_client_address", cluster.scheduler_address,
"--dask_npartitions", "1",
])
pipeline = beam.Pipeline(runner=DaskRunner(), options=options)
_ = pipeline | beam.Create([1, 2, 3]) | beam.ParDo(BufferElements())
result = pipeline.run()
try:
print(result.wait_until_finish())
print(result.client.gather(result.futures))
finally:
result.client.close()
```
Expected: `DONE` followed by `[[1, 2, 3]]`.
Actual on unmodified master: `DONE` followed by `[[]]`.
### Root cause and intended fix
`apply_dofn_to_bundle()` in
`sdks/python/apache_beam/runners/dask/transform_evaluator.py` copies
`tagged_receivers.values` into `results` before calling
`invoke_finish_bundle()`. The output handler appends the finish-bundle values
afterward, but the returned list is already a separate copy.
I have verified a minimal fix that collects the results after
`invoke_finish_bundle()` and before `invoke_teardown()`, with two regression
tests for finish-only and mixed process/finish output. Both tests fail on the
original code. With the fix and freshly restarted workers, the full DaskRunner
suite reports 29 passed, 1 existing expected failure, and 5 subtests passed.
### Issue Priority
Priority: 2
### 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: Website
- [ ] Component: Infrastructure
--
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]
