Data Quality Debugging and Root Cause Analysis Questions

# PySpark Structured Streaming pseudocode

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, expr, current_timestamp, window
from delta.tables import DeltaTable  # if using Delta

spark = SparkSession.builder \
    .appName("stream-late-detection") \
    .getOrCreate()

# Config
allowed_lateness = "10 minutes"       # allowed watermark threshold
late_cutoff_expr = f"current_timestamp() - interval {allowed_lateness}"

# Read stream (e.g., Kafka)
raw = spark.readStream.format("kafka") \
    .option("kafka.bootstrap.servers", "kafka:9092") \
    .option("subscribe", "events") \
    .load()

# Parse and select columns: event_id (unique), event_ts (timestamp), payload...
events = raw.selectExpr("CAST(value AS STRING) as json") \
    .selectExpr(
        "json_tuple(json, 'event_id') as event_id",
        "to_timestamp(json_tuple(json, 'event_ts')) as event_ts",
        "json_tuple(json, 'payload') as payload"
    ).withColumn("ingest_ts", current_timestamp())

# Compute cutoff for "late" detection (processing-time approximation)
cutoff = expr(late_cutoff_expr)

# Split streams
late_events = events.filter(col("event_ts") < cutoff)
on_time_events = events.filter(col("event_ts") >= cutoff)

# Idempotent write for late events using Delta MERGE (upsert by event_id)
# This avoids duplicates if retries happen.
def write_late_batch(df, batch_id):
    delta_path = "/mnt/warehouse/late_events_delta"
    if not DeltaTable.isDeltaTable(spark, delta_path):
        # initial write - create table
        df.dropDuplicates(["event_id"]).write.format("delta").mode("overwrite").save(delta_path)
    else:
        deltaTable = DeltaTable.forPath(spark, delta_path)
        # merge on event_id -> update or insert (idempotent)
        tmp = f"/tmp/late_batch_{batch_id}"
        df.dropDuplicates(["event_id"]).write.format("delta").mode("overwrite").save(tmp)
        batch_df = spark.read.format("delta").load(tmp)
        deltaTable.alias("t").merge(
            batch_df.alias("s"),
            "t.event_id = s.event_id"
        ).whenMatchedUpdateAll().whenNotMatchedInsertAll().execute()

late_query = late_events.writeStream \
    .foreachBatch(write_late_batch) \
    .option("checkpointLocation", "/checkpoints/late_events") \
    .start()

# On-time windowed aggregation using event-time watermark
agg = on_time_events \
    .withWatermark("event_ts", allowed_lateness) \
    .groupBy(window(col("event_ts"), "5 minutes"), col("some_group_key")) \
    .count()  # example aggregation

# Write aggregation to sink (e.g., Delta or Kafka), with checkpointing
agg_query = agg.writeStream \
    .outputMode("append") \
    .format("delta") \
    .option("path", "/mnt/warehouse/aggregates") \
    .option("checkpointLocation", "/checkpoints/aggregates") \
    .start()

spark.streams.awaitAnyTermination()

from pyspark.sql.functions import to_date, coalesce, col
df = raw_df.withColumn("date_iso",
    coalesce(
      to_date(col("date_str"), "dd/MM/yyyy"),
      to_date(col("date_str"), "MM-dd-yyyy"),
      to_date(col("date_str"), "yyyy-MM-dd")
    )
)