Approach: use a two-layer idempotency strategy:1) Batch-level guard: compute a stable batch_id (e.g., source offsets + date) and record it in a metadata table to short-circuit exact re-runs.2) Record-level dedupe + atomic upsert: use Delta Lake (or equivalent ACID store) and MERGE into the date-partitioned table on (partition_date, event_id). Write happens in a transaction so concurrent attempts merge deterministically and retries are safe.Pseudocode (PySpark + Delta semantics):python
from pyspark.sql import SparkSession
from delta.tables import DeltaTable
import uuid
spark = SparkSession.builder.appName("ingest").getOrCreate()
# Inputs
raw_path = "gs://bucket/raw/events/2025-11-22/*"
target_table = "warehouse.events_delta" # partitioned by event_date
meta_table = "warehouse.ingest_metadata" # records processed batch_ids
# 1. derive batch_id deterministically from source (e.g., min/max offsets, file list)
file_list = list_source_files(raw_path) # implementation detail
batch_id = deterministic_hash(file_list) # stable for the same inputs
# 2. Short-circuit if already processed
meta_df = spark.table(meta_table).filter(f"batch_id = '{batch_id}'")
if meta_df.count() > 0:
# batch already applied — idempotent exit
exit(0)
# 3. Read raw events
raw_df = spark.read.json(raw_path) \
.withColumn("event_id", ...) \
.withColumn("event_date", to_date(col("event_timestamp")))
# 4. Lightweight validation / canonicalization
clean_df = raw_df.select("event_id", "event_date", "payload", "ingest_ts") \
.dropna(subset=["event_id", "event_date"])
# 5. Optional: dedupe within batch (keep last by timestamp)
from pyspark.sql import Window
w = Window.partitionBy("event_id").orderBy(col("ingest_ts").desc())
batch_df = clean_df.withColumn("rn", row_number().over(w)).filter("rn = 1").drop("rn")
# 6. Upsert into delta table partitioned by event_date using MERGE (atomic)
delta_table = DeltaTable.forName(spark, target_table)
(
delta_table.alias("t")
.merge(
batch_df.alias("s"),
"t.event_date = s.event_date AND t.event_id = s.event_id"
)
.whenMatchedUpdateAll() # update if duplicate (idempotent)
.whenNotMatchedInsertAll()
.execute()
)
# 7. Record successful batch_id in metadata table (in same transaction if supported)
# If same-transaction is not possible, write meta record and ensure idempotency of meta write
meta_record = spark.createDataFrame([(batch_id, current_timestamp())], ["batch_id","applied_ts"])
meta_record.write.format("delta").mode("append").saveAsTable(meta_table)
