Gowtham Potureddi

Posted on Aug 6

Databricks Workflows vs Airflow: Orchestrating Inside vs Outside the Lakehouse

#python #sql #dataengineering #interview

databricks workflows vs airflow is the pick-one architecture decision that quietly determines whether your data platform is one coherent system or two systems held together with glue jobs and a prayer — and it is the decision senior data engineers get wrong most often because "we already use Airflow" and "Databricks has a scheduler built in" are both true, and both are the wrong place to stop thinking. Every pipeline your team runs — a nightly medallion refresh, a streaming ingestion loop, a cross-cloud reconciliation that touches S3, Snowflake, and a Spark cluster — has to be triggered on a schedule or an event, run its tasks in dependency order, retry the ones that fail, surface a clear failure signal to on-call, and reuse compute where it can. The engineering trade-off does not live in "should we orchestrate" — every non-trivial stack needs orchestration — but in where the orchestrator sits relative to the lakehouse, and what that placement costs you in reach, observability, and dollars.

This guide is the walkthrough you wished existed the first time an interviewer asked "compare databricks workflows vs airflow and tell me which you'd pick for a lakehouse-native team," or "your DAG needs to wait on an SFTP drop, load Snowflake, then run three Databricks notebooks — who owns that?", or "how does the airflow databricks operator avoid paying for a cluster per task?" It walks through the real inside-vs-outside framing: why the orchestration boundary is an architecture decision rather than a tool preference, how Databricks Workflows models databricks jobs as task DAGs with native triggers and ephemeral job clusters, how apache airflow models the same workflow orchestration problem as a Python dag with pluggable executors, the hybrid pattern where Airflow conducts and Workflows executes, and the decision matrix — cost, multi-system reach, observability, team skills — that turns the choice into a defensible one. Each section pairs a teaching block with a Solution-Tail interview answer: code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works.

When you want hands-on reps immediately after reading, drill the ETL practice library →, rehearse on the design practice library →, and sharpen the streaming axis with the streaming practice library →.

On this page

Why orchestration choice is an architecture decision
Databricks Workflows — jobs, tasks, dependencies, triggers
Apache Airflow — DAGs, operators, scheduler, executors
The hybrid reality — Airflow orchestrating Databricks jobs
Decision matrix — cost, DAGs, observability, skills
Cheat sheet — orchestration recipes
Frequently asked questions
Practice on PipeCode

1. Why orchestration choice is an architecture decision, not a tool preference

Inside vs outside the lakehouse — the placement of the orchestrator binds you for years

The one-sentence invariant: choosing between databricks workflows vs airflow is really choosing whether the orchestrator lives inside the lakehouse (Databricks Workflows co-located with the compute and the data it drives) or outside it (Apache Airflow as a standalone control plane that reaches into Databricks and every other system), and that placement trades workload-locality and cost against multi-system reach and portability in a way that cannot be undone cheaply. The orchestrator you pick in month one becomes the system every runbook, alert, and on-call rotation is wired around; migrating from one to the other is a quarter-long project because DAG definitions, retry semantics, alerting integrations, and lineage all have to be re-expressed in the other tool's model.

The trap is treating this as a features bake-off — "which one has better retries?" Both have retries. The real question is where the orchestration control plane should sit relative to your data gravity, and that is an architecture question with a right answer per organization, not a universal winner.

The four axes interviewers actually probe.

Workload locality. How much of your work runs on Databricks? If 95% of tasks are Databricks notebooks, Spark jobs, and DLT pipelines, an external orchestrator is an extra network hop and an extra system to operate for very little reach benefit — Workflows runs the work where the data already lives. If half your tasks are on Snowflake, dbt Cloud, Fivetran, and a Kubernetes microservice, an in-lakehouse orchestrator cannot reach them natively.
Multi-system reach. Airflow's provider ecosystem (hundreds of operators and hooks) is its core advantage — it speaks S3, GCS, Snowflake, BigQuery, dbt, Spark, HTTP, SFTP, and Databricks through a uniform DAG. Databricks Workflows reaches outside the lakehouse only through generic tasks (a notebook that shells out) — it is deliberately lakehouse-first.
Operational ownership. Who runs the orchestrator at 3 AM? Workflows is fully managed by Databricks — no scheduler to babysit, no metadata database to vacuum, no executor fleet to autoscale. Self-managed Airflow means you own the scheduler, the webserver, the metadata Postgres, and the executor workers (or you pay for MWAA / Astronomer / Composer to own them).
Cost model. Workflows on ephemeral job clusters bills Jobs Compute DBUs only while tasks run, then tears the cluster down. Airflow adds a standing infrastructure cost (scheduler + workers + database, always on) on top of whatever compute the tasks trigger — and if a naive Airflow DAG spins one Databricks cluster per task, the DBU bill multiplies.

The 2026 reality — three stable answers, not one winner.

Databricks Workflows is the default for lakehouse-native teams whose work is overwhelmingly Spark, SQL, notebooks, DLT, and dbt-on-Databricks. It ships with the platform, needs no extra infrastructure, and integrates with Unity Catalog lineage and Databricks alerting out of the box.
Apache Airflow is the default for platform teams orchestrating a heterogeneous fleet — where Databricks is one of many execution targets and the DAG must also touch Snowflake, Kafka, dbt Cloud, custom Python services, and cloud storage. Its Python-first DAG model and provider catalogue are unmatched for cross-system task dependencies.
Hybrid — Airflow as the enterprise conductor, Databricks Workflows as the in-lakehouse executor — is the most common answer at scale. Airflow owns the cross-system schedule and dependencies; it launches Databricks jobs (not one-off tasks) so the lakehouse work reuses one cluster and keeps its native lineage. This is covered in section 4.

What interviewers listen for.

Do you frame the choice as inside vs outside the lakehouse rather than "which tool has more features"? — senior signal.
Do you ask "what fraction of the workload runs on Databricks?" before answering? — required.
Do you name the standing-infrastructure cost of Airflow (scheduler + workers + metadata DB) as a real line item? — senior signal.
Do you name hybrid as a legitimate answer instead of forcing a single winner? — senior signal.
Do you describe orchestration as "trigger + dependency-ordered execution + retries + observability" rather than "the thing that runs my cron jobs"? — required.

Worked example — the four-axis comparison table

Detailed explanation. The single most useful artifact for a databricks workflows vs airflow interview is a memorised four-axis comparison. Every senior orchestration discussion converges on it within the first ten minutes; having it in your head is what separates a fluent answer from a stumbling one.

Workload locality. What share of tasks execute on Databricks compute versus elsewhere.
Multi-system reach. How many other systems the DAG must touch natively.
Operational ownership. Managed-by-vendor versus self-operated.
Cost model. Standing infra cost plus per-task compute versus compute-only.

Question. Build the four-axis comparison and state which axis is decisive for each of three teams: a pure-Databricks analytics team, a multi-cloud platform team, and a team that is 70% Databricks but must wait on external file drops.

Input.

Axis	Databricks Workflows	Apache Airflow
Workload locality	best when work is on Databricks	agnostic; any system
Multi-system reach	lakehouse-first; weak outside	hundreds of provider operators
Operational ownership	fully managed by Databricks	self-managed (or MWAA/Astronomer/Composer)
Cost model	Jobs Compute DBU only while running	standing infra + triggered compute

Code.

# A tiny decision helper that scores the two orchestrators per team profile.
def pick_orchestrator(pct_on_databricks: int,
                      external_systems: int,
                      wants_managed: bool) -> str:
    """Return the recommended orchestration control plane for a team."""
    # Heavily lakehouse-native and few external systems -> Workflows.
    if pct_on_databricks >= 90 and external_systems <= 1:
        return "Databricks Workflows"
    # Lots of external systems -> Airflow's provider reach wins.
    if external_systems >= 4:
        return "Apache Airflow"
    # In between: conduct with Airflow, execute lakehouse work in Workflows.
    return "Hybrid (Airflow conducts, Workflows executes)"


print(pick_orchestrator(95, 1, wants_managed=True))
# -> Databricks Workflows

print(pick_orchestrator(40, 6, wants_managed=False))
# -> Apache Airflow

print(pick_orchestrator(70, 3, wants_managed=True))
# -> Hybrid (Airflow conducts, Workflows executes)

Step-by-step explanation.

The pure-Databricks analytics team (95% on Databricks, one external system) short-circuits to Workflows — an external orchestrator would add a standing cost and a network hop for almost no reach benefit. The decisive axis is workload locality.
The multi-cloud platform team (40% on Databricks, six external systems) needs Airflow's provider catalogue — no in-lakehouse scheduler can natively sensor an SFTP drop, load Snowflake, call dbt Cloud, and then run a Databricks job in one DAG. The decisive axis is multi-system reach.
The 70%-Databricks team that waits on file drops is the classic hybrid case. Airflow owns the "wait for the external event, then decide" part; it launches a Databricks job for the lakehouse-heavy middle so that work reuses one cluster. The decisive axis is a blend of reach (the external wait) and cost (cluster reuse inside the lakehouse).
Notice the helper never returns "it depends" — it forces a decision from the constraints. That is exactly what an interviewer wants: a reproducible rule, not a shrug.
The wants_managed flag is deliberately unused in the naive helper to make a point in interview: managed-versus-self-hosted is usually a tie-breaker, not the primary axis. Reach and locality decide first; ownership decides between close calls.

Output.

Team profile	Decisive axis	Recommendation
95% Databricks, 1 external	workload locality	Databricks Workflows
40% Databricks, 6 external	multi-system reach	Apache Airflow
70% Databricks, waits on file drops	reach + cost	Hybrid

Rule of thumb. Never pick an orchestrator by feature count. Score the four axes — workload locality, multi-system reach, operational ownership, cost model — and let the constraints choose. Write the table on a whiteboard first; the answer falls out.

Worked example — what interviewers actually probe

Detailed explanation. The senior orchestration interview has a predictable shape: an ambiguous opener ("how would you schedule our pipelines?"), then progressive narrowing to test whether you understand the placement decision. Candidates who name the inside-vs-outside framing in sentence one score highest; candidates who name a tool by brand loyalty score lowest.

Ambiguous opener. "How would you orchestrate our data platform?" — invites you to name the framing, not a brand.
Follow-up 1. "Most of our work is on Databricks — does that change your answer?" — probes workload locality.
Follow-up 2. "We also load Snowflake and call dbt Cloud — now what?" — probes multi-system reach.
Follow-up 3. "Who operates it, and what does it cost?" — probes ownership and cost.
Follow-up 4. "How would you avoid one cluster per task?" — probes cluster reuse and the hybrid pattern.

Question. Draft a five-minute answer that covers all four axes without waiting to be asked.

Input.

Interview signal	Weak answer	Senior answer
Framing	"we'd use Airflow, it's standard"	"the question is whether orchestration lives inside or outside the lakehouse"
Locality	"Airflow can run anything"	"if 95% is on Databricks, Workflows removes a hop and a standing cost"
Reach	"we'd add operators"	"Airflow's provider catalogue is the reason it wins heterogeneous fleets"
Cost	"Databricks is expensive"	"Airflow adds standing infra; naive DAGs spin a cluster per task"
Reuse	"restart the job"	"launch a Databricks job / task-group so tasks share one cluster"

Code.

Senior orchestration answer template (5 minutes)
================================================

Minute 1 — name the framing
  "This is an inside-vs-outside-the-lakehouse decision, not a tool
   preference. Where should the orchestration control plane sit
   relative to the data gravity?"

Minute 2 — workload locality
  "If ~95% of tasks run on Databricks, I default to Workflows: it
   runs the work where the data lives, needs no extra infrastructure,
   and inherits Unity Catalog lineage. An external orchestrator buys
   almost no reach here and adds a standing cost."

Minute 3 — multi-system reach
  "The moment the DAG must also sensor an S3 drop, load Snowflake,
   and call dbt Cloud, Airflow's provider catalogue wins. Workflows
   is lakehouse-first; it does not natively reach that fleet."

Minute 4 — ownership + cost
  "Airflow is a standing system: scheduler, workers, metadata DB.
   Self-host or pay MWAA/Astronomer/Composer. Workflows is managed;
   ephemeral job clusters bill DBUs only while tasks run."

Minute 5 — the hybrid + cluster reuse
  "At scale I usually land on hybrid: Airflow conducts the cross-
   system schedule and dependencies, and launches a Databricks JOB
   (not one task at a time) so the lakehouse work reuses one cluster
   and keeps native lineage. That avoids the one-cluster-per-task
   DBU blowup."

Step-by-step explanation.

Minute 1 frames the decision as architecture, not brand preference. Naming "inside versus outside the lakehouse" immediately signals you are a decision-maker who understands data gravity, not someone reciting a tool they already know.
Minute 2 addresses locality before being asked. The senior move is to admit that when the work is overwhelmingly on Databricks, the external orchestrator is the one that must justify itself — it is not the automatic default.
Minute 3 is the reach probe. Naming the provider catalogue as Airflow's core advantage — and admitting Workflows is deliberately lakehouse-first — shows you understand each tool's design intent, not just its feature list.
Minute 4 is the ownership-and-cost argument. "Airflow is a standing system" is the line that separates people who have operated it from people who have only written DAGs. Naming MWAA / Astronomer / Composer as the managed escape hatch closes the loop.
Minute 5 pre-empts the reuse question with the hybrid pattern. Saying "launch a Databricks job, not a task at a time" is the senior signal — it shows you know the one-cluster-per-task cost trap and the fix.

Output.

Grading criterion	Weak score	Senior score
Names inside/outside framing	rare	mandatory
Asks about workload locality	rare	required
Names provider reach	occasional	mandatory
Names standing infra cost	rare	senior signal
Names hybrid + cluster reuse	rare	senior signal

Rule of thumb. The senior orchestration answer is a five-minute monologue that covers locality, reach, ownership, and cost without waiting for the follow-ups — and lands on hybrid as a legitimate destination, not a cop-out. Rehearse it once; deploy it every time.

Senior interview question on orchestration architecture

A senior interviewer often opens with: "You inherit a data platform where every pipeline is a hand-rolled cron entry on an EC2 box calling the Databricks REST API. Leadership wants 'real orchestration.' Half the tasks are Databricks notebooks; the other half touch Snowflake, an SFTP drop, and a dbt Cloud job. Walk me through how you'd decide between databricks workflows vs airflow, what you'd actually deploy, and how you'd stage the migration."

