本文へ移動
cccskills
無料GitHub で公開

soda-core

Soda Core data quality — SodaCL checks (row_count, missing, invalid, duplicate, freshness, schema, reference, custom SQL), configuration.yml for PostgreSQL/Spark/ClickHouse/BigQuery, soda scan CLI, Airflow integration, dbt integration, alerting

日本語の概要は準備中です。原文の説明を表示しています。

ivanshamaev/de-agent-skillsスター 22(GitHub 22 + cccskills 0)
連携ツールGoogle CloudPostgreSQL

実行スコア 未検証

インストール方法を見る

含まれるファイル(1)

  • SKILL.md15.6 KB

SKILL.md(原文)

インストールする前に、エージェントに与えられる指示の中身を確認できます。

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:

CodeMeaning
0All checks passed
2One or more checks warned
3One or more checks failed
4Scan 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

  1. 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).

  2. 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.

  3. Hardcoding credentials in configuration.yml — use environment variables ${VAR} syntax everywhere; never commit passwords to git.

  4. 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.

  5. 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.

  6. 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).

  7. 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.

  8. 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

レビュー

まだレビューはありません。使ってみた感想をお寄せください。

コメント

ログインなしで書き込めます。使ってみた感想、質問、動かなかった環境などを気軽にどうぞ。

まだコメントはありません。最初のコメントを書いてみませんか。

送信するとすぐに公開されます。メールアドレスや電話番号などの個人情報は書かないでください。URL は 2 つまでです。

0 / 1000 文字

このフォームは reCAPTCHA で保護されており、Google のプライバシーポリシーと利用規約が適用されます。

同じリポジトリのスキル

AIOps autonomous incident response agent — LLM-driven diagnosis loop (Claude tool-use agent with kubectl/SQL/Prometheus tools), alert-to-action pipeline (PagerDuty webhook → agent trigger), automated RCA generation (failure taxonomy + log correlation), self-healing action executor with approval gate, incident severity classification, runbook automation (structured YAML runbooks executed by agent), escalation logic, incident timeline auto-generation, Slack integration for human-in-the-loop approvals

日本語の概要は準備中です。原文の説明を表示しています。

ivanshamaev/de-agent-skills222026年5月20日 更新

AIOps capacity planning agent — predictive resource forecasting (Prophet/linear regression on Prometheus metrics), Kubernetes HPA/VPA right-sizing recommendations, node capacity headroom calculator, Kafka partition and broker capacity model, Spark executor sizing from job history, data lake storage growth forecast (S3/GCS), cluster autoscaler simulation, automatic HPA threshold tuning, cost-per-job attribution, capacity planning report generator

日本語の概要は準備中です。原文の説明を表示しています。

ivanshamaev/de-agent-skills222026年5月20日 更新

AIOps infrastructure anomaly detection — statistical baseline (z-score/IQR/seasonal decomposition), Prometheus anomaly rules (predict_linear/stddev_over_time/MAD), ML-based detection (Isolation Forest/Prophet/LSTM autoencoder on metrics), metric correlation clustering, automated alert threshold tuning from historical data, Kubernetes resource anomaly detection, data pipeline anomaly signals (row count z-score/freshness drift/throughput drop), Python sklearn/Prophet integration, Grafana anomaly panel

日本語の概要は準備中です。原文の説明を表示しています。

ivanshamaev/de-agent-skills222026年5月20日 更新

AIOps observability copilot — natural language to PromQL/LogQL translation, alert explanation in plain English (what fired/why/impact), Grafana dashboard auto-generation from service topology, anomaly narrative generation from metric patterns, on-call context enrichment (related alerts/recent deploys/runbook links), log pattern clustering for noise reduction, SLO status explanation, automated weekly observability health digest

日本語の概要は準備中です。原文の説明を表示しています。

ivanshamaev/de-agent-skills222026年5月20日 更新

AIOps continuous platform optimization agent — autonomous optimization loop (observe→analyze→recommend→apply), Kubernetes resource rightsizing (VPA recommendation reader + apply), Kafka consumer lag auto-scale, Spark job configuration tuning from history, idle resource detection (unused deployments/PVCs/topics), compaction and VACUUM scheduling, auto-ANALYZE stale statistics, cost savings attribution, optimization audit trail, human approval gate for high-risk changes

日本語の概要は準備中です。原文の説明を表示しています。

ivanshamaev/de-agent-skills222026年5月20日 更新

AIOps autonomous query cost analyzer — Trino system.runtime.query_history cost SQL (CPU time/peak memory/bytes scanned), ClickHouse system.query_log top expensive queries, Spark UI cost attribution by user/job/DAG, BigQuery INFORMATION_SCHEMA.JOBS cost analysis, LLM-driven query rewrite recommendations (missing partition filters/broadcast hints/aggregation pushdown), cost anomaly detection (user over daily budget), automated query tagging, chargeback report by team/project

日本語の概要は準備中です。原文の説明を表示しています。

ivanshamaev/de-agent-skills222026年5月20日 更新

ivanshamaev のスキルをすべて見る

このスキルの問題を報告する