A data engineering team is developing an Apache Beam streaming pipeline deployed on Google Cloud Dataflow. The pipeline ingests high-throughput user activity records from Pub/Sub, processes and enriches the records using custom transforms, and writes the output dataset to Cloud Spanner. During testing and transient worker retries, several pipeline workers fail with com.google.cloud.spanner.SpannerException: ALREADY_EXISTS errors, leading to stalled pipeline execution. Which pipeline mutation logic should the engineering team implement to resolve these write failures?

Construct mutations using the `INSERT_OR_UPDATE` builder (`newInsertOrUpdateBuilder`) when generating Spanner mutations in the pipeline: ### What `INSERT_OR_UPDATE` Mutation Logic Is In Apache Beam and Cloud Dataflow pipelines targeting **Cloud Spanner**, mutation objects represent individual row modifications. The `INSERT_OR_UPDATE` mutation (`Mutation.newInsertOrUpdateBuilder(...)` in the Java SDK or `SpannerInsertOrUpdate` in Python) is an **upsert** operation. It inserts a new record if the primary key does not exist or overwrites existing column values if a matching primary key is already present. ### How It Meets the Requirements - **Resolves `ALREADY_EXISTS` Failures:** Cloud Dataflow operates with **at-least-once** delivery guarantees across distributed workers during retries or bundle replays. When transient network faults or worker restarts occur, bundles of mutations may be re-sent to Cloud Spanner. - **Ensures Idempotent Writes:** Using `INSERT_OR_UPDATE` guarantees that replaying a batch of mutations succeeds seamlessly rather than throwing a duplicate key constraint violation (`ALREADY_EXISTS`). ### Key Advantages and Capabilities - **Idempotency:** Allows seamless retries of failed or unacknowledged write bundles without failing the pipeline. - **Simplified Pipeline Logic:** Eliminates the need for custom read-before-write checks or complex distributed state management. - **High Throughput:** Spanner handles upsert operations efficiently without requiring locking read transactions prior to writing. - **Reliability in Streaming:** Naturally accommodates streaming replays and overlapping window outputs. ### Why This Is Optimal Standard `INSERT` mutations fail immediately upon encountering an existing key. In a distributed processing framework like Dataflow where worker retries and at-least-once mutations are standard behavior, choosing `INSERT_OR_UPDATE` provides an idempotent sink pattern that aligns with distributed streaming resilience.