AI-Native Orchestration, Docs & Cost

This module puts AI to work on the unglamorous majority of the data engineering job: authoring and operating DAGs, surviving on-call, paying down documentation debt, and answering for the warehouse bill. Every pattern here follows the same contract — AI drafts and diagnoses at machine speed, you verify with evidence before anything ships or runs.

The scenario — one Tuesday, run twice

The Tuesday without the patterns. Meena is the senior data engineer who owns the orders_platform Airflow deployment at a mid-size commerce company — 84 DAGs, dbt on Snowflake, a Kafka ingest edge.

The same Tuesday, with this module in place.

She ships the thing she actually planned to ship. Same engineer, same stack — the difference is that assembly, drafting, and correlation moved to the machine, and judgment stayed with her.

AI-assisted DAG authoring — spec in, reviewable DAG out

Start from a spec, not a blank file. Give the assistant the contract — sources, schedule, SLAs, failure policy — and demand modern Airflow 3 idioms, because assistants trained on a decade of tutorials will happily emit deprecated patterns:

Write an Airflow 3 DAG from this spec:
- Ingest s3://commerce-raw/orders/ (hourly parquet drops) → Snowflake RAW.ORDERS
- Then trigger dbt build for tag:orders; downstream DAGs consume via asset, not sensor
- SLA: land by :20 past the hour; retries: 3, exponential backoff; on failure: callback to #data-oncall
Requirements: asset-based scheduling (no ExternalTaskSensor), deferrable operators for the S3 wait,
task groups per stage, no top-level code that runs at parse time, idempotent loads keyed on file name.

The requirements block is where your seniority lives — it encodes the review checklist into the generation step. Tools: Astronomer's Astro IDE (Airflow-specialized assistant; "production-ready from the start" is the vendor's claim — your review is still the gate), Claude Code or Cursor working repo-scale against your existing DAG conventions, and Dagster with an assistant for asset-based stacks. The free path is OSS Airflow 3 plus a generic assistant fed your best existing DAG as a style exemplar.

Verification is a fixed pass, not a vibe:

• Parse-time costno API calls or DB reads at module import.
• Idempotencyof every load task.
• Catchup/backfill semanticscorrect for this DAG's schedule.
• Timezone handlingon the schedule itself.
• Retry policyon exactly the tasks that are safe to retry.

AI-generated DAGs fail review most often on the last two — "almost right" scheduling code is the most expensive kind.

LLM tasks inside DAGs — AI as a first-class operator

The second shift: LLM calls stop being a sidecar script and become orchestrated tasks with the same retries, SLAs, and observability as everything else. The Airflow Common AI provider (built on Pydantic AI, 20+ model providers) and the airflow-ai-sdk give you decorator-level operators:

from pydantic import BaseModel

class TicketLabel(BaseModel):
    category: str          # billing | shipping | product | abuse
    urgency: str           # p0 | p1 | p2
    summary: str

@task.llm(
    model="anthropic:claude-opus-4-8",
    output_type=TicketLabel,
    system_prompt="Classify the support ticket. Choose only from the allowed values.",
)
def classify_ticket(ticket_text: str) -> TicketLabel:
    return ticket_text

@task
def load_labels(labels: list[TicketLabel]):
    ...  # validated rows → warehouse; rejects → review queue

Because the LLM call is a task, failure handling is orchestration, not application code: retries with backoff absorb rate limits, a dead-letter path catches poison inputs, and the task's duration and failure rate land in the same Airflow metrics you already watch. Dagster gets you the same shape with asset-based LLM steps; a plain PythonOperator wrapping the provider SDK is the zero-dependency fallback.

Two disciplines make this production-grade rather than a demo:

• Cost capscompute expected tokens per run (rows × avg tokens), assert against a budget before the map stage fans out, and fail the run, loudly, if a schema change upstream just 10×'d your input size.
• Output validation as a gate taskLLM outputs are untrusted input, so schema-validate and allowed-value-check them between the LLM task and the load task, never inside the same try block.

The failure-triage copilot — logs in, cited hypothesis out

