agentsclimarketplace
Sign in

Airflow testing to orchestra

Skill orchestra-hq/orchestra-skills/skills/migrate-to-orchestra/skills/airflow-testing-to-orchestra

Use this skill when an Airflow DAG contains data quality or testing operators: SqlCheckOperator, SQLThresholdCheckOperator, GreatExpectationsOperator, SodaCheckOperator, DbtTestOperator, or any custom SQL assertion task. Triggers: any task with sql= parameter used for validation, any use of Great Expectations or Soda, any dbt test step.From its SKILL.md

Install
npx -y skills add orchestra-hq/orchestra-skills --skill airflow-testing-to-orchestra

Assembled from the repository path, not quoted from the project. Check it against their README if it does not work.

One thing to look at

  • 9 stars9 stars. Stars are a popularity signal and not a quality one, but at this level it is likely that nobody has read this closely except its author, and you would be relying on your own review.

SKILL.md

10.9 KB, as published. Nobody here has run it

Airflow Testing Operators → Orchestra Test Jobs

Overview

Airflow data quality checks are typically SqlCheckOperator (pass/fail based on SQL result), SQLThresholdCheckOperator (pass/fail with min/max bounds), or framework operators like GreatExpectationsOperator and SodaCheckOperator. Orchestra has native test job types for SQL-based checks across all major warehouses, with configurable error and warning thresholds.

For mapping the conn_id= on these check operators to the right Orchestra connection, see airflow-connections-to-orchestra.

Test Job Pattern

All Orchestra test integrations follow the same parameter pattern:

parameters:
  statement: "SELECT COUNT(*) FROM orders WHERE amount < 0"
  error_threshold_expression: "> 0"    # FAIL if any negative amounts
  warn_threshold_expression: "> 0"     # WARN at same threshold (or set looser)

The statement must return a single numeric value (count, sum, percentage). The result is compared against the threshold expressions. Both thresholds default to "> 0" — fail if any rows returned.

Available Test Integration Jobs

Integrationintegration_jobWarehouse
SNOWFLAKESNOWFLAKE_RUN_TESTSnowflake
GCP_BIG_QUERYGCP_BQ_RUN_TESTBigQuery
POSTGRESPOSTGRES_RUN_TESTPostgreSQL
DATABRICKSDATABRICKS_RUN_TESTDatabricks SQL
FABRIC_SYNAPSEFABRIC_SYNAPSE_RUN_DQ_TESTMicrosoft Fabric
SNOWFLAKESNOWFLAKE_SCHEMA_VALIDATIONColumn type/constraint checks
SNOWFLAKESNOWFLAKE_ANOMALY_DETECTIONML-based anomaly detection
DBT_COREDBT_CORE_EXECUTE with dbt test;dbt test framework

Airflow Operator Mapping

SqlCheckOperator → *_RUN_TEST

SqlCheckOperator passes if the SQL returns a truthy (non-zero, non-null) result; fails otherwise.

# Airflow
check_no_nulls = SqlCheckOperator(
    task_id="check_no_nulls",
    conn_id="snowflake_prod",
    sql="SELECT COUNT(*) FROM orders WHERE customer_id IS NULL",
)
# Passes if COUNT = 0, fails if COUNT > 0

# Orchestra — fails if any nulls found
task-001:
  integration: SNOWFLAKE
  integration_job: SNOWFLAKE_RUN_TEST
  name: check_no_nulls
  connection: snowflake_prod_12345
  parameters:
    statement: 'SELECT COUNT(*) FROM orders WHERE customer_id IS NULL'
    error_threshold_expression: '> 0'
    warn_threshold_expression: '> 0'
  depends_on: []
  condition: null
  tags: []

SQLThresholdCheckOperator → *_RUN_TEST with threshold expressions

# Airflow
threshold_check = SQLThresholdCheckOperator(
    task_id="check_row_count",
    conn_id="snowflake_prod",
    sql="SELECT COUNT(*) FROM daily_orders",
    min_threshold=1000,
    max_threshold=50000,
)

