data-quality-gate

Data Quality Gate

Add or extend validation checks that block bad data from propagating through the pipeline.

Inputs

• Stage name where the gate should run (e.g., customer_events)
• Check type: null_rate | range | referential | row_count | custom
• Column(s) affected
• Threshold or reference table (depending on check type)
• Severity: warn (log and continue) or fail (halt the pipeline)

Steps

1. Identify where to add the gate Gates run after ingestion (raw → staging) or after a transform (staging → marts). Most gaps are caught earliest; prefer adding checks at the earliest stage where the data is available.

2. Write the check

   Null-rate check:

   def check_null_rate(df, column, threshold=0.05):
       rate = df[column].isnull().mean()
       if rate > threshold:
           raise DataQualityError(
               f"{column} null rate {rate:.1%} exceeds threshold {threshold:.1%}"
           )
   

   Range check:

   def check_range(df, column, min_val, max_val):
       out_of_range = df[(df[column] < min_val) | (df[column] > max_val)]
       if len(out_of_range) > 0:
           raise DataQualityError(
               f"{column}: {len(out_of_range)} rows outside [{min_val}, {max_val}]"
           )
   

   Referential integrity check:

   def check_referential(df, fk_column, reference_df, pk_column):
       orphans = df[~df[fk_column].isin(reference_df[pk_column])]
       if len(orphans) > 0:
           raise DataQualityError(
               f"{fk_column}: {len(orphans)} rows with no matching {pk_column}"
           )
   

   Row-count sanity check:

   def check_row_count(df, min_rows, max_rows=None):
       n = len(df)
       if n < min_rows:
           raise DataQualityError(f"Only {n} rows; expected at least {min_rows}")
       if max_rows and n > max_rows:
           raise DataQualityError(f"{n} rows exceeds max {max_rows}")
   

3. Register the check in the stage's test suite Add the check to pipelines/ingestion/{stage_name}/tests/test_quality.py or the dbt schema YAML:

   # dbt schema
   - name: {column}
     tests:
       - not_null
       - dbt_utils.accepted_range:
           min_value: {min}
           max_value: {max}