On-call triage is assembly work — read the log, read the DAG, correlate with what changed — and assembly is what LLMs compress well. The pattern: a failure callback gathers context (the failing task's log tail, the DAG source, recent upstream schema/deploy changes, the last three incidents for this DAG) and asks for a structured brief:

from anthropic import Anthropic
from pydantic import BaseModel, Field

class TriageBrief(BaseModel):
    probable_cause: str
    confidence: float = Field(ge=0, le=1)
    evidence: list[str] = Field(description="Verbatim quotes from the provided logs/context")
    suggested_fix: str
    checks_before_fix: list[str]
    similar_past_incidents: list[str]

client = Anthropic()

response = client.messages.parse(
    model="claude-opus-4-8",
    max_tokens=2000,
    thinking={"type": "adaptive"},
    system=(
        "You are a data-pipeline triage assistant. Diagnose ONLY from the provided context. "
        "Every claim in `evidence` must be a verbatim quote from it. "
        "If the context is insufficient, say so and list what to fetch next."
    ),
    messages=[{"role": "user", "content": context_blob}],
    output_format=TriageBrief,
)
brief = response.parsed_output   # post to Slack/on-call channel

The verbatim-evidence rule is the whole safety model: an uncited hypothesis is a guess you have to re-derive, while a quoted one takes thirty seconds to check against the log. The human workflow is fixed — verify the quotes are real, verify they support the causal claim, then act; the copilot proposes, the engineer disposes. Wire it as a on_failure_callback posting to Slack, or interactively via an MCP connection to the Airflow API from Claude Code; PagerDuty-class incident tools ship the managed version of the same loop.

Feed it structure, not just a log dump. Triage quality tracks context quality almost linearly: including the schema-change feed is what turns "cast error in task X" into "column added upstream at 23:50 overflows the cast" — the difference between a symptom and a cause.

Backfill planning — the runbook you never want to improvise

Backfills are where data engineers do irreversible things while tired, which makes them exactly the wrong task to plan by hand in a scratch file. Give the LLM the dependency graph and the parameters, and demand a runbook:

Draft a backfill runbook: reprocess orders_daily for 2026-04-28..2026-07-26.
Context attached: DAG dependency graph, table sizes, incremental keys, warehouse size.
Produce: (1) partition batching plan (chunk size + rationale), (2) execution order across
the 6 downstream tables with the reason for each edge, (3) cost estimate in bytes scanned
per batch, (4) blast-radius checks BEFORE start (consumers to notify, dashboards affected,
retention windows that make dates unrecoverable), (5) pause/resume and rollback points.

You verify the two things the model cannot know: whether each task is genuinely idempotent for re-runs (the runbook should assert it per table, and you confirm), and whether business logic changed mid-window — a backfill through a logic change silently rewrites history unless you pin the version. Airflow 3's backfill UI then executes the plan with bounded parallelism; the runbook goes in the PR alongside the fix, which is how backfills become reviewable instead of tribal.

Documentation, lineage briefs & semantic models — generate in bulk, verify what matters

Documentation debt is the highest-ROI target in this module because generation is nearly free and the verification burden is triageable. The pattern for a 412-column debt: a metadata-aware pass (dbt Copilot generating from project metadata, Databricks AI-generated comments in Unity Catalog, or catalog AI in Atlan/Collibra-class tools) drafts every description from column names, lineage, and query context — then humans review by tier, not uniformly. Machine-verifiable facts (types, sources, freshness) get spot checks; business-meaning columns (revenue_recognized, anything feeding finance) get a named domain owner's sign-off, and the review status is itself metadata. Measure coverage in CI so the debt cannot silently regrow.

The same generate-then-verify shape covers lineage-aware impact analysis. Before a breaking change, feed column-level lineage plus the proposed diff to the assistant and ask for a consumer brief: affected models, affected dashboards, suggested migration order, and a draft deprecation notice. The lineage graph is ground truth and the LLM is the synthesizer — it turns an hour of graph-walking into a reviewable document, and your verification is checking the brief against the graph, not rebuilding it.

Semantic models raise the stakes because they define what numbers mean. dbt Copilot can draft semantic models (measures, dimensions, joins) from your marts, Cube's assistant and Snowflake semantic views do the analogous job — but the acceptance test is non-negotiable: the generated metric must reproduce known-good report numbers before it merges. Pick three certified numbers from last quarter's reporting, compute them through the generated semantic layer, and require exact agreement; a semantic model that is "almost right" is worse than none, because it exports wrongness to every downstream consumer — including the text-to-SQL systems you'll meet in M8, whose accuracy is bounded by this curation.

Query, Spark & warehouse cost — AI triage, measured verdicts

Query tuning with AI follows a strict loop: measure, diagnose, change, re-measure. Feed the EXPLAIN output plus table statistics to the assistant — the plan alone hides the problem half the time:

Diagnose this query. Attached: EXPLAIN plan, row counts, partitioning/clustering,
current bytes scanned (2.1 TB) and cost per run.
Identify: full scans that pruning would avoid, join-order or skew issues,
materialization candidates. For each fix: expected bytes-scanned reduction and how to verify.

Typical catches are the classics — a filter cast that defeats partition pruning, a fact-to-fact join that wanted pre-aggregation, clustering keys that don't match the dominant predicate. In Meena's morning-standup case the diagnosis was the first kind: the BI extract wrapped the partition column in a function, the warehouse scanned all 2.1 TB on every refresh, and rewriting the predicate dropped the scan to 340 GB — an 84% reduction the assistant proposed but the before/after measurement proved. Implement the top recommendation only, then re-run and record bytes scanned; the measured delta, not the model's confidence, is the verdict. One recommendation at a time also keeps causality clean — apply three fixes at once and you no longer know which one worked, which matters when you roll the pattern across fifty queries. Databricks Genie Code does this in-workspace across notebooks and the SQL editor; a frontier assistant with the plan pasted in is the vendor-neutral equivalent. Spark jobs follow the same loop with different inputs: export stage metrics from the Spark UI (shuffle read/write, spill, task-duration skew) and the assistant proposes salting, broadcast hints, or AQE settings as a config diff — which you A/B on a bounded run before touching the nightly job.

Warehouse-level cost triage scales the loop to the account. Start from query history:

SELECT query_parameterized_hash,
       COUNT(*)                                   AS runs,
       SUM(bytes_scanned) / POW(1024, 4)          AS tb_scanned,
       SUM(credits_attributed_compute)            AS credits,
       MAX(query_text)                            AS example_query
FROM   snowflake.account_usage.query_history      -- BigQuery: INFORMATION_SCHEMA.JOBS
WHERE  start_time > DATEADD('day', -30, CURRENT_TIMESTAMP())
GROUP  BY 1
ORDER  BY credits DESC
LIMIT  50;

Hand the top-50 to the LLM and ask for a FinOps brief: cost drivers ranked, anti-patterns flagged (SELECT * extracts, unpruned scans, over-provisioned warehouses idling), a fix per driver with projected savings, and a chargeback narrative per team. The LLM's leverage here is pattern recognition across hundreds of queries plus turning findings into a document a VP will actually read; your leverage is knowing which "waste" is actually a contractual SLA. Dedicated cost platforms (SELECT/Vantage-class) productize the detection — the brief-writing pattern works on top of any of them, or on bare account_usage for free.

LLM token-cost engineering — the new line item you own

LLM spend inside pipelines is now a warehouse-bill-shaped problem, and the same FinOps discipline applies — with four levers that routinely cut the bill by 5–10× combined:

• Batch APIsanything that isn't interactive (bulk classification, enrichment, embedding, the M5–M6 document work) goes through the provider's batch tier at roughly 50% off synchronous pricing, with the 24-hour turnaround your DAG schedule absorbs for free.
• Prompt cachingwhen thousands of calls share a large fixed prefix (the same schema context, the same instruction block, the same document for multi-chunk work), cache it; cached reads price at a small fraction of fresh input tokens, and pipeline workloads are almost pathologically cache-friendly because you control call ordering.
• Model right-sizingroute by task difficulty: a small fast model for high-volume classification, the frontier model only for hard extraction and synthesis, and let your M3-style quality gates prove the small model is good enough on a sample before the backfill, not after.
• Token budgetinga per-run pre-flight (rows × avg tokens × price) asserted against a cap, plus per-pipeline cost attribution written to a log table, so LLM spend gets a chargeback story exactly like credits do.

The build-vs-buy corners: warehouse-native functions (Snowflake Cortex AISQL, BigQuery AI functions, Databricks ai_query()) price per-token in your existing bill and keep data inside the platform boundary — often the right answer for regulated data even at a unit-cost premium — while self-hosting on vLLM trades API spend for GPU ops and only wins at sustained volume. For India-based learners and teams, the whole module runs on the free path — OSS Airflow 3, BigQuery sandbox or Databricks Free Edition, batch-tier LLM calls sized to a few dollars per lab — with one paid assistant subscription (~₹1,700/month) as the single recurring cost.

Check yourself: Your on-call triage copilot returns a fluent, confident probable cause at 06:12. What single design rule keeps a wrong hypothesis from costing you the morning, and what does the human still do?

Where this lands in your week

This module is the operational spine of the track's first half: DAGs authored from specs and verified against a fixed checklist, LLM tasks running under orchestration discipline, a triage copilot that turns 06:12 pages into cited briefs, backfills executed from reviewable runbooks, documentation and semantic models generated in bulk and verified by tier, and both bills — credits and tokens — triaged with evidence. It anchors two labs. LAB-04 (AI-native Airflow): author an Airflow 3 DAG with AI assist, add an LLM classification task via the Common AI provider, and build the failure-triage prompt that diagnoses an injected failure from logs — artifact: the running DAG repo plus the triage-copilot template with a solved incident transcript. LAB-05 (warehouse cost & query tuning with AI): feed EXPLAIN plans and a synthetic query history to an LLM, implement its top three recommendations, and measure before/after bytes scanned — artifact: the tuning report with measured savings plus a reusable cost-triage prompt. Both artifacts are the kind that survive an interview: not "I used AI," but "here is the incident it triaged and the savings it found, with the evidence attached." Do this at work this week → pb-20 Author an Airflow 3 DAG from a Pipeline Spec, pb-22 Stand Up a Pipeline-Failure Triage Copilot for Data On-Call, and pb-25 Triage Warehouse Spend from Query History and Draft the FinOps Narrative.