tkaymak opened a new issue, #41198: URL: https://github.com/apache/beam/issues/41198
### What needs to happen? In streaming mode the Spark 4 Structured Streaming runner runs a shared upstream once per consuming branch. A DoFn whose output is consumed twice processes every element twice, so attempted metrics are multiplied and side effects repeat. Measured on master plus #40949 with `Create.of(1, 2, 3)` and a ParDo that increments a counter per element: - one consumer: 3 - two PAsserts, two leaves: 6 - two branches into one Flatten, one PAssert: 6 There are two causes. `StreamingEvaluationContext` starts one Structured Streaming query per leaf dataset, so every query reads the source and runs the shared transforms on its own. Inside one query, `PipelineTranslator` does not persist streaming datasets (`persist` on a streaming Dataset throws), so every branch recomputes its input. With #40949 this also covers each consumed output of a multi output ParDo. Candidate fixes are one query per pipeline instead of one per leaf, and a shuffle before a fan-out point so Spark's exchange reuse runs the shared part once per micro batch. This also matters for a Spark 4 portable runner on Structured Streaming. Part of #36841. ### Issue Priority Priority: 2 (default / most bugs should be filed as P2) ### Issue Components - [x] Component: Spark 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]
