This is an automated email from the ASF dual-hosted git repository.
jrmccluskey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 2a540512f76 Fix DaskRunner dropping finish_bundle outputs (#40519)
(#40520)
2a540512f76 is described below
commit 2a540512f760645707442a64fb05053d6bacc547
Author: mola <[email protected]>
AuthorDate: Tue Oct 13 00:13:20 2026 +0800
Fix DaskRunner dropping finish_bundle outputs (#40519) (#40520)
Co-authored-by: Mola-maker <[email protected]>
---
CHANGES.md | 1 +
.../apache_beam/runners/dask/dask_runner_test.py | 35 ++++++++++++++++++++++
.../runners/dask/transform_evaluator.py | 3 +-
3 files changed, 38 insertions(+), 1 deletion(-)
diff --git a/CHANGES.md b/CHANGES.md
index 487d4e67d6a..cbda51c7c60 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -89,6 +89,7 @@
## Bugfixes
+* (Python) DaskRunner now preserves outputs emitted by `DoFn.finish_bundle()`
instead of silently dropping them
([#40519](https://github.com/apache/beam/issues/40519)).
* (Python) BigQuery streaming inserts now reset the retry budget and backoff
for each batch instead of sharing them across a bundle
([#39995](https://github.com/apache/beam/pull/39995)).
* (Go) Fixed a data race on the Prism runner's artifact cache map in
JobServices ([#32656](https://github.com/apache/beam/issues/32656)).
* (Go) Fixed the harness leaking Data/State gRPC streams after the worker
stops, and a deadlock when Send returns EOF
([#40260](https://github.com/apache/beam/issues/40260)).
diff --git a/sdks/python/apache_beam/runners/dask/dask_runner_test.py
b/sdks/python/apache_beam/runners/dask/dask_runner_test.py
index e1e5a4403b4..160a8f3cc51 100644
--- a/sdks/python/apache_beam/runners/dask/dask_runner_test.py
+++ b/sdks/python/apache_beam/runners/dask/dask_runner_test.py
@@ -125,6 +125,41 @@ class DaskRunnerRunPipelineTest(unittest.TestCase):
pcoll = p | beam.Create(list(range(1, 11))) | beam.Map(double)
assert_that(pcoll, equal_to(list(range(2, 21, 2))))
+ def test_pardo_finish_bundle_outputs(self):
+ 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 window.WindowedValue(element, 0, [window.GlobalWindow()])
+
+ with self.pipeline as p:
+ pcoll = p | beam.Create([1, 2, 3]) | beam.ParDo(BufferElements())
+ assert_that(pcoll, equal_to([1, 2, 3]))
+
+ def test_pardo_process_and_finish_bundle_outputs(self):
+ class EmitAtProcessAndFinish(beam.DoFn):
+ def start_bundle(self):
+ self.elements = []
+
+ def process(self, element):
+ self.elements.append(element)
+ yield element
+
+ def finish_bundle(self):
+ for element in self.elements:
+ yield window.WindowedValue(-element, 0, [window.GlobalWindow()])
+
+ with self.pipeline as p:
+ pcoll = (
+ p | beam.Create([1, 2, 3]) | beam.ParDo(EmitAtProcessAndFinish()))
+ assert_that(pcoll, equal_to([1, 2, 3, -1, -2, -3]))
+
def test_create_map_and_groupby(self):
def double(x):
return x * 2, x
diff --git a/sdks/python/apache_beam/runners/dask/transform_evaluator.py
b/sdks/python/apache_beam/runners/dask/transform_evaluator.py
index dbf55a3cee0..43ac488b293 100644
--- a/sdks/python/apache_beam/runners/dask/transform_evaluator.py
+++ b/sdks/python/apache_beam/runners/dask/transform_evaluator.py
@@ -195,9 +195,10 @@ def apply_dofn_to_bundle(
for it in items:
do_fn_invoker.invoke_process(it)
+ do_fn_invoker.invoke_finish_bundle()
+
results = [v.value for v in tagged_receivers.values]
- do_fn_invoker.invoke_finish_bundle()
do_fn_invoker.invoke_teardown()
return results