Solution Using a locality-scored decision plus a staged hybrid migration

# decision.py — score the platform and emit a staged migration plan.
from dataclasses import dataclass


@dataclass
class Platform:
    pct_on_databricks: int      # share of tasks running on Databricks
    external_systems: int       # count of non-Databricks systems in DAGs
    has_event_waits: bool       # sensors on file drops / external events
    team_owns_infra: bool       # willing to operate a scheduler fleet


def recommend(p: Platform) -> dict:
    if p.pct_on_databricks >= 90 and p.external_systems <= 1:
        plane = "Databricks Workflows"
    elif p.external_systems >= 4 or p.has_event_waits:
        plane = "Airflow conducts; Workflows executes lakehouse jobs"
    else:
        plane = "Databricks Workflows with a thin Airflow edge"

    return {
        "control_plane": plane,
        "phase_1": "Lift each cron entry into a Databricks Job (task DAG).",
        "phase_2": "Stand up managed Airflow (MWAA) for cross-system deps.",
        "phase_3": "Airflow launches the Databricks Jobs via run-now; "
                   "lakehouse tasks reuse one job cluster.",
    }


plan = recommend(Platform(pct_on_databricks=50,
                          external_systems=3,
                          has_event_waits=True,
                          team_owns_infra=False))
for k, v in plan.items():
    print(f"{k}: {v}")

# phase_1 target — the lakehouse half expressed as ONE Databricks job
# (Databricks Asset Bundle YAML; databricks bundle deploy)
resources:
  jobs:
    medallion_refresh:
      name: medallion_refresh
      tasks:
        - task_key: bronze_ingest
          notebook_task:
            notebook_path: /Repos/prod/etl/bronze_ingest
          job_cluster_key: shared
        - task_key: silver_transform
          depends_on:
            - task_key: bronze_ingest
          spark_python_task:
            python_file: /Repos/prod/etl/silver_transform.py
          job_cluster_key: shared
        - task_key: gold_marts
          depends_on:
            - task_key: silver_transform
          dbt_task:
            project_directory: /Repos/prod/dbt_marts
            commands: ["dbt run --target prod"]
          job_cluster_key: shared
      job_clusters:
        - job_cluster_key: shared        # ONE cluster for all three tasks
          new_cluster:
            spark_version: 15.4.x-scala2.12
            node_type_id: i3.xlarge
            num_workers: 4

Step-by-step trace.

Phase	Before (cron on EC2)	After (staged hybrid)
Lakehouse tasks	N cron calls to REST API	one Databricks job, one shared cluster
Cross-system deps	shell scripts, no visibility	Airflow DAG with sensors + operators
Retries	manual re-run	Workflows per-task retries + Airflow-level retry
Cluster spend	one cluster per REST call	one cluster reused across three tasks
Lineage	none	Unity Catalog lineage for the lakehouse job
Failure signal	a cron email, maybe	Airflow alert + Databricks job alert

After the migration, the lakehouse-heavy medallion refresh is one Databricks job whose three tasks share a single ephemeral cluster; Airflow (managed, so no scheduler to babysit) waits on the SFTP drop, loads Snowflake, triggers the Databricks job via run-now, then kicks the dbt Cloud job. The EC2 cron box is decommissioned.

Output:

Metric	Before	After
Systems operated	1 fragile EC2 box	managed Airflow + managed Workflows
Cluster starts per run	one per task (~N)	one per job (shared)
Cross-system visibility	none	single Airflow DAG view
Lakehouse lineage	none	native Unity Catalog
On-call signal	best-effort email	structured alerts both layers

Why this works — concept by concept:

Locality-scored decision — the recommendation is derived from where the work runs and how many systems it touches, not from tool familiarity. That is the reproducible framing an interviewer is grading for.
Phase 1: lift crons into a job — collapsing N REST calls into one Databricks job with depends_on gives dependency ordering, retries, and lineage for free, before any Airflow exists. It is the highest-leverage first step.
Shared job cluster — job_cluster_key: shared across three tasks means one ephemeral cluster spins up, runs the DAG, and tears down — instead of paying a start-up tax per task. This is the cost lever most teams miss.
Airflow as the conductor — Airflow owns only what the lakehouse cannot reach: the SFTP sensor, the Snowflake load, the dbt Cloud trigger, and the run-now handoff. It does not micromanage the lakehouse internals.
Cost — one standing (managed) Airflow control plane plus DBUs billed only while the shared cluster runs. Versus the old world (a cluster per REST call, no visibility), this is dramatically cheaper per run and O(1) clusters per job rather than O(tasks). The eliminated cost is the fragile hand-rolled cron layer and its silent failures.

ETL
Topic — etl
ETL problems on pipeline scheduling and dependencies

Practice →

Design Topic — design Design problems on orchestration architecture

Practice →

2. Databricks Workflows — jobs, task types, dependencies, triggers, native lakehouse integration

`databricks jobs` model a pipeline as a task DAG that runs where the data lives — ephemeral clusters, native triggers, Unity Catalog lineage

The mental model in one line: Databricks Workflows is the in-lakehouse orchestrator where a job is a DAG of tasks (notebook, Spark Python, Python wheel, SQL, dbt, DLT pipeline, another job, a for-each fan-out, a condition branch) wired together by depends_on, scheduled by native triggers (cron, file-arrival, table-update, or continuous), and executed on ephemeral job clusters that spin up, run, and tear down — so the orchestration control plane, the compute, and the governed data all live in one platform with zero extra infrastructure to operate. Every Databricks-native team already has this scheduler; the question is whether it reaches far enough for your fleet.

The task types — what a Databricks job can run.

Compute tasks. notebook_task (a Databricks notebook), spark_python_task (a .py file on Spark), python_wheel_task (an installed wheel entry point), spark_jar_task (a JVM main class). These are the workhorses of most jobs.
SQL + transform tasks. sql_task (a query, dashboard, or alert against a SQL warehouse), dbt_task (a dbt project run against Databricks). Native dbt support means the transform layer needs no external runner.
Pipeline task. pipeline_task triggers a Delta Live Tables (DLT) pipeline as a step in the job — declarative streaming/batch ingestion orchestrated by the same job.
Control-flow tasks. run_job_task (call another job — modular sub-DAGs), for_each_task (fan out a task over a parameter list, e.g. one run per region), condition_task (branch on a boolean, e.g. skip gold if silver row-count is zero).

The dependency model — depends_on builds the DAG.

Edges. Each task lists depends_on: [{task_key: ...}]; the job is the transitive closure of those edges. Cycles are rejected at deploy time.
Run-if semantics. A task's run_if controls whether it runs given upstream outcomes: ALL_SUCCESS (default), AT_LEAST_ONE_SUCCESS, NONE_FAILED, ALL_DONE, ALL_FAILED. This is how you build "run cleanup even if the main task failed."
Task values. dbutils.jobs.taskValues.set(key, value) in an upstream task and .get(taskKey, key) downstream passes small values along edges — the Databricks analogue of Airflow XComs.

The trigger types — how a job starts.

Scheduled (cron). A quartz cron expression with a timezone; the classic "run at 02:00 every day."
File-arrival trigger. The job fires when new files land at a Unity Catalog external location or volume path — event-driven ingestion without a polling sensor.
Table-update trigger. The job fires when one or more Delta tables receive an update — reactive downstream refresh.
Continuous. The job restarts itself as soon as a run finishes — the pattern for always-on streaming jobs. There is also a run-now REST/API entry point used by external orchestrators (this is the hook Airflow uses in section 4).

The compute model — job clusters vs all-purpose.

Job clusters (ephemeral). Declared per job; spun up at run start, torn down at run end. Billed at the cheaper Jobs Compute rate. One job_cluster_key can be shared by many tasks so they reuse the same cluster.
All-purpose clusters. Long-lived, interactive, billed at the higher rate. Fine for development; wasteful for production jobs. Interviewers flag "runs production jobs on an all-purpose cluster" as a cost anti-pattern.
Serverless. Serverless jobs compute removes cluster-config management entirely — Databricks provisions and scales the compute. It trades some control for zero cluster tuning.

Common interview probes on Databricks Workflows.

"How do tasks pass data?" — small values via task values; large data via tables in the lakehouse (not through the orchestrator).
"How do you avoid one cluster per task?" — share a job_cluster_key across tasks.
"How does a job react to new files without polling?" — file-arrival trigger.
"What's the lineage story?" — Unity Catalog captures table-level lineage for job tasks automatically.

Worked example — a medallion job in Databricks Asset Bundle YAML

Detailed explanation. The canonical Databricks job: a bronze ingest notebook, a silver transform in Spark Python, and a gold dbt run — three tasks, one shared cluster, a daily cron trigger, and per-task retries. Expressed as a Databricks Asset Bundle (DAB) so it is version-controlled and deployed with databricks bundle deploy.

Tasks. bronze_ingest → silver_transform → gold_marts, wired by depends_on.
Compute. One shared job cluster for all three tasks.
Trigger. Cron at 02:00 UTC daily.
Reliability. Two retries per task, email on failure.

Question. Write the DAB YAML for the medallion job with a shared cluster, a daily schedule, retries, and a failure notification.

Input.

Element	Value
Tasks	bronze_ingest, silver_transform, gold_marts
Dependencies	bronze → silver → gold
Cluster	one shared job cluster (i3.xlarge, 4 workers)
Trigger	quartz cron `0 0 2 * * ?` UTC
Retries	max 2 per task

Code.

# databricks.yml (excerpt) — deploy with: databricks bundle deploy -t prod
resources:
  jobs:
    medallion_daily:
      name: medallion_daily
      # --- native trigger: quartz cron with timezone ---
      schedule:
        quartz_cron_expression: "0 0 2 * * ?"
        timezone_id: "UTC"
        pause_status: "UNPAUSED"

      email_notifications:
        on_failure:
          - data-oncall@example.com

      tasks:
        - task_key: bronze_ingest
          notebook_task:
            notebook_path: /Repos/prod/etl/bronze_ingest
            base_parameters:
              run_date: "{{job.start_time.iso_date}}"
          job_cluster_key: shared_cluster
          max_retries: 2
          min_retry_interval_millis: 60000

        - task_key: silver_transform
          depends_on:
            - task_key: bronze_ingest
          spark_python_task:
            python_file: /Repos/prod/etl/silver_transform.py
            parameters: ["--layer", "silver"]
          job_cluster_key: shared_cluster
          max_retries: 2

        - task_key: gold_marts
          depends_on:
            - task_key: silver_transform
          # only build gold if silver produced rows (condition upstream would set this)
          run_if: ALL_SUCCESS
          dbt_task:
            project_directory: /Repos/prod/dbt_marts
            commands:
              - "dbt deps"
              - "dbt run --target prod"
            warehouse_id: ${var.sql_warehouse_id}
          job_cluster_key: shared_cluster
          max_retries: 1

      job_clusters:
        - job_cluster_key: shared_cluster
          new_cluster:
            spark_version: "15.4.x-scala2.12"
            node_type_id: "i3.xlarge"
            num_workers: 4
            data_security_mode: "SINGLE_USER"   # Unity Catalog access

# silver_transform.py — the Spark Python task body (runnable on the cluster)
import sys
from pyspark.sql import SparkSession, functions as F

def main(layer: str) -> None:
    spark = SparkSession.builder.getOrCreate()
    bronze = spark.read.table("prod.bronze.orders")
    silver = (
        bronze
        .where(F.col("status").isNotNull())
        .withColumn("total", F.col("total_cents") / 100.0)
        .dropDuplicates(["order_id"])
    )
    (silver.write
        .mode("overwrite")
        .option("overwriteSchema", "true")
        .saveAsTable("prod.silver.orders"))
    # pass the row count downstream as a task value
    from databricks.sdk.runtime import dbutils
    dbutils.jobs.taskValues.set(key="silver_rows", value=silver.count())

if __name__ == "__main__":
    # parameters: ["--layer", "silver"]
    layer = sys.argv[sys.argv.index("--layer") + 1]
    main(layer)

Step-by-step explanation.

The schedule block is a native trigger — a quartz cron expression plus a timezone. No external scheduler is involved; Databricks fires the job at 02:00 UTC. pause_status: UNPAUSED means it is live on deploy.
Three tasks are wired by depends_on: silver_transform lists bronze_ingest, and gold_marts lists silver_transform. That two-edge chain is the DAG; Databricks refuses to deploy a cycle.
Every task sets job_cluster_key: shared_cluster, and there is exactly one entry under job_clusters. This is the cost lever — one ephemeral cluster spins up, runs all three tasks in order, and tears down. Without the shared key, each task could provision its own cluster and triple the start-up tax.
Reliability is per task: max_retries: 2 on the ingest and transform, min_retry_interval_millis to back off, and email_notifications.on_failure for the on-call signal. The dbt task passes warehouse_id so it runs against a SQL warehouse.
The Spark task publishes silver_rows via dbutils.jobs.taskValues.set; a downstream condition_task (omitted for brevity) could read it and skip gold when the count is zero. This is how Databricks passes small control values along DAG edges — large data always moves through lakehouse tables, never through the orchestrator.

Output.

Task	Type	depends_on	Cluster	Retries
bronze_ingest	notebook_task	—	shared_cluster	2
silver_transform	spark_python_task	bronze_ingest	shared_cluster	2
gold_marts	dbt_task	silver_transform	shared_cluster	1

Rule of thumb. For any production Databricks job, define it as a DAB (version-controlled), share one job_cluster_key across tasks, set per-task retries, and wire a failure notification. The shared cluster is the single biggest cost decision — never let production tasks each spin their own.

Worked example — the file-arrival trigger for event-driven ingestion

Detailed explanation. A vendor drops CSV files into a Unity Catalog external location at unpredictable times. Instead of polling every five minutes with a sensor, a Databricks job uses a file-arrival trigger that fires when new files land — event-driven, no wasted polling compute.

Trigger. file_arrival watching a UC external location URL.
Debounce. min_time_between_triggers_seconds batches bursts of files into one run.
Idempotency. The ingest reads only files it has not processed (Auto Loader checkpointing).

Question. Configure a Databricks job whose trigger is file arrival at an external location, batching bursts and ingesting new files idempotently with Auto Loader.

Input.

