Design — pinky-airflow

Update date : 2026-05-31 13:45

Bridge between Airflow DAGs and Snowflake Task graphs — not a replacement for either.

---

Two integration patterns

Pattern 1 — Airflow → Snowflake (run and wait)

Airflow owns the schedule. It triggers a Snowflake Task graph and waits for completion.

Airflow DAG
└── SnowflakeTaskOperator   → EXECUTE TASK DB.SCHEMA.MY_DAG USING CONFIG = '{airflow context}'
└── SnowflakeTaskSensor     → polls COMPLETE_GRAPH_TASKS until SUCCEEDED or FAILED

Use case: multi-client pipelines in parallel (1 Airflow DAG = 1 client), cross-DAG dependencies, coordination between Snowflake Task graphs.

Pattern 2 — Snowflake → Airflow (event-driven trigger)

Snowflake detects an event and triggers an Airflow DAG for downstream steps that cannot run inside Snowflake (Tableau refresh, external API, notification).

Snowflake Task graph
└── SnowflakeDagTriggerSP → POST /api/v1/dags/{dag_id}/dagRuns  (Airflow REST API)
                           passes Task run metadata as DAG conf

Use case: Snowflake stream detects new rows → full transformation → trigger Airflow for the external notification or Tableau refresh.

---

Infrastructure context — EC2 Airflow (not MWAA)

Airflow runs on an EC2 t3.small (~15$/month, stopped when idle). MWAA costs ~350$/month regardless of usage — not viable for intermittent workloads.

EC2 startup/stop is automated via EventBridge + Lambda:

EventBridge cron(0 6 * * ? *)   → Lambda: EC2 start
EventBridge cron(0 22 * * ? *) → Lambda: EC2 stop

Elastic IP assigned to keep a stable address across restarts.

---

What Airflow does and does not do here

Airflow does	Airflow does NOT do
Schedule and trigger Snowflake Task graphs	Transform data
Coordinate between Snowflake Task graphs	Replace Snowflake Tasks for internal orchestration
Trigger downstream systems (Tableau, external API)	Run Snowpark code
Run 1 DAG per client for parallel execution	Own the data pipeline logic

Data transformation stays in Snowpark stored procedures inside Snowflake Tasks.

---

AWS architecture around Snowflake

External world
├── Files / partners    → MFT      → S3 → Snowflake Task (polling COPY INTO)
├── Emails with attachments → Lambda → S3 → Snowflake Task (polling)
├── Webhooks / external API → Lambda → S3 or direct Snowflake call

S3 (universal file hub)

Snowflake (compute & transformation hub)
├── Serverless Tasks  → internal orchestration
├── Snowpark Python   → transformation
└── Streamlit         → dashboards / apps

Airflow on EC2 (external orchestration only)
├── Meta DAGs across Snowflake Task graphs
├── Multi-client pipelines in parallel
└── Triggers for external systems

Rule: if Snowflake handles it natively, stay in Snowflake. AWS services only at the boundaries (ingestion, external events, external notifications).

---

Multi-client cost tracking

Each PinkyTaskOperator injects the Airflow context into the Snowflake Task via USING CONFIG. The SP reads it and sets QUERY_TAG — costs are then attributable per client from ACCOUNT_USAGE.QUERY_HISTORY.

SELECT PARSE_JSON(QUERY_TAG):client_id::STRING, SUM(CREDITS_USED_CLOUD_SERVICES)
FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY
WHERE QUERY_TAG IS NOT NULL
GROUP BY 1;