Soda Core Data Quality
When to Use
Activate this skill when the task involves:
- Writing SodaCL checks for data quality validation (nulls, ranges, freshness, schema, uniqueness)
- Setting up Soda Core configuration for PostgreSQL, Spark, ClickHouse, Trino, or BigQuery
- Running
soda scan from CLI or within Airflow DAGs
- Designing a data quality gate pattern: scan → fail pipeline on violation
- Integrating Soda with dbt (supplementing or replacing dbt tests)
- Writing custom SQL metric checks
- Configuring warn vs. fail thresholds
Core Architecture
┌──────────────────────────────────────────────────────────────┐
│ │
│ checks.yml configuration.yml │
│ (SodaCL checks) (data source connection) │
│ ↓ ↓ │
│ ┌─────────────────────────────────┐ │
│ │ soda scan │ │
│ │ translates SodaCL → SQL │ │
│ │ runs against data source │ │
│ │ returns: pass / warn / fail / error │ │
│ └───────────────┬─────────────────┘ │
│ ↓ │
│ ┌───────────────────────────────┐ │
│ │ Results │ │
│ │ • stdout + logs │ │
│ │ • Soda Cloud (optional SaaS) │ │
│ │ • Airflow task state │ │
│ └───────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘
Installation
# Core + specific adapter
pip install soda-core-postgres # PostgreSQL
pip install soda-core-spark-df # Spark DataFrames
pip install soda-core-trino # Trino
pip install soda-core-bigquery # BigQuery
pip install soda-core-clickhouse # ClickHouse
pip install soda-core-duckdb # DuckDB (dev/testing)
pip install soda-core-sqlserver # SQL Server
Configuration File (configuration.yml)
PostgreSQL
# soda/configuration.yml
data_source postgres_prod:
type: postgres
host: ${POSTGRES_HOST}
port: "5432"
username: ${POSTGRES_USER}
password: ${POSTGRES_PASSWORD}
database: analytics
schema: silver
Spark (via DataFrame connector)
data_source spark_local:
type: spark_df
# SparkSession is passed programmatically — no host config here
Trino
data_source trino_prod:
type: trino
host: trino.internal
port: 8080
username: ${TRINO_USER}
auth:
type: kerberos
catalog: iceberg
schema: silver
http_scheme: https
ClickHouse
data_source clickhouse_prod:
type: clickhouse
host: clickhouse.internal
port: 8123
username: ${CH_USER}
password: ${CH_PASSWORD}
database: silver
BigQuery
data_source bigquery_prod:
type: bigquery
account_info_json: ${BIGQUERY_CREDENTIALS}
auth_scopes:
- https://www.googleapis.com/auth/bigquery
project_id: my-gcp-project
dataset: silver
Test connection:
soda test-connection -d postgres_prod -c soda/configuration.yml
SodaCL Check Syntax
File Structure
# soda/checks/silver_orders.yml
checks for orders: # "for <table_name>"
- row_count > 0
- missing_count(order_id) = 0
- duplicate_count(order_id) = 0
- missing_percent(customer_id) < 1%
Multiple tables in one file:
checks for orders:
- row_count > 0
checks for customers:
- missing_count(email) = 0
Row Count
checks for orders:
- row_count > 0 # at least 1 row
- row_count > 1000 # volume check
- row_count between 10000 and 10000000 # range
Missing Values
checks for orders:
- missing_count(order_id) = 0 # no nulls
- missing_count(customer_id) = 0
- missing_percent(discount) < 5% # allow up to 5% null discounts
- missing_count(notes): # custom missing definition
missing values: ["N/A", "n/a", ""]
fail: when > 0
Duplicate / Uniqueness
checks for orders:
- duplicate_count(order_id) = 0 # strict uniqueness
- duplicate_count(order_id, order_date) = 0 # composite key
Numeric Range / Statistics
checks for order_items:
- min(quantity) >= 1 # no zero or negative quantities
- max(price) <= 99999.99
- avg(discount_percent) between 0 and 50
- sum(total_amount) > 0
- stddev(price) < 1000 # sanity check on variance
- invalid_percent(price): # custom invalid definition
valid min: 0
valid max: 999999
fail: when > 1%
Invalid Values
checks for orders:
- invalid_count(status):
valid values: [pending, processing, shipped, delivered, cancelled]
fail: when > 0
- invalid_count(country_code):
valid format: ISO 3166 alpha-2 # built-in format validator
fail: when > 0
- invalid_count(email):
valid regex: ^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$
fail: when > 0
- invalid_percent(order_date):
valid min: 2020-01-01
valid max: today
fail: when > 0
Freshness
checks for orders:
- freshness(created_at) < 24h # most recent row is < 24h old
- freshness(created_at) < 1d # synonym
- freshness(updated_at) < 2h:
warn: when > 1h
fail: when > 2h
Schema
checks for orders:
- schema:
fail:
when required column missing:
- order_id
- customer_id
- total
- created_at
when wrong column type:
order_id: bigint
total: numeric
created_at: timestamp
when wrong index: # column order in result set
order_id: 0
warn:
when forbidden column present:
- password
- ssn
Referential Integrity
# Ensure every order has a valid customer
checks for orders:
- values in (customer_id) must exist in customers (customer_id)
Custom SQL Metric
checks for orders:
- order_date_after_ship_date = 0:
order_date_after_ship_date query: |
SELECT COUNT(*)
FROM orders
WHERE order_date > ship_date
- negative_total_count = 0:
negative_total_count query: |
SELECT COUNT(*)
FROM orders
WHERE total < 0
- distinct_statuses > 0:
distinct_statuses query: |
SELECT COUNT(DISTINCT status) FROM orders
Warn vs. Fail Thresholds
checks for orders:
- row_count:
warn: when < 5000 # warn only
fail: when < 1000 # fail the scan
- missing_percent(email):
warn: when between 1% and 5%
fail: when > 5%
- freshness(created_at):
warn: when > 1h
fail: when > 24h
Variables and Filters
filter orders [daily]:
where: created_at >= CURRENT_DATE - INTERVAL '1 day'
AND created_at < CURRENT_DATE
checks for orders [daily]:
- row_count > 500
- missing_count(order_id) = 0
Filters allow running the same checks on subsets (e.g., yesterday's partition).
Running Scans
# Basic scan
soda scan -d postgres_prod -c soda/configuration.yml soda/checks/
# Scan single checks file
soda scan -d postgres_prod -c soda/configuration.yml soda/checks/silver_orders.yml
# Scan with variable
soda scan -d postgres_prod -c soda/configuration.yml \
soda/checks/daily_orders.yml \
-V batch_date=2024-03-15
# Verbose output
soda scan -d postgres_prod -c soda/configuration.yml soda/checks/ -v
# Send results to Soda Cloud
soda scan -d postgres_prod -c soda/configuration.yml soda/checks/ \
--cloud-api-key-id $SODA_API_KEY_ID \
--cloud-api-key-secret $SODA_API_KEY_SECRET
Exit codes:
| Code | Meaning |
|---|
0 | All checks passed |
2 | One or more checks warned |
3 | One or more checks failed |
4 | Scan error (connection failure, SQL error) |
Python Programmatic API
import logging
from soda.scan import Scan
def run_soda_scan(
data_source_name: str,
table_name: str,
checks_yaml: str,
variables: dict | None = None,
) -> bool:
"""Run a Soda scan programmatically. Returns True if all checks passed."""
scan = Scan()
scan.set_verbose(True)
scan.set_scan_definition_name(f"soda_{table_name}")
scan.set_data_source_name(data_source_name)
scan.add_configuration_yaml_file("soda/configuration.yml")
scan.add_sodacl_yaml_str(checks_yaml)
if variables:
for key, value in variables.items():
scan.add_variables({key: value})
exit_code = scan.execute()
# Log all check results
for check_result in scan.get_scan_results()["checks"]:
status = check_result["outcome"]
name = check_result["name"]
logging.info(f"[{status.upper()}] {name}")
if exit_code == 0:
return True
elif exit_code == 2:
logging.warning("Soda scan completed with warnings")
return True # treat warn as pass (adjust to your policy)
else:
return False # fail = 3, error = 4
# --- Spark DataFrame scanning ---
from pyspark.sql import SparkSession
from soda.core.scan import Scan as SparkScan
def scan_spark_df(spark: SparkSession, df, table_alias: str, checks_yaml: str) -> bool:
scan = SparkScan()
scan.set_data_source_name("spark_local")
scan.add_spark_session(spark)
# Register the DataFrame as a temp view
df.createOrReplaceTempView(table_alias)
scan.add_sodacl_yaml_str(checks_yaml)
exit_code = scan.execute()
return exit_code in (0, 2)
Airflow Integration
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.hooks.base import BaseHook
from datetime import datetime
import os
def run_soda_quality_check(table: str, checks_file: str, data_source: str) -> None:
"""Runs soda scan and raises ValueError on failure to fail the Airflow task."""
from soda.scan import Scan
conn = BaseHook.get_connection("postgres_warehouse")
os.environ["POSTGRES_HOST"] = conn.host
os.environ["POSTGRES_USER"] = conn.login
os.environ["POSTGRES_PASSWORD"] = conn.password
scan = Scan()
scan.set_scan_definition_name(f"airflow_{table}_{datetime.utcnow().date()}")
scan.set_data_source_name(data_source)
scan.add_configuration_yaml_file("/opt/airflow/soda/configuration.yml")
scan.add_sodacl_yaml_file(f"/opt/airflow/soda/checks/{checks_file}")
exit_code = scan.execute()
if exit_code not in (0, 2): # 2=warn is acceptable
raise ValueError(
f"Soda data quality check FAILED for {table} "
f"(exit code {exit_code}). Pipeline halted."
)
with DAG(
dag_id="etl_with_quality_gates",
schedule="@daily",
start_date=datetime(2024, 1, 1),
) as dag:
transform_orders = PythonOperator(
task_id="transform_orders",
python_callable=run_transform,
)
check_orders = PythonOperator(
task_id="check_orders_quality",
python_callable=run_soda_quality_check,
op_kwargs={
"table": "orders",
"checks_file": "silver_orders.yml",
"data_source": "postgres_prod",
},
)
load_to_mart = PythonOperator(
task_id="load_to_mart",
python_callable=run_load,
)
# Gate: load only runs if quality check passes
transform_orders >> check_orders >> load_to_mart
dbt Integration
Soda can ingest dbt test results and push them to Soda Cloud, or run alongside dbt as a complementary quality layer.
Install
pip install soda-core-postgres soda-core-dbt
Run dbt Tests Then Ingest Results
dbt test --target prod
soda ingest dbt --target prod -d postgres_prod -c soda/configuration.yml
Supplement dbt Tests with SodaCL
Pattern: use dbt tests for schema/relationship integrity, use SodaCL for volume anomalies, freshness SLAs, and cross-table referential integrity that dbt can't easily express.
# soda/checks/post_dbt_silver_orders.yml
checks for silver.orders:
- freshness(created_at) < 2h: # SLA check dbt can't do
fail: when > 2h
- row_count > 100000 # volume anomaly
- values in (customer_id) must exist in silver.customers (customer_id)
Recommended File Layout
soda/
├── configuration.yml # data source connections
└── checks/
├── bronze/
│ ├── raw_orders.yml
│ └── raw_customers.yml
├── silver/
│ ├── orders.yml
│ └── customers.yml
└── gold/
└── fct_revenue.yml
Anti-Patterns
-
No warn threshold — only fail — a single rogue null fails the entire pipeline. Use warn for non-critical checks and reserve fail for blocking violations (null primary keys, referential integrity).
-
Scanning massive tables without filters — running full-table scans on multi-billion-row tables is expensive. Use filter blocks to restrict scans to recent partitions.
-
Hardcoding credentials in configuration.yml — use environment variables ${VAR} syntax everywhere; never commit passwords to git.
-
One giant checks file for all tables — hard to maintain and slow (scans all tables serially). Organize by layer and table; run per-table in parallel from Airflow.
-
Treating exit code 2 (warn) as failure in CI — warns indicate degraded quality, not critical failure. Only exit code 3 and 4 should fail the pipeline.
-
Not setting row_count > 0 as a baseline — empty tables are a common failure mode. Always add a minimum row count check before other checks (they run on an empty table and give misleading results).
-
Using custom SQL checks for everything — SodaCL built-in checks (missing_count, duplicate_count, etc.) are optimized and adapter-agnostic. Only use custom SQL for logic that can't be expressed with built-ins.
-
No alerting on scan failures — a failed scan that nobody sees is useless. Route Soda Cloud alerts to Slack/PagerDuty or configure Airflow callbacks on the quality gate task.
References to Consult When Needed
- SodaCL reference:
docs.soda.io/soda-cl/soda-cl-overview.html
- Built-in metrics:
docs.soda.io/soda-cl/numeric-metrics.html
- Configuration:
docs.soda.io/soda-core/configuration.html
- Airflow guide:
docs.astronomer.io/learn/soda-data-quality
- dbt integration:
docs.soda.io/soda/integrate-dbt.html