Component	Value
Location	`s3://vendor-drop/orders/` (UC external location)
Trigger	file_arrival
Debounce	60 s between triggers
Ingest engine	Auto Loader (cloudFiles) with checkpoint

Code.

# File-arrival triggered job (DAB YAML)
resources:
  jobs:
    vendor_orders_ingest:
      name: vendor_orders_ingest
      trigger:
        pause_status: "UNPAUSED"
        file_arrival:
          url: "s3://vendor-drop/orders/"
          min_time_between_triggers_seconds: 60   # debounce bursts
          wait_after_last_change_seconds: 30      # let a drop settle
      tasks:
        - task_key: ingest_new_files
          notebook_task:
            notebook_path: /Repos/prod/etl/autoloader_ingest
          job_cluster_key: ingest_cluster
      job_clusters:
        - job_cluster_key: ingest_cluster
          new_cluster:
            spark_version: "15.4.x-scala2.12"
            node_type_id: "i3.xlarge"
            num_workers: 2

# autoloader_ingest notebook body — idempotent incremental read
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()

CHECKPOINT = "/Volumes/prod/etl/_chk/vendor_orders"

(spark.readStream
    .format("cloudFiles")                      # Auto Loader
    .option("cloudFiles.format", "csv")
    .option("cloudFiles.schemaLocation", CHECKPOINT + "/schema")
    .option("header", "true")
    .load("s3://vendor-drop/orders/")
    .writeStream
    .format("delta")
    .option("checkpointLocation", CHECKPOINT)  # only new files each run
    .trigger(availableNow=True)                # process backlog, then stop
    .toTable("prod.bronze.vendor_orders"))

Step-by-step explanation.

The trigger.file_arrival block replaces a schedule entirely — the job does not run on a clock, it runs when files appear at the watched external location. This eliminates the polling sensor and the compute it would burn checking an empty prefix all day.
min_time_between_triggers_seconds: 60 debounces bursts: if a vendor uploads 200 files in a minute, the job fires once, not 200 times. wait_after_last_change_seconds: 30 lets a multi-file drop finish before the run starts.
The ingest uses Auto Loader (format("cloudFiles")) with a checkpointLocation. Auto Loader tracks which files it has already ingested, so each triggered run processes only the newly arrived files — idempotent by construction, no manual watermark.
.trigger(availableNow=True) runs the stream in a batch-like mode: consume everything available now, then stop. Combined with the file-arrival trigger, this gives event-driven micro-batches without a permanently running stream.
The whole thing is lakehouse-native: the trigger, the compute, the Auto Loader checkpoint, and the target Delta table all live inside Databricks. An external orchestrator is not in the loop at all — this is Workflows at its strongest.

Output.

Event	Trigger fires?	Files processed
3 files land at 09:14	yes (after 30 s settle)	the 3 new files
burst of 200 files 09:20–09:20:40	once (debounced)	the 200 new files
no files for 6 hours	no	0 (no compute spent)
re-run after failure	yes	only unprocessed files (checkpoint)

Rule of thumb. For event-driven lakehouse ingestion, prefer a file-arrival trigger + Auto Loader over a polling sensor. You pay compute only when data actually arrives, and the checkpoint makes every run idempotent for free.

Worked example — for-each fan-out and a condition branch

Detailed explanation. A job must run the same transform for each of a dynamic list of regions, then run a global rollup only if at least one region succeeded. This needs for_each_task (fan-out) and condition_task (branch) — the two control-flow primitives that make Databricks jobs more than a linear chain.

Fan-out. for_each_task iterates a parameter list, running a nested task once per element with bounded concurrency.
Branch. condition_task compares two values and routes to different downstream tasks via run_if.
Rollup. Runs only when the branch says "at least one region has data."

Question. Model a job that fans a regional_load task over a list of regions, then runs global_rollup only if the loaded row count exceeds zero.

Input.

Element	Value
Regions	passed as a JSON array parameter
Fan-out	for_each over regions, concurrency 3
Condition	total_rows > 0
Rollup	runs on condition true

Code.

resources:
  jobs:
    regional_pipeline:
      name: regional_pipeline
      parameters:
        - name: regions
          default: '["us", "eu", "apac"]'
      tasks:
        # 1. Fan out the regional load over the list, 3 at a time
        - task_key: regional_load
          for_each_task:
            inputs: "{{job.parameters.regions}}"
            concurrency: 3
            task:
              task_key: load_one_region
              notebook_task:
                notebook_path: /Repos/prod/etl/load_region
                base_parameters:
                  region: "{{input}}"
              job_cluster_key: shared

        # 2. Gate: did any region load rows? (reads a task value)
        - task_key: check_rows
          depends_on: [{ task_key: regional_load }]
          condition_task:
            op: GREATER_THAN
            left: "{{tasks.regional_load.values.total_rows}}"
            right: "0"

        # 3. Rollup runs only when the condition is TRUE
        - task_key: global_rollup
          depends_on:
            - task_key: check_rows
              outcome: "true"
          notebook_task:
            notebook_path: /Repos/prod/etl/global_rollup
          job_cluster_key: shared

      job_clusters:
        - job_cluster_key: shared
          new_cluster:
            spark_version: "15.4.x-scala2.12"
            node_type_id: "i3.xlarge"
            num_workers: 4

# load_region notebook — accumulate a row count into a task value
from databricks.sdk.runtime import dbutils, spark

region = dbutils.widgets.get("region")
df = spark.read.table(f"prod.raw.orders_{region}")
n = df.count()
df.write.mode("append").saveAsTable("prod.silver.orders_all")

# for_each aggregates child task values; here we sum via a shared accumulator table
spark.sql(f"INSERT INTO prod.ops.load_counts VALUES ('{region}', {n})")
dbutils.jobs.taskValues.set(key="total_rows", value=n)

Step-by-step explanation.

for_each_task takes inputs (a JSON array from a job parameter) and a nested task template. Databricks runs the template once per element, substituting {{input}} for the region. concurrency: 3 bounds parallelism so three regions load at once, not all of them, protecting the cluster.
Each child load_one_region run reads that region's raw table and appends to a shared silver table. Passing region via base_parameters and reading it with dbutils.widgets.get is the standard parameter plumbing.
condition_task evaluates left op right — here total_rows GREATER_THAN 0. It does not run compute of its own beyond the comparison; it produces a boolean outcome that downstream tasks depend on.
global_rollup depends on check_rows with outcome: "true", so it runs only when at least one region produced rows. If every region was empty, the condition is false and the rollup is skipped — no wasted cluster time.
All child and parent tasks share one job_cluster_key, so the fan-out, the gate, and the rollup reuse a single ephemeral cluster. Fan-out plus branch plus shared cluster is the pattern for dynamic, cost-controlled lakehouse jobs.

Output.

Regions with rows	condition (rows > 0)	global_rollup
us, eu, apac all load	true	runs
only us loads	true	runs
all three empty	false	skipped
region list empty	false	skipped

Rule of thumb. Use for_each_task for dynamic fan-out with bounded concurrency, and condition_task to gate expensive downstream work on an upstream signal. Together they turn a linear job into a real DAG without leaving the lakehouse.

SQL interview question on Databricks Workflows

A senior interviewer might ask: "Design a Databricks-native pipeline that ingests vendor files the moment they arrive, runs a medallion transform, fans a mart build over a dynamic region list, and skips the global rollup when there is no new data — all on one shared cluster. Walk me through the job structure, the trigger, the control flow, and how you'd keep the DBU bill flat."

Solution Using a file-arrival-triggered job with shared cluster, for-each, and a condition gate

resources:
  jobs:
    vendor_medallion:
      name: vendor_medallion
      # Event-driven: fire on file arrival, not a clock
      trigger:
        pause_status: "UNPAUSED"
        file_arrival:
          url: "s3://vendor-drop/orders/"
          min_time_between_triggers_seconds: 120
      parameters:
        - name: regions
          default: '["us","eu","apac"]'
      email_notifications:
        on_failure: ["data-oncall@example.com"]
      tasks:
        - task_key: bronze
          notebook_task: { notebook_path: /Repos/prod/etl/autoloader_ingest }
          job_cluster_key: shared
          max_retries: 2

        - task_key: silver
          depends_on: [{ task_key: bronze }]
          spark_python_task: { python_file: /Repos/prod/etl/silver_transform.py }
          job_cluster_key: shared
          max_retries: 2

        - task_key: mart_by_region
          depends_on: [{ task_key: silver }]
          for_each_task:
            inputs: "{{job.parameters.regions}}"
            concurrency: 3
            task:
              task_key: mart_one
              dbt_task:
                project_directory: /Repos/prod/dbt_marts
                commands: ["dbt run --select tag:region --vars 'region: {{input}}'"]
                warehouse_id: ${var.sql_warehouse_id}
              job_cluster_key: shared

        - task_key: has_new_data
          depends_on: [{ task_key: silver }]
          condition_task:
            op: GREATER_THAN
            left: "{{tasks.silver.values.silver_rows}}"
            right: "0"

        - task_key: global_rollup
          depends_on:
            - task_key: mart_by_region
            - task_key: has_new_data
              outcome: "true"
          run_if: ALL_SUCCESS
          notebook_task: { notebook_path: /Repos/prod/etl/global_rollup }
          job_cluster_key: shared

      job_clusters:
        - job_cluster_key: shared
          new_cluster:
            spark_version: "15.4.x-scala2.12"
            node_type_id: "i3.xlarge"
            num_workers: 6
            data_security_mode: "SINGLE_USER"

# silver_transform.py — sets the gate value the condition_task reads
import sys
from pyspark.sql import SparkSession, functions as F
from databricks.sdk.runtime import dbutils

spark = SparkSession.builder.getOrCreate()
bronze = spark.read.table("prod.bronze.vendor_orders")
silver = bronze.dropDuplicates(["order_id"]).where(F.col("status").isNotNull())
silver.write.mode("overwrite").saveAsTable("prod.silver.orders")
dbutils.jobs.taskValues.set(key="silver_rows", value=silver.count())

Step-by-step trace.

Layer	Config	Result
Trigger	file_arrival, 120 s debounce	runs only when files land; no idle polling
Compute	one shared job cluster (6 workers)	one start/stop per run, not per task
bronze → silver	depends_on chain	ordered execution, 2 retries each
mart_by_region	for_each over regions, concurrency 3	parallel marts, bounded
has_new_data	condition on silver_rows	boolean gate
global_rollup	depends on mart + condition=true	skipped when no new data

After deployment, the job sleeps until vendor files arrive, then wakes one shared cluster that runs bronze → silver, fans the mart build across three regions concurrently, and runs the global rollup only when silver produced rows. Empty drops cost nothing beyond the ingest; the DBU bill tracks actual data volume, not wall-clock time.

Output:

Metric	Value
Clusters started per run	1 (shared)
Idle polling compute	0 (event-driven)
Mart parallelism	3 concurrent regions
Rollup on empty drop	skipped
Retries	2 per compute task
On-call signal	email on failure

Why this works — concept by concept:

File-arrival trigger — the job is reactive, not scheduled. Compute is spent only when data actually lands, eliminating the all-day polling tax a sensor-based design would pay.
Shared job cluster — one job_cluster_key across every task means a single ephemeral cluster serves the whole DAG. This is the flat-DBU lever: O(1) cluster starts per run instead of O(tasks).
for_each fan-out with bounded concurrency — the mart build scales with the region list without hand-writing a task per region, and concurrency: 3 protects the cluster from a thundering-herd of parallel dbt runs.
condition_task gate — global_rollup runs only when silver_rows > 0, so empty drops skip the most expensive step. Control flow lives in the job, not in application code.
Cost — DBUs billed only while the single shared cluster runs, only when files arrive, with the rollup skipped on empty data. Compared to a scheduled poll-and-run design on per-task clusters, this is dramatically cheaper and scales with data volume, not with the clock. O(1) clusters per run; O(regions) parallel marts bounded by concurrency.

ETL
Topic — etl
ETL problems on medallion and incremental jobs

Practice →

Data Transformation Topic — data-transformation Transformation problems on bronze-silver-gold pipelines

Practice →

3. Apache Airflow — DAGs, operators, the DatabricksSubmitRunOperator, scheduler & executors

`apache airflow` models the same problem as a Python DAG that reaches every system — operators, a scheduler, pluggable executors, and a first-class Databricks provider

The mental model in one line: Apache Airflow is the out-of-lakehouse orchestrator where a dag is a Python object whose nodes are operators (or @task-decorated functions) wired by >> into task dependencies, a scheduler parses those DAGs and enqueues ready tasks, and a pluggable executor (Local, Celery, Kubernetes) runs them — so the same workflow orchestration problem Databricks solves in-lakehouse is solved by a portable control plane that speaks S3, Snowflake, dbt, HTTP, and, through the airflow databricks operator provider, Databricks itself. Airflow's superpower is reach; its cost is that you (or a managed vendor) operate a standing system.

The DAG model — Python defines the graph.

DAG object. A DAG (or @dag) sets the schedule, start date, catchup policy, default args, and tags. Everything about when and how the graph runs lives here.
Tasks. Each node is an operator instance or a @task function (TaskFlow API). The task's task_id is its identity in the UI and the metadata DB.
Dependencies. a >> b means "b runs after a"; chain(a, b, c) and a >> [b, c] >> d build fan-out/fan-in. These edges are the DAG.
Idempotency + params. Tasks are keyed by logical date (the data interval), so a run for 2026-08-03 always processes that interval — re-running is deterministic. params and Variable/Connection carry config.

The operator + sensor catalogue — Airflow's reach.

Core operators. PythonOperator / @task, BashOperator, EmptyOperator (structure), plus branching (@task.branch) and dynamic task mapping (.expand(...)).
Provider operators. S3*, SnowflakeOperator, BigQueryInsertJobOperator, SparkSubmitOperator, dbt Cloud, HTTP — hundreds of them. This catalogue is why Airflow wins heterogeneous fleets.
Sensors. S3KeySensor, SqlSensor, ExternalTaskSensor wait for a condition. Deferrable sensors release the worker slot while waiting (see below).
XComs. Small cross-task values (ti.xcom_push / xcom_pull, or TaskFlow return values). Large data moves through storage, never through XCom.

The Databricks provider — the airflow databricks operator family.