# Orchestra — warn if <1000, fail if >50000
# Note: Run two test tasks or use a single query that returns 0 within bounds
task-001:
  integration: SNOWFLAKE
  integration_job: SNOWFLAKE_RUN_TEST
  name: check_row_count
  connection: snowflake_prod_12345
  parameters:
    statement: |
      SELECT CASE
        WHEN COUNT(*) < 1000 THEN 1
        WHEN COUNT(*) > 50000 THEN 1
        ELSE 0
      END FROM daily_orders
    error_threshold_expression: '> 0'
    warn_threshold_expression: '> 0'
  depends_on: []

DbtTestOperator → DBT_CORE_EXECUTE

# Airflow
dbt_test = DbtTestOperator(
    task_id="dbt_test",
    dir="/opt/dbt",
    dbt_bin="/usr/local/bin/dbt",
)

# Orchestra — include test in commands string
task-001:
  integration: DBT_CORE
  integration_job: DBT_CORE_EXECUTE
  name: dbt_test
  connection: dbt_core_conn_12345
  parameters:
    commands: 'dbt test;'
    package_manager: PIP
    python_version: '3.12'
  depends_on: []

Or chain with the dbt run:

parameters:
  commands: 'dbt run; dbt test;'   # run then test in sequence

GreatExpectationsOperator → PYTHON_EXECUTE_SCRIPT

No native GE integration exists. Run GE via Python:

# scripts/run_ge_checkpoint.py
import os
import great_expectations as ge
from orchestra_sdk.orchestra import OrchestraSDK

client = OrchestraSDK(api_key=os.environ.get("ORCHESTRA_API_KEY"))

context = ge.get_context()
results = context.run_checkpoint(checkpoint_name="my_checkpoint")
passed = results.success
client.set_output("passed", passed)
client.set_output("failed_expectations", results.statistics.get("unsuccessful_expectations", 0))

task-001:
  integration: PYTHON
  integration_job: PYTHON_EXECUTE_SCRIPT
  name: run_ge_checkpoint
  connection: my_python_conn_12345
  parameters:
    command: 'python scripts/run_ge_checkpoint.py'
    python_version: '3.12'
    set_outputs: true
  depends_on: []

SodaCheckOperator → PYTHON_EXECUTE_SCRIPT

# scripts/run_soda_checks.py
import os
from soda.scan import Scan
from orchestra_sdk.orchestra import OrchestraSDK

client = OrchestraSDK(api_key=os.environ.get("ORCHESTRA_API_KEY"))

scan = Scan()
scan.set_data_source_name("snowflake_prod")
scan.add_configuration_yaml_file(file_path="soda_config.yml")
scan.add_sodacl_yaml_file(file_path="checks/orders.yml")
scan.execute()

client.set_output("passed", scan.get_error_logs_text() == "")
client.set_output("errors", len(scan.get_error_logs()))

task-001:
  integration: PYTHON
  integration_job: PYTHON_EXECUTE_SCRIPT
  name: run_soda_checks
  connection: my_python_conn_12345
  parameters:
    command: 'python scripts/run_soda_checks.py'
    python_version: '3.12'
    set_outputs: true
  depends_on: []

Schema Validation (Snowflake)

Orchestra has a dedicated schema validation job for Snowflake column type checks:

task-001:
  integration: SNOWFLAKE
  integration_job: SNOWFLAKE_SCHEMA_VALIDATION
  name: validate_schema
  connection: snowflake_prod_12345
  parameters:
    table_name: orders
    column_name: order_id
    expected_type: NUMBER
    is_nullable: false
    is_primary_key: true
    is_unique: true
  depends_on: []

Non-Blocking Tests with treat_failure_as_warning

For quality checks that shouldn't halt the pipeline:

task-001:
  integration: SNOWFLAKE
  integration_job: SNOWFLAKE_RUN_TEST
  name: soft_row_count_check
  connection: snowflake_prod_12345
  treat_failure_as_warning: true    # FAILED → WARNING; pipeline continues
  parameters:
    statement: 'SELECT COUNT(*) FROM orders WHERE created_at < CURRENT_DATE - 7'
    error_threshold_expression: '> 0'
    warn_threshold_expression: '> 0'
  depends_on: []

