soda-core

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:

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 i

---

*Content truncated.*