Orchestration layer for the SearchFlow data pipeline.

Three DAGs manage the complete data pipeline:

Ingestion → Transformation → Reverse-ETL

airflow/
├── config/            # Airflow configuration
├── dags/
│   ├── ingestion_dag.py        # Raw data ingestion
│   ├── transformation_dag.py   # dbt run + test
│   ├── reverse_etl_dag.py      # Sync to Redis/Postgres
│   ├── training_dag.py          # Weekly ML model retraining
│   └── monitoring_dag.py        # Daily drift detection
└── plugins/           # Custom operators (if any)

File: dags/ingestion_dag.py

start → ingest_search_events → ingest_click_events → ingest_conversion_events → log_metrics → end

• Reads JSONL files from /data/raw/
• Loads to DuckDB raw.* tables
• Idempotent: Uses INSERT OR IGNORE with event_id as key

File: dags/transformation_dag.py

start → dbt_deps → dbt_run_staging → dbt_run_marts → dbt_test → end

• Runs dbt models in dependency order
• Executes 78 data quality tests
• Tags: transformation, dbt

File: dags/reverse_etl_dag.py

start → sync_user_segments → sync_recommendations → end

• Syncs mart_user_segments → Postgres CRM
• Syncs mart_recommendations → Redis cache

File: dags/training_dag.py

train_churn → train_sentiment → train_recommender

• Runs models via docker exec into ml-engine container
• MLflow tracking for all training runs
• SLA: 30 minutes per task

File: dags/monitoring_dag.py

check_drift → evaluate_drift → [trigger_retrain | skip_retrain]

• Evidently AI drift detection on hotel booking features
• Conditional retraining when drift_score > 0.3
• SLA: 20 minutes

# Via docker-compose
docker-compose exec airflow-scheduler airflow dags trigger searchflow_ingestion

# Via Make
make run-ingest
make run-transform
make run-reverse-etl

docker-compose exec airflow-scheduler airflow dags list
docker-compose exec airflow-scheduler airflow dags list-runs -d searchflow_ingestion

• URL: http://localhost:8080
• Username: admin
• Password: admin

# Check for import errors
docker-compose exec airflow-scheduler airflow dags list-import-errors

# View task logs
docker-compose logs airflow-scheduler | grep ERROR