DatabricksSubmitRunOperator. Submits a one-off run (a notebook, spark_python, etc.) to Databricks, optionally spinning a new cluster. Good for ad-hoc runs defined entirely in the DAG.
DatabricksRunNowOperator. Triggers an existing Databricks job by job_id — the preferred production hook, because the job (its tasks, cluster, retries, lineage) is owned in Databricks, and Airflow just fires it.
DatabricksNotebookOperator / DatabricksTaskOperator. Run a specific notebook or task; DatabricksWorkflowTaskGroup (section 4) groups several so they share one cluster.
DatabricksSqlOperator. Run SQL against a Databricks SQL warehouse. All of these authenticate through an Airflow connection (databricks_default).

The scheduler + executor — how tasks actually run.

Scheduler. Continuously parses DAG files, computes which task instances are ready (dependencies met, schedule due), and enqueues them. It is the always-on heart of Airflow.
Executors. SequentialExecutor (dev only), LocalExecutor (one host, parallel), CeleryExecutor (a worker fleet behind a broker), KubernetesExecutor (one pod per task — elastic, isolated). The executor is the biggest ops decision.
Deferrable operators + triggerer. A deferrable operator (e.g. DatabricksSubmitRunDeferrableOperator) suspends while waiting on Databricks, freeing the worker slot; a lightweight triggerer process resumes it on the event. This is how you wait on 10,000 long jobs without 10,000 occupied workers.

Common interview probes on Airflow.

"Why DatabricksRunNowOperator over DatabricksSubmitRunOperator in production?" — the job is owned in Databricks; Airflow just triggers it (cluster reuse, native lineage, one source of truth).
"How do you wait on a long Databricks job without hogging a worker?" — deferrable operator + triggerer.
"Which executor for elastic, isolated tasks?" — KubernetesExecutor.
"How do tasks pass data?" — XCom for small values; storage/tables for large data.

Worked example — an Airflow TaskFlow DAG driving a Databricks job

Detailed explanation. The canonical cross-system DAG: sensor waits on an S3 drop, a Python task validates it, DatabricksRunNowOperator fires an existing Databricks job, and a final task publishes a metric. Written with the TaskFlow API plus provider operators.

Wait. S3KeySensor (deferrable) for the input file.
Validate. A @task checks row count / schema.
Execute lakehouse work. DatabricksRunNowOperator triggers job by job_id.
Finish. A @task records success.

Question. Write the Airflow DAG that waits on S3, validates, triggers a Databricks job by id, and records completion — daily, no catchup, with retries.

Input.

Element	Value
Schedule	`@daily`, catchup=False
Sensor	S3KeySensor (deferrable)
Databricks	RunNow on `job_id=712` via `databricks_default`
Retries	2, 5-minute delay

Code.

# dags/daily_lakehouse.py
from __future__ import annotations
import pendulum
from airflow.decorators import dag, task
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.providers.databricks.operators.databricks import DatabricksRunNowOperator

DEFAULT_ARGS = {"retries": 2, "retry_delay": pendulum.duration(minutes=5)}

@dag(
    schedule="@daily",
    start_date=pendulum.datetime(2026, 8, 1, tz="UTC"),
    catchup=False,                 # do not backfill history on first deploy
    default_args=DEFAULT_ARGS,
    tags=["lakehouse", "databricks"],
)
def daily_lakehouse():

    wait_for_drop = S3KeySensor(
        task_id="wait_for_drop",
        bucket_name="vendor-drop",
        bucket_key="orders/{{ ds }}/_SUCCESS",   # templated by run date
        deferrable=True,                          # release the worker slot
        poke_interval=60,
        timeout=6 * 3600,
    )

    @task
    def validate(ds: str | None = None) -> dict:
        """Cheap pre-flight before we pay for a Databricks cluster."""
        # (in real life: read a manifest, check row counts / schema)
        return {"run_date": ds, "ok": True}

    run_databricks = DatabricksRunNowOperator(
        task_id="run_databricks_job",
        databricks_conn_id="databricks_default",
        job_id=712,                               # an EXISTING Databricks job
        notebook_params={"run_date": "{{ ds }}"},
    )

    @task
    def record_success(meta: dict) -> None:
        print(f"Lakehouse refresh complete for {meta['run_date']}")

    meta = validate()
    wait_for_drop >> meta          # sensor gates validation
    meta >> run_databricks         # validate gates the Databricks run
    run_databricks >> record_success(meta)

daily_lakehouse()

Step-by-step explanation.

The @dag decorator declares when the graph runs: @daily, starting 2026-08-01, catchup=False so deploying today does not backfill every past day. default_args gives every task two retries with a five-minute delay — reliability declared once.
S3KeySensor with deferrable=True waits for orders/{{ ds }}/_SUCCESS. {{ ds }} is the run's logical date, so each daily run watches its own prefix. Because it is deferrable, the sensor suspends and frees its worker slot while waiting up to six hours — a thousand such DAGs do not occupy a thousand workers.
validate is a TaskFlow @task; its return dict becomes an XCom automatically. It runs a cheap pre-flight (manifest / row-count / schema) before the pipeline pays for a Databricks cluster — fail fast on bad input.
DatabricksRunNowOperator triggers existing job 712 through the databricks_default connection, passing run_date as a notebook parameter. Airflow does not define the lakehouse tasks or cluster — those are owned in Databricks. Airflow's job is to fire it and track its status.
The dependency wiring is explicit: wait_for_drop >> meta >> run_databricks >> record_success. The DAG reaches four systems (S3, Airflow-Python, Databricks, a metric sink) in one graph — exactly the heterogeneous reach an in-lakehouse orchestrator lacks.

Output.

Task	Type	Waits on	Runs on
wait_for_drop	S3KeySensor (deferrable)	schedule	triggerer (suspended)
validate	@task	sensor	executor worker
run_databricks_job	DatabricksRunNowOperator	validate	Databricks (job 712)
record_success	@task	Databricks run	executor worker

Rule of thumb. In production, trigger an existing Databricks job with DatabricksRunNowOperator rather than defining the run inline — the job stays owned in Databricks (cluster reuse, retries, lineage), and Airflow stays the cross-system conductor. Use deferrable sensors so long waits do not occupy workers.

Worked example — choosing and configuring an executor

Detailed explanation. The executor decides how tasks physically run and is the biggest Airflow ops decision. A team with bursty, isolated, heterogeneous tasks wants the KubernetesExecutor (one pod per task); a steady mid-volume team is well served by CeleryExecutor (a fixed worker fleet). Walk through the trade-off and a minimal KubernetesExecutor task-level override.

LocalExecutor. One host, threads/processes. Simple; caps at one machine.
CeleryExecutor. A broker (Redis/RabbitMQ) plus a worker fleet. Horizontal scale; you operate the fleet.
KubernetesExecutor. One pod per task; elastic and isolated; you operate a cluster.

Question. Configure a heavy task to run in its own Kubernetes pod with a larger resource request, while the rest of the DAG runs on the default pod spec.

Input.

Component	Value
Cluster executor	KubernetesExecutor
Default task	1 CPU / 1 Gi pod
Heavy task	4 CPU / 8 Gi pod (override)
Isolation	one pod per task

Code.

# dags/executor_demo.py
from __future__ import annotations
import pendulum
from airflow.decorators import dag, task
from kubernetes.client import V1Pod, V1ObjectMeta, V1PodSpec, V1Container, V1ResourceRequirements

HEAVY_POD = V1Pod(
    metadata=V1ObjectMeta(labels={"tier": "heavy"}),
    spec=V1PodSpec(containers=[
        V1Container(
            name="base",
            resources=V1ResourceRequirements(
                requests={"cpu": "4", "memory": "8Gi"},
                limits={"cpu": "4", "memory": "8Gi"},
            ),
        )
    ]),
)

@dag(schedule="@hourly",
     start_date=pendulum.datetime(2026, 8, 1, tz="UTC"),
     catchup=False, tags=["k8s"])
def executor_demo():

    @task
    def light_step() -> int:
        return 42                              # runs in the default pod

    # Per-task pod override — KubernetesExecutor merges this on top of the base pod
    @task(executor_config={"pod_override": HEAVY_POD})
    def heavy_step(x: int) -> int:
        # a CPU/memory-hungry transform gets its own 4-CPU / 8-Gi pod
        return x * x

    heavy_step(light_step())

executor_demo()

Step-by-step explanation.

Under the KubernetesExecutor, every task instance becomes a Kubernetes pod: the scheduler asks the API server to launch a pod, the task runs, the pod dies. Isolation and elasticity are automatic — no standing worker fleet to size.
light_step uses the cluster's default pod template (say 1 CPU / 1 Gi). Most tasks want this — cheap, uniform.
heavy_step passes executor_config={"pod_override": HEAVY_POD}. The KubernetesExecutor merges that partial pod spec over the base template, so only this task gets a 4-CPU / 8-Gi pod. You right-size per task instead of sizing a whole worker fleet for the largest task.
Contrast with CeleryExecutor: there, all tasks land on pre-provisioned workers, so the heavy task either fits the standard worker or needs a dedicated queue. Kubernetes per-task pods make per-task sizing trivial.
The choice generalizes: LocalExecutor for a single-box dev/small prod, CeleryExecutor for steady predictable throughput on a fleet you manage, KubernetesExecutor for bursty, isolated, heterogeneously-sized tasks. Managed Airflow (MWAA/Astronomer/Composer) picks and operates one of these for you.

Output.

Executor	Task placement	Scale unit	You operate
Local	threads on one host	vertical	the host
Celery	pre-provisioned workers	worker count	broker + workers
Kubernetes	one pod per task	pods (elastic)	the k8s cluster
Managed (MWAA/etc.)	vendor-run	vendor-scaled	almost nothing

Rule of thumb. Pick LocalExecutor to start, CeleryExecutor for steady fleets, and KubernetesExecutor for bursty, isolated, variably-sized tasks — and use executor_config pod overrides to right-size heavy tasks instead of oversizing everything. If you do not want to operate any of it, buy managed Airflow.

Worked example — dynamic task mapping over a partition list

Detailed explanation. Airflow's .expand() (dynamic task mapping) generates one task instance per element of a list computed at runtime — the Airflow analogue of Databricks for_each_task. A DAG discovers the day's partitions, then processes each in its own mapped task with bounded parallelism.

Discover. A @task returns a list of partition keys.
Map. A downstream @task is .expand(partition=...) — one instance per key.
Bound. max_active_tis_per_dag (or pool) caps concurrency.

Question. Write a DAG that lists partitions at runtime and processes each in a mapped task, capped at four concurrent instances.

Input.

Element	Value
Discovery	`@task` returns partition list
Mapping	`.expand(partition=...)`
Concurrency cap	4 (via pool)
Reduce	a final task aggregates results

Code.

# dags/dynamic_partitions.py
from __future__ import annotations
import pendulum
from airflow.decorators import dag, task

@dag(schedule="@daily",
     start_date=pendulum.datetime(2026, 8, 1, tz="UTC"),
     catchup=False, tags=["mapping"])
def dynamic_partitions():

    @task
    def list_partitions(ds: str | None = None) -> list[str]:
        # discovered at runtime — could be an S3 listing or a metastore query
        return [f"{ds}/region=us", f"{ds}/region=eu", f"{ds}/region=apac"]

    @task(pool="partition_pool")     # pool with 4 slots caps concurrency
    def process(partition: str) -> int:
        # process one partition; return a row count
        return len(partition)        # stand-in for real work

    @task
    def summarize(counts: list[int]) -> None:
        print(f"processed {len(counts)} partitions, total={sum(counts)}")

    parts = list_partitions()
    counts = process.expand(partition=parts)   # one mapped task per partition
    summarize(counts)

dynamic_partitions()

Step-by-step explanation.

list_partitions returns a Python list at runtime — the partitions are not known when the DAG file is parsed. This is the key difference from a static loop: the graph's width is data-driven.
process.expand(partition=parts) tells Airflow to create one process task instance per element of parts. The UI shows a mapped task with N indices; each processes exactly one partition, idempotently keyed by its input.
pool="partition_pool" (a pool defined with four slots) caps how many mapped instances run at once, protecting downstream systems from a thundering herd — the same role concurrency: 3 plays in Databricks for_each_task.
summarize(counts) receives the list of all mapped return values (an automatic gather/reduce) and aggregates them. Mapping plus reduce is the fan-out/fan-in idiom expressed in pure Python.
This is Airflow's reach and expressiveness in one screen: runtime-dynamic width, bounded concurrency, and a reduce — all portable across whatever systems the mapped task touches, not just Databricks.

Output.

Stage	Instances	Concurrency
list_partitions	1	—
process (mapped)	N (= partitions)	≤ 4 (pool)
summarize	1	gathers all

Rule of thumb. Use .expand() for runtime-dynamic fan-out and a pool to bound concurrency, then a plain @task to reduce. It is Airflow's for_each — with the added benefit that each mapped task can touch any system, not only the lakehouse.

Python interview question on Airflow orchestration

A senior interviewer might ask: "Build an Airflow DAG that waits on an external SFTP drop without hogging a worker, validates the file cheaply, triggers an existing Databricks job by id, dynamically fans a downstream export over the day's partitions with bounded concurrency, and alerts on failure. Explain your executor choice and how you keep Airflow from becoming the bottleneck."

Solution Using deferrable sensors, RunNow, dynamic mapping, and a bounded pool

# dags/sftp_to_lakehouse.py
from __future__ import annotations
import pendulum
from airflow.decorators import dag, task
from airflow.providers.sftp.sensors.sftp import SFTPSensor
from airflow.providers.databricks.operators.databricks import DatabricksRunNowOperator

DEFAULT_ARGS = {
    "retries": 2,
    "retry_delay": pendulum.duration(minutes=5),
}

