Airflow interview questions for Data Engineer roles

Airflow schedules a run for a period of data, the data interval, and starts it after the period has ended, so the run can see all of that period's data. A @daily run for 1 January starts just after midnight on 2 January. The first interval begins at start_date, so a Dag with start_date today has its first run tomorrow, covering today. The run's logical date is the start of its interval, not the moment it ran. For the 30 missed days, run a backfill over those dates rather than turning catchup on.

1. One daily run, step by step

Inside a task, the interval is data_interval_start and data_interval_end, and those, not today's clock, say which day to load.

2. Why the new Dag waited

from airflow.sdk import DAG
import pendulum

with DAG(
    dag_id="daily_orders",
    schedule="@daily",
    start_date=pendulum.datetime(2026, 10, 16, tz="UTC"),
    catchup=False,
):
    ...

The first interval is 16 October. Its run starts after it ends, on 17 October. Nothing was wrong.

3. Catchup and backfill

• Catchup (catchup=True) makes the scheduler create a run for every interval since start_date that has not run. It is off by default, and it should stay off for a Dag that is not written to load only its own interval.
• Backfill runs a chosen range once, on purpose:

airflow backfill create --dag-id daily_orders \
  --from-date 2026-09-16 --to-date 2026-10-15 \
  --max-active-runs 3 \
  --reprocess-behavior failed

--max-active-runs keeps 30 runs from hitting the source at once, and --reprocess-behavior says what to do with intervals that already ran (here, rerun only the failed ones).

4. What makes a backfill safe

Each run must load exactly its own interval and overwrite it, so running 30 old days gives 30 correct days, and running one twice changes nothing. See idempotent pipelines.

What interviewers want: a run covers an interval and starts after it ends, the first run one interval after start_date, logical date as the interval's start, catchup off by default, backfill for a chosen range with a limit on parallel runs, and tasks that load their own interval.

Make every task idempotent: running it again for the same interval leaves the same result. 3 rules from Airflow's best practices cover both bugs. Never append with INSERT; write with an UPSERT (MERGE) on a key, or replace the partition the run owns. Read and write by the run's data interval (data_interval_start), never by now() or "the latest data", so a rerun next week still loads its own day. And treat the task like a database transaction: it commits all of its output or none of it, so a failure halfway leaves nothing to clean up.

1. Why the retry duplicated rows

The task inserted rows in batches. It failed after half of them, and the retry inserted all of them again. The first half is now in the table twice.

2. Why the rerun loaded the wrong day

# wrong: depends on when the task runs
day = (datetime.now() - timedelta(days=1)).date()

Rerun on 20 October for the run of 12 October, it loads 19 October.

3. The rewrite

from airflow.sdk import task
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

load_orders = SQLExecuteQueryOperator(
    task_id="load_orders",
    conn_id="warehouse",
    sql='''
        BEGIN;
        DELETE FROM orders WHERE order_date = '{{ data_interval_start | ds }}';
        INSERT INTO orders
        SELECT * FROM staging_orders
        WHERE order_date = '{{ data_interval_start | ds }}';
        COMMIT;
    ''',
)

• The day comes from the interval, so the run of 12 October always loads 12 October, whenever it runs.
• Delete then insert inside one transaction replaces that day as a whole: a retry, a rerun or a backfill gives the same rows. A MERGE on the order id does the same for tables updated row by row.
• The transaction means a failure halfway rolls back, so there is never a half-loaded day.

4. The same rule for files

Write files to a path keyed by the interval, such as s3://lake/orders/date=2026-10-12/, and overwrite that path. Airflow's best practices give the same advice for S3 and HDFS.

5. A habit to check in review

What interviewers want: idempotency as the goal, UPSERT or partition overwrite instead of INSERT, the run's data interval instead of now(), all-or-nothing writes, the same rule for file paths, and why it makes retries and backfills safe.

A sensor in the default poke mode holds a worker slot for its whole wait, so 200 waiting sensors are 200 slots doing nothing. There are 2 better modes. Reschedule mode takes a slot only while it checks, then frees it until the next check. Deferrable mode hands the wait to the triggerer, a separate component where many small asynchronous triggers run in one process, so the task holds no worker slot and, by default, no pool slot while it waits. For hundreds of long waits, choose deferrable where the sensor supports it, reschedule where it does not, and set a timeout on every sensor.

1. The 3 modes

Airflow's docs put the poke and reschedule choice as latency: checking every second suits poke, every minute suits reschedule.

2. The change

from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor

wait_for_file = S3KeySensor(
    task_id="wait_for_partner_file",
    bucket_key="s3://inbound/partner_a/{{ ds }}/orders.csv",
    deferrable=True,          # the triggerer waits, not a worker
    poke_interval=300,        # check every 5 minutes
    timeout=8 * 60 * 60,      # fail after 8 hours instead of waiting forever
)

For a sensor with no deferrable mode:

