Data Quality and Validation Questions

# pyspark (DataFrame) outline
from pyspark.sql import SparkSession, functions as F, Window

spark = SparkSession.builder.getOrCreate()

raw = spark.read.parquet("s3://bucket/events/")  # input partitioned by event_date if available

# canonical fields: event_id, event_ts (timestamp), ingest_ts (timestamp), payload, event_date
# add deterministic tiebreaker: hash of payload or event_id
raw = raw.withColumn("tie_hash", F.abs(F.hash(F.col("event_id"))) )

w = Window.partitionBy("event_id").orderBy(F.col("event_ts").desc(),
                                          F.col("ingest_ts").desc(),
                                          F.col("tie_hash").asc())

deduped = raw.withColumn("rn", F.row_number().over(w)).filter("rn = 1").drop("rn", "tie_hash")

# write idempotently using Delta MERGE to target table keyed by event_id
from delta.tables import DeltaTable
target_path = "s3://bucket/clean/events_delta/"

if not DeltaTable.isDeltaTable(spark, target_path):
    deduped.write.format("delta").partitionBy("event_date").mode("overwrite").save(target_path)
else:
    delta = DeltaTable.forPath(spark, target_path)
    (delta.alias("t")
         .merge(deduped.alias("s"), "t.event_id = s.event_id")
         .whenMatchedUpdate(condition = "s.event_ts > t.event_ts OR (s.event_ts = t.event_ts AND (s.ingest_ts > t.ingest_ts OR (s.ingest_ts = t.ingest_ts AND s.event_id < t.event_id)))",
                             set = {c: f"s.{c}" for c in deduped.columns})
         .whenNotMatchedInsertAll()
         .execute())

WITH normalized AS (
  SELECT
    users_id,
    email,
    TRIM(LOWER(email)) AS canonical_email,
    created_at
  FROM `project.dataset.users`
),

ranked AS (
  SELECT
    users_id,
    email,
    canonical_email,
    created_at,
    ROW_NUMBER() OVER (PARTITION BY canonical_email ORDER BY created_at DESC) AS rn,
    COUNT(1) OVER (PARTITION BY canonical_email) AS group_size
  FROM normalized
)

SELECT
  users_id,
  email,
  canonical_email,
  created_at,
  group_size
FROM ranked
WHERE rn = 1
  AND group_size > 1  -- keep only canonical emails that have potential duplicates
ORDER BY group_size DESC, canonical_email;