@dag(
    schedule="0 3 * * *",                      # 03:00 daily (cron)
    start_date=pendulum.datetime(2026, 8, 1, tz="UTC"),
    catchup=False,
    default_args=DEFAULT_ARGS,
    tags=["sftp", "databricks", "hybrid-edge"],
)
def sftp_to_lakehouse():

    # 1. Wait on the SFTP drop WITHOUT holding a worker slot
    wait = SFTPSensor(
        task_id="wait_for_sftp",
        sftp_conn_id="vendor_sftp",
        path="/incoming/orders_{{ ds_nodash }}.csv",
        deferrable=True,                       # suspend; triggerer resumes
        poke_interval=120,
        timeout=4 * 3600,
    )

    @task
    def validate(ds: str | None = None) -> str:
        # cheap pre-flight before paying for lakehouse compute
        return ds or "unknown"

    # 2. Trigger the EXISTING Databricks job (owned in Databricks)
    refresh = DatabricksRunNowOperator(
        task_id="databricks_refresh",
        databricks_conn_id="databricks_default",
        job_id=712,
        notebook_params={"run_date": "{{ ds }}"},
    )

    # 3. Dynamically fan the export over the day's partitions
    @task
    def partitions_of(run_date: str) -> list[str]:
        return [f"{run_date}/region=us", f"{run_date}/region=eu", f"{run_date}/region=apac"]

    @task(pool="export_pool")                  # 4-slot pool bounds concurrency
    def export_partition(partition: str) -> int:
        return len(partition)                  # stand-in for a real export

    @task
    def finalize(results: list[int]) -> None:
        print(f"exported {len(results)} partitions")

    run_date = validate()
    wait >> run_date >> refresh
    parts = partitions_of(run_date)
    refresh >> parts
    finalize(export_partition.expand(partition=parts))

sftp_to_lakehouse()

# airflow.cfg (excerpt) — executor + triggerer for the deferrable sensor
[core]
executor = KubernetesExecutor        # bursty, isolated, per-task pods

[triggerer]
default_capacity = 1000              # thousands of suspended sensors, few workers

Step-by-step trace.

Layer	Choice	Reasoning
Wait	deferrable SFTPSensor	suspends; frees the worker slot for 4 h
Triggerer	capacity 1000	resumes suspended sensors cheaply
Validate	cheap `@task`	fail fast before lakehouse spend
Lakehouse work	DatabricksRunNowOperator (job 712)	job owned in Databricks; cluster reuse + lineage
Fan-out	`.expand()` over partitions	runtime-dynamic width
Concurrency	pool `export_pool` (4)	bound downstream load
Executor	KubernetesExecutor	per-task pods; elastic, isolated

After deployment, the DAG fires at 03:00, the deferrable sensor waits up to four hours for the SFTP file without occupying a worker, a cheap validate gate runs, an existing Databricks job does the lakehouse-heavy refresh (reusing one cluster, keeping native lineage), and a dynamically mapped export fans over the day's partitions four-at-a-time. Airflow conducts; it never becomes the compute bottleneck because the heavy work runs on Databricks pods, not on the scheduler.

Output:

Metric	Value
Worker slots held while waiting	0 (deferrable + triggerer)
Databricks clusters per run	1 (job 712 shared)
Export parallelism	≤ 4 (pool)
Executor	KubernetesExecutor (per-task pods)
Retries	2 per task
Systems reached	SFTP, Databricks, export sink

Why this works — concept by concept:

Deferrable sensor + triggerer — the SFTP wait suspends and releases its worker slot; a single lightweight triggerer resumes thousands of such waits. This is how Airflow waits on slow external events without a worker-per-wait blowup.
DatabricksRunNowOperator — Airflow triggers an existing Databricks job rather than defining the run inline, so cluster reuse, retries, and Unity Catalog lineage stay owned in the lakehouse. Airflow is the conductor, not the executor.
Dynamic task mapping + pool — .expand() gives runtime-dynamic fan-out and a 4-slot pool bounds concurrency, protecting the export sink — Airflow's portable equivalent of Databricks for_each_task.
KubernetesExecutor — bursty, isolated, per-task pods size compute to the task and scale to zero between runs, so the standing footprint is just the scheduler + triggerer + a metadata DB.
Cost — a standing (ideally managed) Airflow control plane, near-zero worker occupancy during waits, and DBUs billed by Databricks only while job 712's shared cluster runs. Compared to a worker-per-wait, cluster-per-task naive design, this keeps both the Airflow footprint and the DBU bill flat. O(1) Databricks clusters per run; O(partitions) mapped exports bounded by the pool.

ETL
Topic — etl
ETL problems on cross-system DAGs

Practice →

Design Topic — design Design problems on scheduler and executor topologies

Practice →

4. The hybrid reality — Airflow orchestrating Databricks jobs, when each layer owns what

Airflow conducts the fleet, Databricks Workflows executes the lakehouse — and `DatabricksWorkflowTaskGroup` reuses one cluster across Airflow-launched tasks

The mental model in one line: the hybrid pattern draws an ownership boundary — Airflow owns the cross-system schedule, the external-event waits, and the fleet-wide task dependencies; Databricks Workflows owns the in-lakehouse task DAG, the ephemeral cluster, and the native lineage — and the two are stitched together either by triggering an existing Databricks job (DatabricksRunNowOperator) or by declaring a DatabricksWorkflowTaskGroup inside the Airflow DAG so several Databricks tasks launched from Airflow share one job cluster instead of paying a per-task cluster start. This is the answer most large teams actually ship, because it gives Airflow's reach and the lakehouse's cost/lineage benefits.

The ownership boundary — who owns what.

Airflow owns. The wall-clock schedule (or the cross-system event that starts everything), sensors on external systems, dependencies between systems (Snowflake → Databricks → dbt Cloud), fleet-wide alerting and SLAs, and backfills across the whole platform.
Databricks Workflows owns. The dependency graph within the lakehouse, the ephemeral job cluster (and its reuse), per-task retries on Spark/SQL/dbt tasks, and Unity Catalog table lineage.
The seam. Airflow hands off to Databricks either by run-now on an existing job_id, or by embedding a DatabricksWorkflowTaskGroup that materializes a Databricks Workflow from the Airflow DAG.

Two stitching styles — RunNow vs WorkflowTaskGroup.

RunNow (loose coupling). The Databricks job is defined and owned in Databricks (DAB/UI). Airflow triggers it by id and polls status. Cleanest ownership split; the job is one source of truth. Best when the lakehouse job is stable and reused by many DAGs.
WorkflowTaskGroup (tight coupling). You declare Databricks tasks inside the Airflow DAG using DatabricksWorkflowTaskGroup + DatabricksNotebookOperator/DatabricksTaskOperator. On run, the provider creates an ephemeral Databricks Workflow whose tasks all share one job cluster. Best when the lakehouse steps are DAG-specific and you want them visible as Airflow tasks and cheap (one cluster).

Why the WorkflowTaskGroup exists — the one-cluster-per-task tax.

The naive anti-pattern. Five DatabricksSubmitRunOperator tasks in a row, each with its own new_cluster, start five clusters — five start-up taxes (~2–5 min each) and five times the idle-edge DBU waste.
The fix. DatabricksWorkflowTaskGroup launches those five as one Databricks Workflow on one shared job cluster. The cluster starts once, runs all five in dependency order, and tears down once.
The visibility bonus. The grouped tasks appear both in the Airflow UI (as a task group) and in the Databricks Jobs UI (as a workflow run) — one lineage, two lenses.

When to prefer each layer's native features.

Retries. Let Databricks retry within a task (transient Spark failure); let Airflow retry the handoff (Databricks API blip). Do not stack unbounded retries on both.
Alerting. Databricks alerts on task-level failures with lakehouse context; Airflow alerts on DAG-level SLA misses and cross-system failures. Route each to where the fix lives.
Lineage. Unity Catalog gives table-level lineage for lakehouse work automatically; Airflow gives task-level DAG lineage across systems. Neither replaces the other.

Common interview probes on hybrid orchestration.

"How do you avoid one cluster per Databricks task launched from Airflow?" — DatabricksWorkflowTaskGroup (shared cluster) or trigger one multi-task job via RunNow.
"Where should retries live?" — task-transient in Databricks; handoff/cross-system in Airflow.
"RunNow vs WorkflowTaskGroup — when?" — RunNow for stable reused jobs; WorkflowTaskGroup for DAG-specific lakehouse steps you want visible in Airflow and cheap.
"Who owns lineage?" — both, at different granularities.

Worked example — DatabricksWorkflowTaskGroup with a shared cluster

Detailed explanation. An Airflow DAG must run three Databricks notebooks (bronze → silver → gold) as part of a larger cross-system pipeline, and they must share one cluster. DatabricksWorkflowTaskGroup materializes them as a single Databricks Workflow on one job cluster.

Group. DatabricksWorkflowTaskGroup defines the shared job_clusters.
Tasks. Three DatabricksNotebookOperator tasks inside the group, wired by >>.
Result. One Databricks cluster start for all three.

Question. Write the Airflow DAG fragment that runs three Databricks notebooks in a WorkflowTaskGroup on one shared cluster, gated by an upstream Snowflake load.

Input.

Element	Value
Upstream	a Snowflake load task (Airflow)
Group	DatabricksWorkflowTaskGroup, one job cluster
Tasks	bronze, silver, gold notebooks
Cluster starts	1 (shared)

Code.

# dags/hybrid_workflow.py
from __future__ import annotations
import pendulum
from airflow.decorators import dag, task
from airflow.providers.databricks.operators.databricks_workflow import (
    DatabricksWorkflowTaskGroup,
)
from airflow.providers.databricks.operators.databricks import DatabricksNotebookOperator

JOB_CLUSTERS = [
    {
        "job_cluster_key": "hybrid_shared",
        "new_cluster": {
            "spark_version": "15.4.x-scala2.12",
            "node_type_id": "i3.xlarge",
            "num_workers": 4,
            "data_security_mode": "SINGLE_USER",
        },
    }
]

@dag(schedule="@daily",
     start_date=pendulum.datetime(2026, 8, 1, tz="UTC"),
     catchup=False, tags=["hybrid"])
def hybrid_workflow():

    @task
    def snowflake_load() -> str:
        # cross-system step Airflow owns (stand-in)
        return "loaded"

    up = snowflake_load()

    # All Databricks tasks below share ONE job cluster
    with DatabricksWorkflowTaskGroup(
        group_id="lakehouse",
        databricks_conn_id="databricks_default",
        job_clusters=JOB_CLUSTERS,
    ) as lakehouse:
        bronze = DatabricksNotebookOperator(
            task_id="bronze",
            notebook_path="/Repos/prod/etl/bronze_ingest",
            job_cluster_key="hybrid_shared",
            databricks_conn_id="databricks_default",
        )
        silver = DatabricksNotebookOperator(
            task_id="silver",
            notebook_path="/Repos/prod/etl/silver_transform",
            job_cluster_key="hybrid_shared",
            databricks_conn_id="databricks_default",
        )
        gold = DatabricksNotebookOperator(
            task_id="gold",
            notebook_path="/Repos/prod/etl/gold_marts",
            job_cluster_key="hybrid_shared",
            databricks_conn_id="databricks_default",
        )
        bronze >> silver >> gold

    up >> lakehouse

hybrid_workflow()

Step-by-step explanation.

snowflake_load is a cross-system step Airflow owns — the kind of task an in-lakehouse orchestrator cannot reach natively. It gates the lakehouse group.
DatabricksWorkflowTaskGroup declares job_clusters=[{... "job_cluster_key": "hybrid_shared" ...}] once. Every task inside the group references that key, so the provider builds a single Databricks Workflow whose tasks all bind to one cluster.
The three DatabricksNotebookOperator tasks are wired bronze >> silver >> gold. When the DAG runs, the provider creates an ephemeral Databricks Workflow: one cluster starts, runs the three notebooks in order, and tears down. That is one cluster start for the whole group, not three.
up >> lakehouse connects the Airflow-owned Snowflake step to the whole Databricks group. In the Airflow UI the group renders as a collapsible task group; in the Databricks Jobs UI the same run appears as a workflow with three tasks and native lineage.
Contrast with three standalone DatabricksSubmitRunOperator tasks each carrying new_cluster: that would start three clusters (three start-up taxes, three idle edges). The WorkflowTaskGroup is the cost fix and the visibility win.

Output.

Task	Layer	Cluster	Starts
snowflake_load	Airflow	—	—
bronze	Databricks (grouped)	hybrid_shared	shared
silver	Databricks (grouped)	hybrid_shared	shared
gold	Databricks (grouped)	hybrid_shared	shared
cluster total	—	1	1 start

Rule of thumb. When several Databricks tasks launched from Airflow belong to one DAG, wrap them in a DatabricksWorkflowTaskGroup sharing one job_cluster_key. One cluster start replaces N, and the run is visible in both UIs with native lineage.

Worked example — placing retries and alerts on the right layer

Detailed explanation. A hybrid pipeline fails in two very different ways: a transient Spark shuffle error inside a Databricks task, and a Databricks-API timeout on the Airflow handoff. Putting all retries in one layer either masks real failures or thrashes clusters. The fix is to place each retry where the failure lives.

Task-transient failure. Retry inside Databricks (per-task max_retries) — cheap, keeps the shared cluster warm.
Handoff failure. Retry in Airflow (operator retries) — re-fires run-now if the API blipped.
Alerts. Databricks alerts carry lakehouse context; Airflow alerts carry cross-system context.

Question. Configure retries so a transient Spark failure is retried by Databricks and an API-handoff failure is retried by Airflow, with alerts routed to the layer that can act.

Input.

Failure	Retry owner	Alert owner
Spark shuffle blip	Databricks task max_retries	Databricks task alert
Databricks API timeout	Airflow operator retries	Airflow DAG alert
Bad input data	neither (fail fast)	Airflow (upstream validate)

Code.

# Airflow side — retry the HANDOFF, not the Spark work
from airflow.providers.databricks.operators.databricks import DatabricksRunNowOperator
import pendulum

refresh = DatabricksRunNowOperator(
    task_id="databricks_refresh",
    databricks_conn_id="databricks_default",
    job_id=712,
    retries=3,                                  # retry the run-now API handoff
    retry_delay=pendulum.duration(minutes=2),
    retry_exponential_backoff=True,
    # If the underlying Databricks RUN fails after its own retries, do NOT
    # let Airflow blindly re-run heavy Spark work — surface it instead.
)

# Databricks side (DAB) — retry TRANSIENT task failures on the warm cluster
resources:
  jobs:
    lakehouse_refresh:            # this is job_id 712
      name: lakehouse_refresh
      tasks:
        - task_key: heavy_transform
          spark_python_task:
            python_file: /Repos/prod/etl/heavy_transform.py
          job_cluster_key: shared
          max_retries: 3                        # retry transient Spark failures
          min_retry_interval_millis: 30000
          retry_on_timeout: true
      job_clusters:
        - job_cluster_key: shared
          new_cluster:
            spark_version: "15.4.x-scala2.12"
            node_type_id: "i3.xlarge"
            num_workers: 8
      # Databricks-native alert: task failure with lakehouse context
      webhook_notifications:
        on_failure:
          - id: ${var.databricks_alert_webhook}

