Below are five data quality checks for a “paid” dataset derived from table orders with fields: oid (INT), uid (INT), pay_ts (TIMESTAMP), amt (DECIMAL). Each check includes its objective, rule, and an example SQL query (ANSI SQL) applicable to most data warehouses.

1. Paid definition compliance and required-field completeness

• Objective: Ensure the dataset strictly adheres to the “paid” definition and contains required values.

• Rule:

  • Include only rows where pay_ts IS NOT NULL.
  • amt must be positive.
  • oid and uid must be present (NOT NULL).

• Example SQL: SELECT SUM(CASE WHEN pay_ts IS NULL THEN 1 ELSE 0 END) AS violations_missing_pay_ts, SUM(CASE WHEN amt IS NULL OR amt <= 0 THEN 1 ELSE 0 END) AS violations_amt_nonpositive_or_null, SUM(CASE WHEN oid IS NULL THEN 1 ELSE 0 END) AS violations_missing_oid, SUM(CASE WHEN uid IS NULL THEN 1 ELSE 0 END) AS violations_missing_uid FROM orders WHERE pay_ts IS NOT NULL OR pay_ts IS NULL;

  -- Optional hard filter audit (ensures no unpaid rows leak into “paid” dataset) SELECT COUNT(*) AS unpaid_rows_present FROM orders WHERE pay_ts IS NULL;

2. Primary key uniqueness (duplicate detection)

• Objective: Prevent duplicate paid orders for the same oid.

• Rule: Each oid appears at most once in the paid dataset.

• Example SQL: SELECT oid, COUNT() AS dup_count FROM orders WHERE pay_ts IS NOT NULL GROUP BY oid HAVING COUNT() > 1;

  -- If duplicates exist, inspect conflicting values SELECT o.* FROM ( SELECT oid FROM orders WHERE pay_ts IS NOT NULL GROUP BY oid HAVING COUNT(*) > 1 ) d JOIN orders o USING (oid) ORDER BY o.oid, o.pay_ts;

3. Timestamp validity (temporal sanity)

• Objective: Validate pay_ts against reasonable temporal bounds to catch clock, timezone, or ingestion errors.
• Rule:
  • pay_ts must not be in the future beyond a small tolerance.
  • pay_ts must be later than a configured lower bound (e.g., system go‑live date).
• Example SQL (using a 5-minute future tolerance and a configurable lower bound): WITH params AS ( SELECT CURRENT_TIMESTAMP AS now_ts, TIMESTAMP '2020-01-01 00:00:00' AS lower_bound_ts ) SELECT COUNT(*) AS invalid_timestamps FROM orders o CROSS JOIN params p WHERE o.pay_ts IS NOT NULL AND (o.pay_ts > p.now_ts + INTERVAL '5' MINUTE OR o.pay_ts < p.lower_bound_ts);

4. Referential integrity of uid (if a user dimension exists)

• Objective: Ensure that uid in orders maps to a valid user record.

• Rule: For paid rows, uid must exist in the users dimension (or source of truth).

• Example SQL: SELECT COUNT(*) AS missing_users FROM orders o LEFT JOIN dim_users u ON o.uid = u.uid WHERE o.pay_ts IS NOT NULL AND u.uid IS NULL;

  -- If there is a known set of valid uids (e.g., from a lookup table), use that instead of dim_users.

5. Completeness and reconciliation of paid counts/amounts (source vs. curated)

• Objective: Detect loss or duplication between source events and the curated paid dataset.
• Rule:
  • For each date, paid counts and sums should reconcile within a defined tolerance against the authoritative source.
• Example SQL (assuming a source table orders_src contains raw payment events): WITH src AS ( SELECT CAST(pay_ts AS DATE) AS dt, COUNT() AS src_cnt, SUM(amt) AS src_amt FROM orders_src WHERE pay_ts IS NOT NULL GROUP BY CAST(pay_ts AS DATE) ), tgt AS ( SELECT CAST(pay_ts AS DATE) AS dt, COUNT() AS tgt_cnt, SUM(amt) AS tgt_amt FROM orders WHERE pay_ts IS NOT NULL GROUP BY CAST(pay_ts AS DATE) ) SELECT s.dt, s.src_cnt, t.tgt_cnt, s.src_amt, t.tgt_amt, (t.tgt_cnt - s.src_cnt) AS diff_cnt, (t.tgt_amt - s.src_amt) AS diff_amt FROM src s JOIN tgt t USING (dt) WHERE ABS(t.tgt_cnt - s.src_cnt) > 0 OR ABS(t.tgt_amt - s.src_amt) > 0;

Implementation notes:

• If orders_src or dim_users is not available, substitute the appropriate upstream/source-of-truth tables for reconciliation and referential integrity checks.
• Thresholds (future tolerance, lower bound dates, amount maxima) should be parameterized and aligned with business SLAs and known system constraints.
• These checks can be automated via scheduled jobs or embedded into a data quality framework (e.g., Great Expectations, dbt tests, or Airflow sensors) and should emit metrics to monitoring/alerting systems.

Below are five practical data quality checks tailored for the training dataset clicks with fields: sid (int), label (binary), age (int), last_ts (timestamp), ctr (double). Each check includes what to assert and an example implementation (PySpark) you can embed in a data validation job.

1. Schema and Mandatory Field Conformance

• Assertion:
  • Columns exist with expected types: sid:int, label:int (binary), age:int, last_ts:timestamp, ctr:double.
  • Mandatory fields are non-null: sid, label, last_ts.
