[Bug]: Spark Structured Streaming Runner memory leak: Datasets are persisted but never unpersisted in ParDoTranslatorBatch #40243

What happened?

In the Spark Structured Streaming runner, ParDoTranslatorBatch.java explicitly calls .persist() on intermediate datasets when a ParDo transform has multiple outputs (side outputs). This is correctly done to avoid re-evaluating the parent mapPartitions transformation multiple times when dividing the dataset by tuple tags. However, the runner never calls .unpersist() to free these datasets from Spark's storage memory.

This creates a significant memory leak. The materialized datasets will accumulate in Spark's block manager indefinitely until the Spark application terminates (or they are forcefully evicted by Spark's LRU cache, which degrades performance).

Code Pointers & Steps to Reproduce:

1. Navigate to runners/spark/src/main/java/org/apache/beam/runners/spark/structuredstreaming/translation/batch/ParDoTranslatorBatch.java (around line 138-146).
2. The code explicitly acknowledges this technical debt with a comment: // FIXME What's the strategy to unpersist Datasets / RDDs?
3. Immediately following the comment, allTagsDS.persist(storageLevel); is called to cache the dataset, but no corresponding unpersist() logic exists anywhere in the translation lifecycle.
4. To reproduce: Run a large Beam batch pipeline with the Spark Structured Streaming runner that contains several multi-output ParDo transforms. Monitor Spark's Storage memory tab in the Spark UI. The cached datasets will continue to grow without ever being released.

Impact:

• Causes Out-Of-Memory (OOM) errors and uncontrolled disk spilling in Spark clusters due to exhaustion of Spark's storage memory.
• Major performance degradation for large batch pipelines relying on the Spark Structured Streaming runner.

Proposed Solution:

• Implement a cleanup strategy/lifecycle hook for unpersisting intermediate datasets when they are no longer needed by downstream consumers in the Spark Structured Streaming execution plan.
• Alternatively, attach a Spark SparkListener or use a reference counting mechanism within the translation context to explicitly unpersist these datasets once all downstream stages consuming the side outputs have completed.

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• 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