Step-by-step explanation.

The Databricks job (job_id 712) owns retries for transient task failures: max_retries: 3 on heavy_transform, retried on the already-warm shared cluster. A shuffle blip does not bubble up to Airflow at all — it is resolved where it happened, cheaply.
The Airflow DatabricksRunNowOperator owns retries for the handoff: retries=3 with exponential backoff. If the run-now API call times out or the connection blips, Airflow re-fires it — but it does not re-run Spark work that already ran, because the retry is on the API call, not the compute.
If the Databricks run genuinely fails after its own three retries, that is a real failure. Airflow should surface it (alert), not blindly re-trigger job 712 and re-pay for heavy Spark work. The layers must not stack unbounded retries.
Alerts split by context: the Databricks webhook_notifications.on_failure carries lakehouse detail (which task, which cluster, the Spark error) to the team that fixes lakehouse code; Airflow's DAG-level alert carries cross-system context (which upstream, which SLA) to the platform on-call.
Bad input data is caught by an upstream Airflow validate task and fails fast — neither layer retries it, because retrying will never fix malformed input. Fail-fast beats retry-forever for deterministic errors.

Output.

Failure mode	Retried by	Alert to	Re-runs Spark?
Spark shuffle blip	Databricks (3×)	Databricks webhook	yes (warm cluster)
Databricks API timeout	Airflow (3×, backoff)	Airflow DAG	no (handoff only)
Run fails after Databricks retries	neither	Airflow (surface)	no
Malformed input	neither (fail fast)	Airflow validate	no

Rule of thumb. Put transient-task retries in Databricks (warm cluster, lakehouse context) and handoff retries in Airflow (API blips), and never stack unbounded retries on both. Route each alert to the layer whose team can actually fix the failure.

Worked example — one shared cluster across an Airflow-launched job

Detailed explanation. A cost review finds a hybrid DAG is starting eight Databricks clusters per run because eight DatabricksSubmitRunOperator tasks each declare new_cluster. The fix is to consolidate the eight into one Databricks job (owned in Databricks) triggered once by RunNow, with all eight tasks sharing one cluster — cutting cluster starts from eight to one.

Before. 8 SubmitRun operators × 8 new_cluster blocks = 8 cluster starts.
After. 1 Databricks job with 8 tasks on one shared cluster, triggered by 1 RunNow.
Saving. 8 start-up taxes → 1; 8 idle edges → 1.

Question. Show the before/after and quantify the cluster-start saving for a DAG that runs eight Databricks tasks per run, 30 runs/day.

Input.

Metric	Before	After
Databricks tasks	8	8
Cluster starts / run	8	1
Runs / day	30	30
Cluster starts / day	240	30

Code.

# BEFORE — anti-pattern: one cluster per task (8 starts per run)
from airflow.providers.databricks.operators.databricks import DatabricksSubmitRunOperator

NEW_CLUSTER = {"spark_version": "15.4.x-scala2.12",
               "node_type_id": "i3.xlarge", "num_workers": 4}

tasks = []
for step in ["s1", "s2", "s3", "s4", "s5", "s6", "s7", "s8"]:
    tasks.append(DatabricksSubmitRunOperator(
        task_id=f"db_{step}",
        databricks_conn_id="databricks_default",
        new_cluster=NEW_CLUSTER,               # <-- 8 separate clusters!
        notebook_task={"notebook_path": f"/Repos/prod/etl/{step}"},
    ))
# chain them ... each start-up tax paid 8 times

# AFTER — one Databricks job (owned in Databricks) triggered once
from airflow.providers.databricks.operators.databricks import DatabricksRunNowOperator

refresh = DatabricksRunNowOperator(
    task_id="db_refresh_all",
    databricks_conn_id="databricks_default",
    job_id=990,                                # a job whose 8 tasks share ONE cluster
)
# The Databricks job 990 (DAB) declares one job_cluster shared by s1..s8.

# job_id 990 (DAB) — 8 tasks, ONE shared cluster
resources:
  jobs:
    refresh_all:
      name: refresh_all
      tasks:
        - { task_key: s1, notebook_task: { notebook_path: /Repos/prod/etl/s1 }, job_cluster_key: one }
        - { task_key: s2, depends_on: [{ task_key: s1 }], notebook_task: { notebook_path: /Repos/prod/etl/s2 }, job_cluster_key: one }
        - { task_key: s3, depends_on: [{ task_key: s2 }], notebook_task: { notebook_path: /Repos/prod/etl/s3 }, job_cluster_key: one }
        # ... s4..s8 likewise, all job_cluster_key: one
      job_clusters:
        - job_cluster_key: one
          new_cluster:
            spark_version: "15.4.x-scala2.12"
            node_type_id: "i3.xlarge"
            num_workers: 4

Step-by-step explanation.

The before-code loops eight DatabricksSubmitRunOperator tasks, each with its own new_cluster. Every task therefore provisions a fresh cluster: eight cold starts (~2–5 min each) and eight idle-edge windows per run — pure waste.
The after-code replaces all eight with one DatabricksRunNowOperator triggering job 990. The eight steps now live inside Databricks as one job whose tasks all set job_cluster_key: one, so a single cluster serves the whole chain.
At 30 runs/day, cluster starts drop from 8 × 30 = 240/day to 1 × 30 = 30/day — an 8× reduction in start-up tax and idle-edge DBUs, with identical work performed.
Ownership improves too: the eight-step logic is now one Databricks job (one source of truth, native lineage, per-task retries), and Airflow's DAG is simpler — one handoff task instead of eight. Airflow conducts; Databricks executes.
If the eight steps are genuinely DAG-specific (not reused elsewhere), the equivalent fix is a DatabricksWorkflowTaskGroup (previous example) — same one-cluster outcome, but the steps stay visible as Airflow tasks. Either way, the rule is one cluster for the lakehouse leg.

Output.

Metric	Before	After
Cluster starts / run	8	1
Cluster starts / day	240	30
Start-up tax paid	8×	1×
Databricks source of truth	scattered in DAG	one job (990)
Airflow tasks for lakehouse leg	8	1

Rule of thumb. If an Airflow DAG launches multiple Databricks tasks, never give each its own new_cluster. Consolidate them into one Databricks job triggered by RunNow, or a DatabricksWorkflowTaskGroup — one cluster start replaces N, and the DBU bill drops proportionally.

Design interview question on hybrid orchestration

A senior interviewer might ask: "You run Airflow for the enterprise and Databricks for the lakehouse. A new pipeline waits on an external event, loads Snowflake, runs six Databricks transforms, then triggers a dbt Cloud job. Design the hybrid orchestration: draw the ownership boundary, decide RunNow vs WorkflowTaskGroup for the lakehouse leg, place retries and alerts, and keep the DBU bill flat."

Solution Using an ownership boundary with a WorkflowTaskGroup and layered retries

# dags/enterprise_hybrid.py
from __future__ import annotations
import pendulum
from airflow.decorators import dag, task
from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
from airflow.providers.databricks.operators.databricks_workflow import DatabricksWorkflowTaskGroup
from airflow.providers.databricks.operators.databricks import DatabricksNotebookOperator

SHARED = [{
    "job_cluster_key": "shared",
    "new_cluster": {"spark_version": "15.4.x-scala2.12",
                    "node_type_id": "i3.xlarge", "num_workers": 6,
                    "data_security_mode": "SINGLE_USER"},
}]

@dag(schedule="0 2 * * *",
     start_date=pendulum.datetime(2026, 8, 1, tz="UTC"),
     catchup=False,
     default_args={"retries": 2, "retry_delay": pendulum.duration(minutes=5)},
     tags=["enterprise", "hybrid"])
def enterprise_hybrid():

    # --- Airflow owns: external wait + Snowflake load ---
    wait = S3KeySensor(task_id="wait_event", bucket_name="events",
                       bucket_key="ready/{{ ds }}", deferrable=True,
                       poke_interval=120, timeout=4 * 3600)

    @task
    def snowflake_load() -> str:
        return "snowflake_ready"

    # --- Databricks owns: six transforms on ONE shared cluster ---
    with DatabricksWorkflowTaskGroup(
        group_id="lakehouse",
        databricks_conn_id="databricks_default",
        job_clusters=SHARED,
    ) as lakehouse:
        prev = None
        for step in ["t1", "t2", "t3", "t4", "t5", "t6"]:
            cur = DatabricksNotebookOperator(
                task_id=step,
                notebook_path=f"/Repos/prod/etl/{step}",
                job_cluster_key="shared",
                databricks_conn_id="databricks_default",
            )
            if prev:
                prev >> cur
            prev = cur

    # --- Airflow owns: dbt Cloud trigger + finalize ---
    @task(retries=3, retry_delay=pendulum.duration(minutes=2))
    def trigger_dbt_cloud() -> None:
        # call dbt Cloud run API (cross-system step Airflow owns)
        print("dbt Cloud job triggered")

    sf = snowflake_load()
    wait >> sf >> lakehouse >> trigger_dbt_cloud()

enterprise_hybrid()

Step-by-step trace.

Segment	Owner	Mechanism
wait_event	Airflow	deferrable S3 sensor (no worker held)
snowflake_load	Airflow	provider task
t1..t6 transforms	Databricks	WorkflowTaskGroup, one shared cluster
dbt Cloud trigger	Airflow	cross-system `@task` with handoff retries
retries (transient Spark)	Databricks	per-task max_retries
retries (handoff/API)	Airflow	operator/task retries
alerts	both	Databricks webhook (task) + Airflow DAG (SLA)

After deployment, the ownership boundary is crisp: Airflow waits on the external event, loads Snowflake, then hands the six lakehouse transforms to a single Databricks Workflow on one shared cluster, then triggers dbt Cloud. Six transforms cost one cluster start, not six. Transient Spark failures retry inside Databricks; handoff blips retry in Airflow; each alert reaches the team that can fix it.

Output:

Metric	Value
Cluster starts / run (lakehouse leg)	1 (shared across t1..t6)
Worker slots held during wait	0 (deferrable)
Systems reached	S3, Snowflake, Databricks, dbt Cloud
Retry placement	transient→Databricks, handoff→Airflow
Lineage	Airflow DAG + Unity Catalog table lineage
Alert routing	task→Databricks, SLA→Airflow

Why this works — concept by concept:

Ownership boundary — Airflow owns cross-system reach and scheduling; Databricks owns the in-lakehouse DAG, cluster, and lineage. Drawing the line explicitly is what makes a hybrid design maintainable rather than a tangle.
DatabricksWorkflowTaskGroup — the six transforms materialize as one Databricks Workflow on one shared job cluster, visible in both UIs. One cluster start replaces six — the core DBU lever of the whole design.
Layered retries — transient Spark failures retry on the warm cluster inside Databricks; API-handoff blips retry in Airflow. No unbounded double-retry, so a genuine failure surfaces instead of thrashing.
Deferrable external wait — the S3 sensor suspends and frees its worker, so a four-hour wait costs no worker slot. Airflow stays a lightweight conductor.
Cost — a standing (managed) Airflow control plane, zero worker occupancy during waits, and DBUs billed only while one shared cluster runs the six transforms. Compared to a cluster-per-task hybrid, this is roughly a 6× cut in cluster starts on the lakehouse leg while gaining full cross-system reach. O(1) clusters per run; O(systems) reach.

Streaming
Topic — streaming
Streaming problems on event-driven pipeline handoffs

Practice →

Design Topic — design Design problems on multi-layer orchestration ownership

Practice →

5. Decision matrix — cost, multi-system DAGs, observability, team skills + interview signals

Score four axes and let the constraints pick — Workflows, Airflow, or hybrid

The mental model in one line: the databricks workflows vs airflow decision resolves cleanly once you score four axes — total cost of ownership (standing infra + compute), multi-system DAG reach, observability and lineage, and team-skill fit — because each axis points at a different winner, and the combination of axes for your specific team selects Workflows (lakehouse-only), Airflow (heterogeneous fleet), or hybrid (both), rather than any universal champion. The senior skill is refusing to answer "which is better" and instead producing the scorecard that makes the answer fall out.

Axis 1 — cost / total cost of ownership.

Workflows. No standing orchestration infra; billed Jobs Compute DBUs only while tasks run. Cost tracks data work, not wall-clock. Shared job clusters keep it flat.
Airflow (self-managed). A standing cost: scheduler + webserver + triggerer + metadata DB + executor fleet, running whether or not any DAG is active. Plus the triggered compute.
Airflow (managed). MWAA / Astronomer / Composer fold the standing cost into a monthly bill — less ops, still a floor cost independent of throughput.

Axis 2 — multi-system DAG reach.

Workflows. Lakehouse-first; reaches outside only via generic tasks. Weak for heterogeneous fleets.
Airflow. Hundreds of provider operators/hooks — S3, Snowflake, BigQuery, dbt, Kafka, HTTP, SFTP, Databricks. The clear winner whenever the DAG spans many systems.
Hybrid. Airflow supplies the reach; Workflows supplies the cheap in-lakehouse execution.

Axis 3 — observability + lineage.

Workflows. Native Jobs UI, task run history, and automatic Unity Catalog table-level lineage for lakehouse work. Best in-lakehouse visibility.
Airflow. Rich DAG/Gantt/graph UI, logs, SLAs, and cross-system task lineage — but no automatic table-level lineage; you integrate OpenLineage/Marquez for that.
Hybrid. Airflow for cross-system DAG observability; Unity Catalog for lakehouse table lineage — two complementary lenses.

Axis 4 — team skills + operational fit.

Workflows. Lowest barrier for a Databricks-native team; DAB YAML + notebooks, nothing new to operate.
Airflow. Python-fluent platform engineers who already think in DAGs; but someone must own the deployment (or buy managed).
Hybrid. Requires both skill sets, but partitions them — platform team owns Airflow, data team owns Databricks jobs.

Common interview probes on the decision.

"One-line rule?" — inside the lakehouse → Workflows; across many systems → Airflow; both → hybrid.
"Where does Airflow cost hide?" — the always-on control plane, independent of throughput.
"Which gives table-level lineage for free?" — Databricks Workflows via Unity Catalog.
"When is hybrid not worth it?" — a pure-Databricks shop with no external systems; the second control plane is pure overhead.