• Rationale: Prevents ingestion of malformed records and ensures downstream transformations do not fail due to type or null issues.
• Example (PySpark):
  • Enforce schema on read: schema = StructType([ StructField("sid", IntegerType(), False), StructField("label", IntegerType(), False), StructField("age", IntegerType(), True), StructField("last_ts", TimestampType(), False), StructField("ctr", DoubleType(), True), ]) df = spark.read.schema(schema).parquet(".../clicks")
  • Null check: violations = df.filter( col("sid").isNull() | col("label").isNull() | col("last_ts").isNull() ).count() assert violations == 0, f"Nulls found in mandatory fields: {violations}"

2. Identifier Uniqueness and Duplicate Detection

• Assertion:
  • If sid is expected to uniquely identify training records, enforce uniqueness (no duplicate sids).
  • If duplicates can occur (e.g., multiple events per sid), flag the duplicate rate and apply a deterministic dedup rule (e.g., keep the latest last_ts).
• Rationale: Duplicate records distort distributions and can bias model training.
• Example (PySpark):
  • Detect duplicates: dup_df = df.groupBy("sid").count().filter(col("count") > 1) dup_rate = dup_df.count() / df.select("sid").distinct().count() assert dup_rate == 0, f"Duplicate sid rate: {dup_rate:.6f}"
  • Optional dedup (keep most recent): window = Window.partitionBy("sid").orderBy(col("last_ts").desc()) df_dedup = df.withColumn("rn", row_number().over(window)).filter(col("rn") == 1).drop("rn")

3. Domain and Range Constraints

• Assertion:

  • label ∈ {0, 1}
  • age ∈ [0, 120] (adjust threshold to business context)
  • ctr ∈ [0.0, 1.0]
  • last_ts ≤ current time (no future timestamps)

• Rationale: Catches impossible or out-of-contract values that indicate upstream defects or parsing issues.

• Example (PySpark): from pyspark.sql.functions import current_timestamp

  invalid = df.filter( (col("label").isin(0, 1) == False) | (col("age").isNotNull() & ( (col("age") < 0) | (col("age") > 120) )) | (col("ctr").isNotNull() & ( (col("ctr") < 0.0) | (col("ctr") > 1.0) )) | (col("last_ts") > current_timestamp()) ) assert invalid.count() == 0, f"Domain/range violations: {invalid.count()}"

4. Timeliness and Coverage (Freshness within Training Window)

• Assertion:

  • The dataset’s max(last_ts) is within the expected SLA window (e.g., within 24 hours of now for daily training).
  • The dataset covers the intended training period (e.g., min(last_ts) and max(last_ts) span expected window).

• Rationale: Ensures the model trains on timely and representative data; stale inputs degrade performance.

• Example (PySpark): from pyspark.sql.functions import max as spark_max, min as spark_min import datetime

  stats = df.agg(spark_min("last_ts").alias("min_ts"), spark_max("last_ts").alias("max_ts")).collect()[0] now = datetime.datetime.utcnow() freshness_ok = (now - stats["max_ts"]).total_seconds() <= 24 * 3600 # 24h SLA assert freshness_ok, f"Data not fresh. max(last_ts)={stats['max_ts']} UTC"

  Optional coverage check for a 7-day window

  coverage_ok = (stats["max_ts"] - stats["min_ts"]).days >= 7 assert coverage_ok, f"Insufficient coverage: {stats['min_ts']}..{stats['max_ts']}"

5. Missingness and Distribution Stability (Drift Detection)

• Assertion:

  • Missingness thresholds:
    • age null rate ≤ 1% (example)
    • ctr null rate ≤ 1% (example)
  • Distribution stability vs. a baseline snapshot (e.g., previous day/week):
    • label rate within expected band (e.g., 0.3–0.7 if historically balanced)
    • ctr mean/variance within tolerance
    • age distribution similarity (use PSI or KS test)

• Rationale: Detects upstream shifts and silent failures that maintain schema but change data characteristics.

• Example (PySpark): from pyspark.sql.functions import mean, stddev, count, when

  total = df.count() age_null_rate = df.filter(col("age").isNull()).count() / total ctr_null_rate = df.filter(col("ctr").isNull()).count() / total assert age_null_rate <= 0.01, f"age null rate too high: {age_null_rate:.4f}" assert ctr_null_rate <= 0.01, f"ctr null rate too high: {ctr_null_rate:.4f}"

  label_rate = df.filter(col("label") == 1).count() / total ctr_stats = df.select(mean("ctr").alias("ctr_mean"), stddev("ctr").alias("ctr_std")).collect()[0]

  Compare against stored baselines (example, load from a configuration table or file)

  baseline = {"label_rate": 0.5, "ctr_mean": 0.08, "ctr_std": 0.12, "label_rate_tol": 0.05, "ctr_mean_tol": 0.01, "ctr_std_tol": 0.02}

  assert abs(label_rate - baseline["label_rate"]) <= baseline["label_rate_tol"], "Label rate drift"

  assert abs(ctr_stats["ctr_mean"] - baseline["ctr_mean"]) <= baseline["ctr_mean_tol"], "CTR mean drift"

  assert abs(ctr_stats["ctr_std"] - baseline["ctr_std"]) <= baseline["ctr_std_tol"], "CTR std drift"

Notes for productionization:

• Run these checks as part of your pipeline’s validation stage and fail fast with clear incident logs.
• Persist metrics (null rates, duplicate rates, distribution stats) to a data quality table for trend analysis.
• Calibrate thresholds to historical data and business requirements; start with conservative bounds and refine over time.
• Where duplicates are expected, define and apply deterministic deduplication logic before training.