Data Validation for Analytics Questions

# pyspark
from pyspark.sql import functions as F
df = spark.table("prod_features")
stats = df.agg(
 F.mean("feat").alias("mean"),
 F.expr("percentile_approx(feat, 0.5)").alias("median"),
 F.count(F.when(F.col("feat").isNull(),1))/F.count("*").alias("missing_rate"),
 F.min("feat").alias("min"), F.max("feat").alias("max")
).collect()[0]
cardinality = df.select("feat").distinct().count()
# For distribution shift: compute histogram and compare via JS/KL/KS on sampled arrays

SELECT
  COUNT_IF(feat IS NULL)/COUNT(*) AS missing_rate,
  MIN(feat) AS min, MAX(feat) AS max,
  COUNT(DISTINCT feat) AS cardinality
FROM PROD.FEATURES_DAILY;

WITH ordered AS (
  SELECT
    *,
    LAG(occurred_at) OVER (PARTITION BY user_id, event_type ORDER BY occurred_at, ingestion_time) AS prev_occurred_at,
    LAG(event_id) OVER (PARTITION BY user_id, event_type ORDER BY occurred_at, ingestion_time) AS prev_event_id
  FROM raw_events
)
SELECT
  event_id,
  user_id,
  event_type,
  occurred_at,
  ingestion_time,
  CASE
    WHEN prev_occurred_at IS NOT NULL
      AND EXTRACT(EPOCH FROM (occurred_at - prev_occurred_at)) <= 1
    THEN TRUE ELSE FALSE
  END AS likely_duplicate_of_prev,
  prev_event_id
FROM ordered
WHERE
  -- optional: restrict to recent timeframe for performance
  occurred_at >= CURRENT_DATE - INTERVAL '7 days';

WITH ranked AS (
  SELECT
    *,
    ROW_NUMBER() OVER (
      PARTITION BY user_id, event_type, FLOOR(EXTRACT(EPOCH FROM occurred_at)/1)
      ORDER BY occurred_at, ingestion_time, event_id
    ) AS rn
  FROM raw_events
)
SELECT * EXCEPT(rn)
FROM ranked
WHERE rn = 1;