Worked example — scoring the four axes for three teams

Detailed explanation. The decision becomes objective when you score each axis per team and read off the winner. Score Workflows and Airflow 1–3 on each axis for three teams and sum.

Team A. Pure Databricks analytics, no external systems.
Team B. Multi-cloud platform, Databricks is 1 of 7 systems.
Team C. 70% Databricks, waits on external drops, loads Snowflake.

Question. Produce the per-team scorecard and state each team's decision.

Input.

Axis (weight)	Team A leans	Team B leans	Team C leans
Cost	Workflows	Airflow (managed)	hybrid
Reach	tie (no external)	Airflow	Airflow
Observability	Workflows (lineage)	Airflow (cross-system)	both
Team skills	Workflows	Airflow	both

Code.

# score.py — sum the four axes and pick the orchestrator
def decide(scores: dict[str, dict[str, int]]) -> str:
    """scores: axis -> {'workflows': 1..3, 'airflow': 1..3}"""
    wf = sum(a["workflows"] for a in scores.values())
    af = sum(a["airflow"] for a in scores.values())
    if abs(wf - af) <= 1:
        return f"Hybrid (wf={wf}, af={af}) — axes disagree, use both"
    return ("Databricks Workflows" if wf > af else "Apache Airflow") + f" (wf={wf}, af={af})"

team_a = {"cost": {"workflows": 3, "airflow": 1},
          "reach": {"workflows": 2, "airflow": 2},
          "obs":   {"workflows": 3, "airflow": 2},
          "skills":{"workflows": 3, "airflow": 1}}

team_b = {"cost": {"workflows": 1, "airflow": 2},
          "reach": {"workflows": 1, "airflow": 3},
          "obs":   {"workflows": 1, "airflow": 3},
          "skills":{"workflows": 1, "airflow": 3}}

team_c = {"cost": {"workflows": 2, "airflow": 2},
          "reach": {"workflows": 1, "airflow": 3},
          "obs":   {"workflows": 2, "airflow": 2},
          "skills":{"workflows": 2, "airflow": 2}}

print("Team A:", decide(team_a))
print("Team B:", decide(team_b))
print("Team C:", decide(team_c))

Step-by-step explanation.

Team A scores Workflows 11 vs Airflow 6 — a clear Workflows win. With no external systems, reach is a tie and every other axis favors the in-lakehouse tool. Standing up Airflow would add cost and ops for zero reach benefit.
Team B scores Airflow 11 vs Workflows 4 — a clear Airflow win. Databricks is one of seven systems; only Airflow's provider catalogue can wire the whole fleet into one DAG. The Databricks work is triggered as one job among many.
Team C scores Workflows 7 vs Airflow 9 — within one point, so the helper returns hybrid. The axes genuinely disagree: reach favors Airflow, cost/observability are near-tied. Hybrid resolves it — Airflow for the external wait and Snowflake, Workflows for the lakehouse leg.
The abs(wf - af) <= 1 band encodes a real interview truth: when the axes are close, forcing a single tool is worse than embracing hybrid. The scorecard tells you when hybrid is the honest answer.
The method generalizes: weight axes if your org cares more about one (e.g. double cost in a cost-cutting year), rescore, and re-read. The output is defensible because it is derived, not asserted.

Output.

Team	Workflows	Airflow	Decision
A (pure Databricks)	11	6	Databricks Workflows
B (multi-cloud fleet)	4	11	Apache Airflow
C (70% Databricks + external)	7	9	Hybrid

Rule of thumb. Score cost, reach, observability, and skills 1–3 for each tool, sum, and pick — but when the totals are within one point, choose hybrid. The scorecard converts "which is better?" into a defensible, team-specific answer.

Worked example — the total-cost-of-ownership comparison

Detailed explanation. Cost debates go in circles until someone separates standing cost from compute cost. Build a simple monthly TCO model for the same workload under three deployments: pure Workflows, self-managed Airflow + Databricks compute, and managed Airflow + Databricks compute.

Standing cost. Always-on infra (Airflow control plane) — independent of throughput.
Compute cost. DBUs for the actual Databricks work — the same in all three (assuming shared clusters).
Ops cost. Engineer-time to operate the control plane.

Question. Model monthly TCO for a workload whose Databricks compute is $6,000/month, comparing the three deployments.

Input.

Deployment	Standing infra	Databricks compute	Ops load
Workflows only	$0	$6,000	~none
Self-managed Airflow	~$900 (infra)	$6,000	high (you run it)
Managed Airflow	~$1,500 (vendor)	$6,000	low

Code.

# tco.py — monthly total cost of ownership, standing vs compute vs ops
def monthly_tco(standing_usd: float, databricks_usd: float,
                ops_hours: float, hourly_rate: float = 120.0) -> dict:
    ops = ops_hours * hourly_rate
    return {
        "standing": standing_usd,
        "databricks_compute": databricks_usd,
        "ops": ops,
        "total": standing_usd + databricks_usd + ops,
    }

workflows_only = monthly_tco(standing_usd=0,    databricks_usd=6000, ops_hours=2)
self_airflow   = monthly_tco(standing_usd=900,  databricks_usd=6000, ops_hours=30)
managed_airflow= monthly_tco(standing_usd=1500, databricks_usd=6000, ops_hours=6)

for name, t in [("workflows_only", workflows_only),
                ("self_airflow", self_airflow),
                ("managed_airflow", managed_airflow)]:
    print(f"{name:16s} total=${t['total']:.0f} "
          f"(standing=${t['standing']:.0f}, ops=${t['ops']:.0f})")

Step-by-step explanation.

The Databricks compute ($6,000) is identical across all three — the work is the same, and shared clusters mean the orchestrator choice does not change DBU volume. This isolates the difference to standing + ops cost.
Workflows-only carries no standing infra and near-zero ops (2 hours/month of babysitting a managed scheduler that Databricks runs). Total ≈ $6,240.
Self-managed Airflow adds ~$900/month of always-on infra (scheduler + workers + metadata DB) and ~30 hours/month of engineer time to operate it — the hidden cost. Total ≈ $10,500, dominated by ops.
Managed Airflow costs more in infra (~$1,500 vendor bill) but slashes ops to ~6 hours/month. Total ≈ $8,220 — cheaper than self-managed once you price engineer time honestly.
The lesson interviewers grade: Airflow's real cost is rarely the compute — it is the standing control plane and the ops labor. If your workload is pure-Databricks, that cost buys you nothing, which is why Workflows-only wins for Team A. For a heterogeneous fleet, that same cost buys reach you cannot get otherwise.

Output.

Deployment	Standing	Ops (est.)	Total / month
Workflows only	$0	$240	~$6,240
Self-managed Airflow	$900	$3,600	~$10,500
Managed Airflow	$1,500	$720	~$8,220

Rule of thumb. Always split orchestration TCO into standing infra + compute + ops. The compute is usually equal; Airflow's premium is the always-on control plane and the labor to run it. Buy that premium only when you need the reach it delivers.

Worked example — the pick-the-orchestrator decision tree

Detailed explanation. Given any new pipeline, a senior architect runs a short decision tree out loud. Codifying it makes the interview answer reproducible — hand me a scenario, get a recommendation in under a minute.

Q1. Does the pipeline touch systems outside Databricks? → no = Workflows; yes = Q2.
Q2. Is Databricks the majority of the work? → yes = hybrid (Airflow conducts, Workflows executes); no = Airflow.
Q3 (parallel). Do you already operate Airflow? → yes = lower bar to hybrid/Airflow; no = weigh managed Airflow vs staying Workflows-only.

Question. Walk the tree for four scenarios and record the recommendation.

Input.

Scenario	Outside systems?	Databricks majority?	Already run Airflow?
Lakehouse-only marts	no	yes	no
Fleet ETL (7 systems)	yes	no	yes
70% DBX + SFTP + Snowflake	yes	yes	yes
New team, DBX + one API	yes	yes	no

Code.

# tree.py — the pick-the-orchestrator decision tree
def pick(outside: bool, dbx_majority: bool, runs_airflow: bool) -> str:
    if not outside:
        return "Databricks Workflows"
    if not dbx_majority:
        return "Apache Airflow"
    # outside systems AND Databricks-majority -> hybrid
    base = "Hybrid (Airflow conducts, Workflows executes)"
    if not runs_airflow:
        base += " — or start Workflows-only + thin managed-Airflow edge"
    return base

print(pick(False, True,  False))   # lakehouse-only marts
print(pick(True,  False, True))    # fleet ETL
print(pick(True,  True,  True))    # 70% DBX + external
print(pick(True,  True,  False))   # new team, DBX + one API

Step-by-step explanation.

Lakehouse-only marts: Q1 is "no external systems," so the tree short-circuits to Workflows. No reason to introduce a second control plane.
Fleet ETL across seven systems: Q1 yes, Q2 "Databricks is not the majority," so Airflow — its provider catalogue is the only thing that spans the fleet, and Databricks is just one triggered target.
70% Databricks with SFTP + Snowflake: Q1 yes, Q2 "Databricks is the majority," so hybrid. Airflow owns the external wait and Snowflake; Workflows owns the lakehouse-heavy middle on a shared cluster.
New team, Databricks plus one small API: Q1 yes, Q2 yes, but Q3 "does not already run Airflow." The pragmatic answer is Workflows-only today plus a thin managed-Airflow edge if and when the one API dependency grows — do not stand up a whole Airflow deployment for a single external call.
The tree encodes the senior instinct: minimize control planes. Add Airflow only when reach demands it, and prefer hybrid over a full migration when Databricks still owns most of the work.

Output.

Scenario	Recommendation
Lakehouse-only marts	Databricks Workflows
Fleet ETL (7 systems)	Apache Airflow
70% DBX + external	Hybrid
New team, DBX + one API	Workflows-only + optional thin Airflow edge

Rule of thumb. Walk the tree: no external systems → Workflows; external systems but Databricks-majority → hybrid; external systems and Databricks-minority → Airflow. Minimize control planes; add Airflow only when reach demands it.

Design interview question on the orchestration decision matrix

A senior interviewer might ask: "Justify an orchestration choice for a team that is 60% Databricks, 40% Snowflake + dbt Cloud + custom services, cares about cost, and has two Python engineers who know Airflow. Produce the four-axis scorecard, the TCO argument, the decision-tree walk, and the final recommendation — and defend it against 'just use Databricks Workflows for everything.'"

Solution Using a weighted four-axis scorecard, a TCO split, and a decision-tree walk

# recommend.py — combine scorecard + TCO + tree into one defensible answer
def scorecard() -> dict[str, int]:
    axes = {
        # axis: (workflows, airflow)  on a 1..3 scale for THIS team
        "cost":         (2, 2),   # both fine; shared clusters + managed Airflow
        "reach":        (1, 3),   # 40% is Snowflake/dbt Cloud/services -> Airflow
        "observability":(2, 2),   # UC lineage vs cross-system DAG view -> tie
        "skills":       (1, 3),   # two engineers already fluent in Airflow
    }
    wf = sum(w for w, _ in axes.values())
    af = sum(a for _, a in axes.values())
    return {"workflows": wf, "airflow": af}

def tco() -> dict[str, float]:
    dbx = 6000.0
    return {
        "workflows_only": dbx + 240,        # cannot reach 40% of the fleet!
        "hybrid_managed": dbx + 1500 + 720, # managed Airflow edge + ops
    }

def tree() -> str:
    outside, dbx_majority, runs_airflow = True, True, True
    if not outside:
        return "Workflows"
    if not dbx_majority:
        return "Airflow"
    return "Hybrid (Airflow conducts, Workflows executes)"

s = scorecard()
print("scorecard:", s)                       # workflows=6, airflow=10
print("tco:", tco())
print("tree:", tree())
print("RECOMMENDATION: Hybrid — Airflow conducts the 40% external fleet; "
      "Workflows executes the 60% lakehouse leg on shared clusters.")

Step-by-step trace.

Input	Value	Effect on recommendation
60% Databricks	majority	keep lakehouse work in Workflows (cost + lineage)
40% Snowflake/dbt/services	external reach	need Airflow's provider catalogue
cost-sensitive	shared clusters + managed Airflow	keeps DBU + standing cost bounded
2 Airflow-fluent engineers	skills axis → Airflow	low barrier to the Airflow edge
scorecard	wf=6, af=10	Airflow-leaning, but DBX is 60%
tree	outside + DBX-majority	hybrid

The scorecard leans Airflow (reach + skills), but Databricks is still the majority of the work, so a pure-Airflow answer would waste the lakehouse's cheap shared-cluster execution and free Unity Catalog lineage. The tree lands on hybrid; the TCO shows Workflows-only is cheaper on paper but cannot reach 40% of the fleet — a disqualifier, not a saving. The defensible recommendation is hybrid: Airflow conducts the external fleet, Workflows executes the lakehouse leg.

Output:

Component	Result
Scorecard (wf / af)	6 / 10
Workflows-only TCO	~$6,240 but misses 40% of systems
Hybrid (managed) TCO	~$8,220, reaches everything
Decision-tree result	Hybrid
Final recommendation	Hybrid — Airflow conducts, Workflows executes
Rebuttal to "Workflows for everything"	it cannot natively reach Snowflake/dbt Cloud/services

Why this works — concept by concept:

Weighted scorecard — scoring reach and skills honestly surfaces Airflow's edge for this team, while cost and observability tie. The number (6 vs 10) is derived, not asserted, which is what an interviewer grades.
TCO split — separating standing + compute + ops shows Workflows-only is cheaper only if you ignore that it cannot reach 40% of the fleet. A cheaper tool that cannot do the job is not cheaper; it is incomplete.
Decision tree — "external systems + Databricks-majority → hybrid" gives the reproducible verdict in one line, independent of the scorecard, and the two agree.
Defending against 'Workflows for everything' — the rebuttal is concrete: Workflows is lakehouse-first and does not natively orchestrate Snowflake, dbt Cloud, and custom services. Naming the specific missing reach beats hand-waving.
Cost — hybrid adds a managed-Airflow standing cost (~$2,200/month all-in over compute) to buy reach across 40% of the platform and keep the 60% lakehouse leg on cheap shared clusters with native lineage. Versus a pure-Airflow design (loses free lineage, risks cluster-per-task) or pure-Workflows (cannot reach the fleet), hybrid is the constrained optimum. O(1) control planes beyond the necessary two; O(systems) reach.

