Airflow

You can manage and orchestrate data pipelines using the airflow CLI. Use this when the user asks about scheduling, DAG runs, pipeline monitoring, or workflow management.

For data queries, use the duckdb or snowflake skill. For transformations, use the dbt skill.

Commands

DAGs

Tasks

Connections & Variables

System

Add -o json to most commands for JSON output.

When to use

• •User asks about pipelines or DAGs → airflow dags list
• •User wants to run a pipeline → airflow dags trigger <dag_id>
• •User asks about a failed run → airflow dags list-runs, then airflow tasks log
• •User wants to schedule something → create a DAG file, then airflow dags unpause
• •User asks about connections → airflow connections list

Typical workflow

1. •

   See available pipelines:

   bash

   airflow dags list -o json
   

2. •

   Trigger a run:

   bash

   airflow dags trigger my_etl_pipeline
   

3. •

   Monitor progress:

   bash

   airflow dags list-runs -d my_etl_pipeline -o json
   

4. •

   Check task logs on failure:

   bash

   airflow tasks list my_etl_pipeline
   airflow tasks log my_etl_pipeline load_data 2026-02-19
   

5. •

   Clear and retry failed tasks:

   bash

   airflow tasks clear my_etl_pipeline -t load_data -s 2026-02-19 -e 2026-02-19
   

Creating a DAG

DAG files live in apps/airflow/dags/ (configured via AIRFLOW_DAGS_FOLDER).

The project ships with two sample DAGs:

Example (apps/airflow/dags/daily_etl.py):

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

with DAG(
    "daily_etl",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
    default_args={"retries": 1, "retry_delay": timedelta(minutes=5)},
) as dag:

    ingest = BashOperator(
        task_id="ingest_data",
        bash_command="clawdata data ingest-all",
    )

    transform = BashOperator(
        task_id="run_dbt",
        bash_command="clawdata dbt run",
    )

    test = BashOperator(
        task_id="test_dbt",
        bash_command="clawdata dbt test",
    )

    ingest >> transform >> test

After creating, unpause with airflow dags unpause daily_etl.

Error recovery

Environment

Notes

• •Airflow's scheduler must be running for scheduled DAGs to execute
• •Use airflow tasks test to run a task without recording it in the DB (useful for debugging)
• •The example DAG above integrates with clawdata — tying Airflow orchestration to dbt transforms
• •Add -o json for machine-readable output on most commands