Analytics Infrastructure and Query Performance Questions

# Pseudocode (PySpark)
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, expr
spark = SparkSession.builder.appName("dedup-watermark").getOrCreate()

kafka_df = (spark.readStream.format("kafka")
  .option("kafka.bootstrap.servers","...").option("subscribe","topic")
  .option("startingOffsets","earliest").load())

events = (kafka_df.selectExpr("CAST(value AS STRING) as json")
  .selectExpr("from_json(json, 'event_id STRING, payload STRUCT<...>, event_ts TIMESTAMP') as t")
  .select("t.*"))

# 1) Watermark + dropDuplicates on event_id (keeps earliest within watermark)
deduped = (events
  .withWatermark("event_ts", "30 minutes")        # watermark bounds late data
  .dropDuplicates(["event_id", "event_ts"]))     # keeps one event_id per event_ts within watermark

def foreach_batch(df, epoch_id):
    # df is micro-batch deterministic subset; idempotent write using MERGE
    # Use Delta or Iceberg MERGE on primary key event_id and use event_ts to keep latest
    (df.createOrReplaceTempView("batch"))
    spark.sql("""
      MERGE INTO target_table t
      USING batch s
      ON t.event_id = s.event_id
      WHEN MATCHED AND s.event_ts > t.event_ts THEN UPDATE SET *
      WHEN NOT MATCHED THEN INSERT *
    """)
    # commit is atomic for Delta/Iceberg

(deduped.writeStream
  .foreachBatch(foreach_batch)
  .option("checkpointLocation","/mnt/checkpoints/dedup")
  .start())