Design
Topic — design
Design problems on tool-selection and TCO trade-offs

Practice →

ETL
Topic — etl
ETL problems on orchestration and scheduling design

Practice →

Cheat sheet — orchestration recipes

Which orchestrator when. Inside the lakehouse (work is overwhelmingly Databricks) → Databricks Workflows; across a heterogeneous fleet (Databricks is one of many systems) → Apache Airflow; both (external reach and a Databricks-majority workload) → hybrid (Airflow conducts, Workflows executes). Never pick by feature count — score workload locality, multi-system reach, operational ownership, and cost model, and let the constraints choose.
Databricks job DAB skeleton. resources.jobs.<name> with tasks: each carrying a task type (notebook_task / spark_python_task / python_wheel_task / sql_task / dbt_task / pipeline_task / run_job_task), depends_on: [{task_key: ...}] for the DAG, job_cluster_key pointing at a single shared entry under job_clusters:, per-task max_retries, and email_notifications.on_failure. Deploy with databricks bundle deploy -t prod.
Databricks triggers. schedule.quartz_cron_expression + timezone_id for cron; trigger.file_arrival.url (+ min_time_between_triggers_seconds) for event-driven ingestion; trigger.table_update for reactive downstream refresh; continuous for always-on streaming; external run-now for orchestrator handoff. Pair file-arrival triggers with Auto Loader (cloudFiles + checkpointLocation) for idempotent incremental reads.
Databricks control flow. for_each_task (with inputs + bounded concurrency) for dynamic fan-out; condition_task (op / left / right) to branch; run_if (ALL_SUCCESS / AT_LEAST_ONE_SUCCESS / NONE_FAILED / ALL_DONE) to control whether a task runs given upstream outcomes; dbutils.jobs.taskValues.set/get to pass small values along edges (large data goes through lakehouse tables).
Airflow DAG skeleton. @dag(schedule=..., start_date=..., catchup=False, default_args={"retries": N}); tasks are @task functions (TaskFlow) or operator instances; dependencies via a >> b, chain(...), or a >> [b, c] >> d; keep tasks idempotent and keyed by the data interval ({{ ds }}). Use Connection/Variable for config, XCom for small values only.
Airflow Databricks provider. Prefer DatabricksRunNowOperator(job_id=...) in production (job owned in Databricks → cluster reuse + native lineage) over DatabricksSubmitRunOperator (inline run) for ad-hoc; DatabricksSqlOperator for SQL warehouse queries; DatabricksWorkflowTaskGroup + DatabricksNotebookOperator to run several Databricks tasks on one shared cluster from inside the DAG. Auth via the databricks_default connection.
Airflow executors. SequentialExecutor (dev), LocalExecutor (one host), CeleryExecutor (worker fleet behind a broker), KubernetesExecutor (one pod per task; right-size heavy tasks via executor_config={"pod_override": ...}). Managed Airflow (MWAA / Astronomer / Composer) picks and operates one for you.
Deferrable waits. Use deferrable sensors/operators (deferrable=True) plus a triggerer so long external waits (S3/SFTP/SQL) suspend and free the worker slot — thousands of waits, a handful of workers. This is how Airflow avoids a worker-per-wait blowup.
The one-cluster-per-task tax. N DatabricksSubmitRunOperator tasks each with new_cluster start N clusters (N start-up taxes + N idle edges). Fix: consolidate into one Databricks job triggered by DatabricksRunNowOperator, or a DatabricksWorkflowTaskGroup sharing one job_cluster_key. One start replaces N; the DBU bill drops proportionally.
Retry placement (hybrid). Transient in-task failures (Spark shuffle blip) → retry in Databricks (per-task max_retries, warm cluster). Handoff/API failures (run-now timeout) → retry in Airflow (operator retries + backoff). Deterministic errors (bad input) → fail fast, retry nowhere. Never stack unbounded retries on both layers.
Alerting + lineage split. Databricks: webhook_notifications / email_notifications for task-level failures with lakehouse context; automatic Unity Catalog table-level lineage. Airflow: DAG-level SLAs, cross-system alerts, graph/Gantt UI; add OpenLineage/Marquez for table lineage. Route each alert to the layer whose team can fix it.
TCO model. Always split orchestration cost into standing infra + triggered compute + ops labor. Databricks compute is roughly equal across choices (shared clusters); Airflow's premium is the always-on control plane and the engineer-time to run it (self-managed) or the vendor bill (managed). Buy that premium only when you need the reach.
Migration cost between models. Cron/REST → Databricks jobs: ~1–2 weeks (lift each cron into a task DAG, share a cluster). Add an Airflow edge for cross-system deps: ~1–2 weeks with managed Airflow. Full Airflow ↔ Workflows re-platform: a quarter (re-express DAGs, retries, alerting, lineage). Choose the control plane deliberately; the migration cost is real.

Frequently asked questions

What is the core difference in `databricks workflows vs airflow`?

The core difference is where the orchestrator sits. Databricks Workflows is an in-lakehouse orchestrator: a job is a DAG of tasks (notebook, Spark Python, SQL, dbt, DLT) that runs on ephemeral job clusters right next to the governed data, with native triggers (cron, file-arrival, table-update, continuous) and automatic Unity Catalog lineage — and zero extra infrastructure to operate. Apache Airflow is an out-of-lakehouse control plane: a dag is a Python object whose operators reach any system (S3, Snowflake, dbt Cloud, HTTP, and Databricks via the provider), run through a scheduler and a pluggable executor, and give you portability at the cost of operating (or buying) a standing system. In one line: Workflows optimizes for lakehouse-native execution and cost; Airflow optimizes for multi-system reach and portability. The senior framing is inside-vs-outside the lakehouse, not a feature bake-off.

When should I choose Databricks Workflows over Airflow?

Choose Databricks Workflows when the overwhelming majority of your work runs on Databricks and you touch few or no external systems. In that case an external orchestrator adds a standing infrastructure cost, an extra network hop, and a second system to operate — for almost no reach benefit. Workflows ships with the platform, bills Jobs Compute DBUs only while tasks run, reuses one ephemeral cluster across many tasks via a shared job_cluster_key, and captures table-level lineage automatically through Unity Catalog. It also has native event triggers (file-arrival, table-update) that replace polling sensors entirely. Reach for Airflow instead the moment your DAG must natively orchestrate Snowflake loads, dbt Cloud jobs, Kafka, SFTP drops, or custom microservices alongside Databricks — that heterogeneous reach is exactly what Workflows is not designed for.

What does the `airflow databricks operator` actually do?

The Databricks provider for Airflow (apache-airflow-providers-databricks) gives you operators that let an Airflow dag drive Databricks. The two you use most are DatabricksRunNowOperator, which triggers an existing Databricks job by job_id (the production-preferred hook, because the job — its tasks, cluster, retries, and lineage — stays owned in Databricks and Airflow just fires it), and DatabricksSubmitRunOperator, which submits a one-off run defined inline in the DAG (handy for ad-hoc work but prone to the one-cluster-per-task cost trap). There is also DatabricksSqlOperator for SQL-warehouse queries and DatabricksWorkflowTaskGroup (with DatabricksNotebookOperator) to run several Databricks tasks on a single shared cluster from inside the DAG. All authenticate through an Airflow connection, conventionally databricks_default.

How do triggers and task dependencies differ between the two?

Task dependencies: Databricks Workflows declares them with depends_on: [{task_key: ...}] inside a job (plus run_if to control execution given upstream outcomes, and for_each_task / condition_task for fan-out and branching). Airflow declares them in Python with >>, chain(), and .expand() for dynamic mapping. Both build a true DAG; both reject cycles. Triggers: Workflows has native ones — quartz cron, file-arrival at a Unity Catalog location, table-update on Delta tables, and continuous for always-on jobs — so event-driven ingestion needs no polling. Airflow triggers via schedule (cron or a custom timetable) and reacts to external state through sensors (ideally deferrable, so long waits do not hold a worker). The practical upshot: Workflows has cheaper native event triggers inside the lakehouse; Airflow has broader ways to wait on external systems.

Can I run Airflow and Databricks Workflows together?

Yes — the hybrid model is the most common answer at scale. You draw an ownership boundary: Airflow owns the cross-system schedule, external-event waits (deferrable sensors), fleet-wide dependencies, and platform SLAs; Databricks Workflows owns the in-lakehouse task DAG, the ephemeral cluster, per-task retries, and Unity Catalog lineage. Airflow hands off to Databricks either by triggering an existing job with DatabricksRunNowOperator (loose coupling; the job is one source of truth) or by declaring a DatabricksWorkflowTaskGroup so several Airflow-launched Databricks tasks share one job cluster (tight coupling; visible in both UIs). The hybrid rule of thumb for retries: keep transient Spark retries in Databricks and handoff/API retries in Airflow, and never stack unbounded retries on both. This gives Airflow's reach and the lakehouse's cost and lineage benefits at the same time.

How do I keep the DBU bill flat when orchestrating from Airflow?

The number-one cost mistake is one cluster per task: N DatabricksSubmitRunOperator tasks, each with its own new_cluster, start N clusters — N start-up taxes and N idle-edge windows. The fix is to make the lakehouse leg reuse one cluster. Two ways: consolidate the steps into a single Databricks job whose tasks all set the same job_cluster_key, then trigger it once with DatabricksRunNowOperator; or wrap the Databricks tasks in a DatabricksWorkflowTaskGroup that declares one shared job_clusters entry, so the provider materializes a single Databricks Workflow on one cluster. Either way, one cluster starts, runs the whole lakehouse leg in dependency order, and tears down — O(1) cluster starts per run instead of O(tasks). Add per-task max_retries on the warm cluster, use serverless or right-sized job clusters, and gate expensive downstream tasks behind a condition_task so empty runs skip them. The DBU bill then tracks data volume, not the number of orchestrated steps.

Practice on PipeCode

Drill the ETL practice library → for the scheduling, dependency-ordering, incremental-load, and medallion-pipeline problems that both orchestrators live and die on.
Rehearse on the design practice library → for the orchestration-architecture, ownership-boundary, and TCO trade-off questions senior interviewers open with when databricks workflows vs airflow is on the table.
Sharpen the streaming axis with the streaming practice library → for the event-driven triggers, file-arrival ingestion, and continuous-job handoff patterns that separate reactive pipelines from polling ones.
Cement the transform layer with the data-transformation practice library → for the bronze-silver-gold and dbt-on-Databricks work that the tasks inside your jobs and DAGs actually run.
Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the four-axis decision matrix against real graded inputs.

Lock in orchestration decision muscle memory

Docs explain features. PipeCode drills explain the decision — when Databricks Workflows wins because the work lives in the lakehouse, when Airflow's provider reach is the only thing that spans the fleet, when the hybrid ownership boundary keeps the DBU bill flat, and when a shared job cluster turns eight cluster starts into one. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the production trade-offs senior data engineers actually face.

Practice ETL problems →
Practice design problems →

1. Why orchestration choice is an architecture decision, not a tool preference

Inside vs outside the lakehouse — the placement of the orchestrator binds you for years

Worked example — the four-axis comparison table

Worked example — what interviewers actually probe

Senior interview question on orchestration architecture

Solution Using a locality-scored decision plus a staged hybrid migration

2. Databricks Workflows — jobs, task types, dependencies, triggers, native lakehouse integration

databricks jobs model a pipeline as a task DAG that runs where the data lives — ephemeral clusters, native triggers, Unity Catalog lineage

Worked example — a medallion job in Databricks Asset Bundle YAML

Worked example — the file-arrival trigger for event-driven ingestion

Worked example — for-each fan-out and a condition branch

SQL interview question on Databricks Workflows

Solution Using a file-arrival-triggered job with shared cluster, for-each, and a condition gate

3. Apache Airflow — DAGs, operators, the DatabricksSubmitRunOperator, scheduler & executors

apache airflow models the same problem as a Python DAG that reaches every system — operators, a scheduler, pluggable executors, and a first-class Databricks provider

Worked example — an Airflow TaskFlow DAG driving a Databricks job

Worked example — choosing and configuring an executor

Worked example — dynamic task mapping over a partition list

Python interview question on Airflow orchestration

Solution Using deferrable sensors, RunNow, dynamic mapping, and a bounded pool

4. The hybrid reality — Airflow orchestrating Databricks jobs, when each layer owns what

Airflow conducts the fleet, Databricks Workflows executes the lakehouse — and DatabricksWorkflowTaskGroup reuses one cluster across Airflow-launched tasks

Worked example — DatabricksWorkflowTaskGroup with a shared cluster

Worked example — placing retries and alerts on the right layer

Worked example — one shared cluster across an Airflow-launched job

Design interview question on hybrid orchestration

Solution Using an ownership boundary with a WorkflowTaskGroup and layered retries

5. Decision matrix — cost, multi-system DAGs, observability, team skills + interview signals

Score four axes and let the constraints pick — Workflows, Airflow, or hybrid

Worked example — scoring the four axes for three teams

Worked example — the total-cost-of-ownership comparison

Worked example — the pick-the-orchestrator decision tree

Design interview question on the orchestration decision matrix

Solution Using a weighted four-axis scorecard, a TCO split, and a decision-tree walk

Cheat sheet — orchestration recipes

Frequently asked questions

What is the core difference in databricks workflows vs airflow?

When should I choose Databricks Workflows over Airflow?

What does the airflow databricks operator actually do?

How do triggers and task dependencies differ between the two?

Can I run Airflow and Databricks Workflows together?

How do I keep the DBU bill flat when orchestrating from Airflow?

Practice on PipeCode

Lock in orchestration decision muscle memory

`databricks jobs` model a pipeline as a task DAG that runs where the data lives — ephemeral clusters, native triggers, Unity Catalog lineage

`apache airflow` models the same problem as a Python DAG that reaches every system — operators, a scheduler, pluggable executors, and a first-class Databricks provider

Airflow conducts the fleet, Databricks Workflows executes the lakehouse — and `DatabricksWorkflowTaskGroup` reuses one cluster across Airflow-launched tasks

What is the core difference in `databricks workflows vs airflow`?

What does the `airflow databricks operator` actually do?