wait_for_file = SomeSensor(task_id="wait", mode="reschedule", poke_interval=300, timeout=8 * 60 * 60)

The operators.default_deferrable setting turns deferrable mode on for every operator that supports it.

3. What to check after

• A triggerer is running. Airflow needs at least one triggerer process, beside the scheduler, for deferred tasks to work.
• Pools. Deferred tasks do not take pool slots by default; if a pool exists to limit calls to a partner's server, set the pool to count deferred tasks.
• Timeouts. A sensor with no timeout can wait for a file that never comes; a timeout turns that into a failure someone sees.

4. A better design where possible

If the files land through a process you control, have it update an Airflow asset when it finishes and schedule the Dag on that asset (see the question on making one Dag wait for another), and there is nothing to sense.

What interviewers want: poke holding a worker slot, reschedule freeing it between checks, deferrable moving the wait to the triggerer with no worker or pool slot, the latency trade-off, timeouts on every sensor, a running triggerer, and replacing the sensor with an event where possible.

Schedule the reporting Dag on data, not the clock. In Airflow, the tasks that write orders and payments declare an asset in their outlets, and the reporting Dag sets schedule=[orders, payments]. Airflow marks an asset updated only when the producing task succeeds, and with 2 assets the reporting Dag runs once both have been updated since its last run. A failed or late load means no report, rather than a report on old data. A sensor waiting on the other Dags also works, but in poke mode it holds a worker slot for the whole wait.

1. The producers

from airflow.sdk import DAG, Asset, task

orders = Asset("s3://lake/orders")
payments = Asset("s3://lake/payments")

with DAG(dag_id="load_orders", schedule="@daily", ...):
    @task(outlets=[orders])
    def write_orders(): ...
    write_orders()

with DAG(dag_id="load_payments", schedule="@daily", ...):
    @task(outlets=[payments])
    def write_payments(): ...
    write_payments()

2. The consumer

with DAG(dag_id="daily_report", schedule=[orders, payments], ...):
    @task
    def build_report(): ...
    build_report()

3. What happens on a bad day

The report never runs on half-fresh data, and when it starts late, the delay shows where the problem was.

4. The options compared

5. Points to raise

• An asset updates only on success: a skipped or failed task does not count.
• The asset name is a URI that names the data, so a Dag outside the team can depend on it without knowing which Dag writes it.
• Keep an alert for "no report by 10:00", since an event-driven Dag that never starts makes no noise of its own.

What interviewers want: scheduling on data with assets and outlets, an update only on success, a Dag on several assets waiting for all of them, the problem with a fixed start time, the sensor alternative and its cost, and an alert for a report that never starts.

XComs are for small values, such as a path, a count or a list of file names, not dataframes: by default they are stored in Airflow's metadata database, which suffers with large values. Write the data to object storage or a table and pass its location through XCom. For the files, use dynamic task mapping: one task lists the files, and expand() creates one task copy per file at run time, so each file succeeds, fails and retries on its own. Cap the fan-out with max_active_tis_per_dag, and keep in mind that expand() creates at most 1024 copies by default.

1. The redesign

from airflow.sdk import dag, task
import pendulum

@dag(schedule="@daily", start_date=pendulum.datetime(2026, 1, 1, tz="UTC"), catchup=False)
def partner_files():

    @task
    def list_files(data_interval_start=None) -> list[str]:
        day = data_interval_start.strftime("%Y-%m-%d")
        return list_keys(f"s3://inbound/{day}/")   # a list of paths: small

    @task(max_active_tis_per_dag=16, retries=3)
    def load_file(path: str, table: str) -> int:
        rows = copy_into(table, path)              # the data moves inside the warehouse
        return rows                                # a count: small

    @task
    def check_total(counts: list[int]) -> None:
        assert sum(counts) > 0

    counts = load_file.partial(table="raw.orders").expand(path=list_files())
    check_total(counts)

partner_files()

• partial() fixes what is the same for every copy (the table); expand() maps over what varies (the path).
• Each file is its own task instance, shown with its index in the UI: a bad file fails alone and is retried alone.
• max_active_tis_per_dag=16 keeps 300 files from taking every worker slot; it counts copies across all active runs of the Dag.

2. Why not XCom for the dataframe

If tasks must hand over larger values, an object storage XCom backend keeps them out of the database; it is still better for the data to stay in the warehouse and move there.

3. Limits to know

• [core] max_map_length, 1024 by default, caps copies per expand(). A day with 3000 files fails the mapping: batch the paths (100 per copy) or raise the limit knowingly.
• Several mapped arguments give a cross product: expand(path=[...300], table=[...3]) makes 900 copies.
• List the files inside a task, not at the top of the Dag file. Airflow parses Dag files often, and top-level calls to S3 would run on every parse.

What interviewers want: XCom for small values and object storage or tables for data, passing locations instead of data, expand and partial for one task per file, failures and retries per file, max_active_tis_per_dag, the 1024 default of max_map_length and batching past it, cross products, and no work at the top level of a Dag file.