Before / After Example

Airflow DAG (before)

from airflow.providers.common.sql.operators.sql import SQLCheckOperator
from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator

with DAG("quality_pipeline") as dag:
    load = SnowflakeOperator(
        task_id="load_data",
        snowflake_conn_id="snowflake_prod",
        sql="INSERT INTO orders SELECT * FROM orders_staging",
    )
    check_count = SQLCheckOperator(
        task_id="check_row_count",
        conn_id="snowflake_prod",
        sql="SELECT COUNT(*) > 0 FROM orders WHERE load_date = CURRENT_DATE",
    )
    check_nulls = SQLCheckOperator(
        task_id="check_no_nulls",
        conn_id="snowflake_prod",
        sql="SELECT COUNT(*) = 0 FROM orders WHERE order_id IS NULL",
    )
    load >> check_count >> check_nulls

Orchestra YAML (after)

version: v1
name: quality-pipeline

pipeline:
  stage-load:
    tasks:
      load-data:
        integration: SNOWFLAKE
        integration_job: SNOWFLAKE_RUN_QUERY
        name: load_data
        connection: snowflake_prod_12345
        parameters:
          statement: 'INSERT INTO orders SELECT * FROM orders_staging'
        depends_on: []
        condition: null
        tags: []
    depends_on: []

  stage-quality:
    tasks:
      check-row-count:
        integration: SNOWFLAKE
        integration_job: SNOWFLAKE_RUN_TEST
        name: check_row_count
        connection: snowflake_prod_12345
        parameters:
          statement: 'SELECT COUNT(*) FROM orders WHERE load_date = CURRENT_DATE'
          error_threshold_expression: '= 0'    # fail if zero rows
          warn_threshold_expression: '= 0'
        depends_on: []
        condition: null
        tags: []

      check-no-nulls:
        integration: SNOWFLAKE
        integration_job: SNOWFLAKE_RUN_TEST
        name: check_no_nulls
        connection: snowflake_prod_12345
        parameters:
          statement: 'SELECT COUNT(*) FROM orders WHERE order_id IS NULL'
          error_threshold_expression: '> 0'    # fail if any nulls
          warn_threshold_expression: '> 0'
        depends_on: []
        condition: null
        tags: []
    depends_on: [stage-load]

Gotchas

  • SQL must return a single number — test queries must resolve to one row, one column. Wrap complex queries in a subquery with a COUNT or SUM.
  • Threshold expression syntax — a single-character comparator followed by a non-negative integer: "> 0", "= 0", ">= 100", "< 0.05", "!= 0", "<= 0". Confirmed live against Orchestra's /pipelines/schema validator: "== 0" (Python-style double-equals) is rejected with "Invalid threshold expression format" — use a single =, not ==, for equality. This is an easy mistake since == reads naturally and looks like a comparator.
  • error_threshold_expression vs warn_threshold_expression — set warn looser than error (e.g. error > 10, warn > 5). If both are the same, the task goes directly to FAILED with no WARNING state.
  • SqlCheckOperator semantics inversion — Airflow's SqlCheckOperator passes when the SQL returns truthy. Orchestra's test fails when the result matches the threshold. Invert your SQL: Airflow SELECT COUNT(*) > 0 FROM ... becomes Orchestra SELECT COUNT(*) FROM ... with error_threshold_expression: '= 0' (single =, not ==).
  • GreatExpectations / Soda — no native integration. Use PYTHON_EXECUTE_SCRIPT with set_outputs: true to expose pass/fail to downstream conditions.
  • treat_failure_as_warning: true — use this for non-blocking quality checks. The pipeline continues even if the test fails, but the run gets a WARNING status and alerts fire.

Adding Alerts

alerts:
  - name: quality-check-failed
    statuses: [FAILED, WARNING]
    destinations:
      - integration: SLACK
        destination: '#data-quality'
    custom_message: 'Data quality check failed — review before proceeding.'

References

Keep looking

Skills are one crate of 627,801. Ordering is by how many stacks a row turns up in, so the top of any crate is what has actually been picked rather than what has the most stars.