Advanced Airflow Operations

Data Engineering & Modern Data Platforms · Lesson 21 · Data Ingestion, ETL and Orchestration

Lesson Overview

tlp_orders_daily from Apache Airflow Architecture and DAGs runs on a quiet morning. Operations begin when the morning is not quiet. The reseller file arrives at 06:40 instead of 05:10. A backfill of eight open days coincides with the daily run. Two feature jobs start the moment finance publishes, and a third job polls a SQL sensor in poke mode until it has occupied a worker for an hour. The scheduler is healthy. The warehouse is not, because nothing in the DAG asked for a slot limit, and nothing in the sensor gave the slot back.

This lesson is the operating manual for that DAG on Airflow 2.9, with notes where a feature arrived earlier or later in 2.x. Retries and backoff, clearing versus backfill, dataset scheduling, pools, sensors and the triggerer, dynamic task mapping, secrets, and the small set of signals that mean the deployment is actually scheduling. Airflow's own sla timedelta is included so you can recognize it and so you do not confuse it with the 07:00 UTC freshness promise from Data Pipeline Orchestration Fundamentals.

The merge and the grain do not change here. If a retry or a backfill is unsafe, the fix is still the idempotent load from Data Cleansing and Transformation, not a larger worker.

Learning Objectives

By the end of this lesson, you should be able to:

1. Configure retries, exponential backoff, and a max delay, and keep reconciliation on zero retries.
2. Clear a task instance and explain how that differs from a backfill.
3. Run airflow dags backfill for an explicit date range with a concurrency cap.
4. Publish an Airflow Dataset from a task outlet and schedule a consumer DAG on it.
5. Combine a dataset event with a timetable using DatasetOrTimeSchedule on Airflow 2.9.
6. Create a pool and assign slots so type-2 loads and warehouse writes cannot stampede.
7. Choose sensor mode="reschedule" or a deferrable sensor, and run a triggerer for the latter.
8. Map a task dynamically with .expand() when the file list is known only at runtime.
9. Store secrets in Connections or a secrets backend, not in DAG source or Variables that get logged.
10. Explain why Airflow's sla parameter is a weak substitute for a freshness row.
11. Watch scheduler heartbeat, import errors, queued tasks, and pool occupancy.
12. Name which of these controls remain when a task body becomes a Spark submit.

Prerequisites

• Apache Airflow Architecture and DAGs: DAG, DagRun, task instance, XCom, data_interval_start, and tlp_orders_daily.
• Data Pipeline Orchestration Fundamentals: retries versus data failures, pools as a concept, backfill safety, freshness SLA.
• A local Airflow 2.9 is enough for the exercises. Dataset scheduling needs 2.4 or newer. DatasetOrTimeSchedule needs 2.9.

Detailed Core Concepts

Attempts are not new runs

A retry is another attempt of the same task instance, same DagRun, same data interval. retries, retry_delay, retry_exponential_backoff, and max_retry_delay live on the task, usually via default_args, with overrides for tasks that must not retry.

pythonCopy
default_args = {
    "retries": 2,
    "retry_delay": pendulum.duration(minutes=5),
    "retry_exponential_backoff": True,
    "max_retry_delay": pendulum.duration(minutes=30),
}

With two retries, a task may execute three times: the first try and two retries. Exponential backoff grows the delay and then respects max_retry_delay. The attempt number is in the log path. Read the latest attempt first.

on_retry_callback and on_failure_callback fire on those transitions. Callbacks run in the worker process for task-level callbacks and must be importable and fast. A callback that opens a warehouse transaction and "fixes" rows is a hidden task. Page a person or emit a metric. Do not repair revenue inside a callback.

Mark data failures so they do not look like lock timeouts. Raising AirflowFailException fails the task immediately and does not consume the remaining retries. Raising AirflowSkipException skips. A reconciliation mismatch should raise AirflowFailException if you want the task to stop even when retries was inherited from default_args. That is a useful belt next to an explicit retries=0.

Clear versus backfill

Clear targets existing task instances. airflow tasks clear tlp_orders_daily -s 2026-03-01 -e 2026-03-01 -t load_fact_order_line -y resets that instance so the scheduler can run it again. Downstream can be included. The data interval does not change. Use clear when a run exists and one node failed or must be repeated after a code fix.

Backfill creates DagRuns for intervals that you name. On Airflow 2.9 the CLI is:

bashCopy
airflow dags backfill tlp_orders_daily \
  --start-date 2026-03-01 \
  --end-date 2026-03-07 \
  --reset-dagruns \
  -y

Confirm the flags in your exact patch release with airflow dags backfill --help. The intent is stable: a start, an end, and a decision about existing runs. --reset-dagruns clears runs that already exist in the range so they execute again. Without a reset, existing successful runs are left alone and missing ones are created. That is usually what you want when the scheduler was down and some days actually completed.

Backfill still obeys max_active_runs, pools, and depends_on_past. depends_on_past=True means today's task instance waits for yesterday's same task to succeed. During a backfill that is sometimes what you want for a type-2 load that must see days in order, and it is a trap when yesterday failed for an unrelated reason and the whole range stalls. TechLearningPro does not set depends_on_past on the fact task. It serializes the dimension task with a pool of one slot instead, so order is a resource rule you can see in the pool, and a single bad day does not pin every later day through a hidden flag.

Frozen periods stay out of the command. The SQL predicate from lesson 18 remains the second lock, so a mistaken date in the CLI cannot rewrite September.

Datasets schedule consumers when producers succeed

An Airflow Dataset, added in 2.4, is a URI string that stands for a logical dataset. Airflow does not open the URI. It records that a task updated it. A consumer DAG whose schedule is a list of datasets runs after each listed dataset has been updated at least once since the consumer's previous run. A failed or skipped producer task does not update the dataset, so the consumer does not start. That matches the lesson 19 rule: features run after a successful publish, not after a red reconcile.

pythonCopy
from airflow.datasets import Dataset

FACT_ORDER_LINE = Dataset("postgres://warehouse/analytics.fact_order_line")

The URI is an identifier. postgres://warehouse/analytics.fact_order_line and a second string with different case are different datasets. Globs do not expand. One URI, one dataset. Put the same Python object, or the exact same URI, on the producer outlet and the consumer schedule.

On the producer, the publish task declares the outlet. On TaskFlow:

pythonCopy
@task(outlets=[FACT_ORDER_LINE], retries=0)
def publish_freshness(recon_ok: list) -> dict:
    from tlp.warehouse import upsert_freshness
    return upsert_freshness(recon_ok, **_window())

On the consumer:

pythonCopy
@dag(
    dag_id="tlp_features_daily",
    schedule=[FACT_ORDER_LINE],
    start_date=pendulum.datetime(2026, 3, 1, tz="UTC"),
    catchup=False,
    tags=["techlearningpro"],
)
def tlp_features_daily():
    ...

A list is an AND: every dataset must update. Airflow 2.9 also allows expressions, FACT_ORDER_LINE & LEARNER_DIM and FACT_ORDER_LINE | CATALOG, for boolean combinations. A consumer that only needs the fact should schedule on the fact dataset alone. Adding the learner dataset "to be safe" stalls the feature job whenever the learner outlet was not updated in that run, even if the fact publish succeeded.

Dataset-scheduled runs are event-driven. Their data interval is not the same idea as a cron interval. Read the producer's business date from the freshness table or from the dataset event, which Airflow stores, rather than assuming data_interval_start means "yesterday" the way it does on tlp_orders_daily. Document that in the consumer. A feature job that calls date.today() will stamp the wall clock on a late publish, which is the bug lesson 20 already removed from the warehouse DAG.

Airflow 2.9 adds DatasetOrTimeSchedule when a DAG must run on a dataset update or on a cron, for example a cleanup that should run at midnight even if no file arrived:

pythonCopy
from airflow.timetables.datasets import DatasetOrTimeSchedule
from airflow.timetables.trigger import CronTriggerTimetable

schedule=DatasetOrTimeSchedule(
    timetable=CronTriggerTimetable("0 0 * * *", timezone="UTC"),
    datasets=[FACT_ORDER_LINE],
)

Use it when both reasons are real. Do not use it to paper over an uncertain dependency. The feature job wants the dataset only. The freshness alarm wants the clock only.

Pools are the warehouse's seat belt

bashCopy
airflow pools set warehouse_write 4 "heavy warehouse sessions"
airflow pools set dim_learner_scd 1 "serialize type-2 learner loads"

pythonCopy
@task(pool="dim_learner_scd", pool_slots=1)
def load_dim_learner(changes: dict) -> dict:
    ...

@task(pool="warehouse_write", pool_slots=2)
def load_fact_order_line(...) -> dict:
    ...

A pool of 4 with a fact task that takes 2 slots allows two fact tasks, or one fact task and two lighter tasks that take one slot each. The dimension pool of 1 allows a single SCD apply anywhere on the deployment, including a backfill overlapping today's run. Slots are held from the moment the task starts until it finishes. A sensor in poke mode that uses the warehouse pool holds a slot while it sleeps. That is why sensors do not belong in the write pool, and why poke mode is the wrong mode for a long wait.

Pools are global to the deployment, not local to the DAG. Name them for the resource. pool="default_pool" is Airflow's built-in pool, and every task that does not name a pool uses it. A mapped task with hundreds of instances can exhaust default_pool and stall unrelated DAGs. Set max_active_tis_per_dag on mapped tasks.

Sensors wait, and they should not hog a worker

A sensor is a task that succeeds when a condition becomes true and fails when timeout is reached. poke_interval is how often it checks.

mode="poke" holds the worker slot between checks. A five-minute poke interval for an hour is an occupied slot that mostly sleeps.

mode="reschedule" records the next check time and frees the worker. The scheduler resubmits the task later. Use this for the reseller file when the producer cannot emit a dataset.

pythonCopy
from airflow.providers.common.sql.sensors.sql import SqlSensor

wait_for_file_row = SqlSensor(
    task_id="wait_for_reseller_manifest",
    conn_id="warehouse",
    sql="""
        SELECT 1
        FROM raw.file_manifest
        WHERE dataset_name = 'reseller_enrollments'
          AND business_date = DATE '{{ data_interval_start }}'
        LIMIT 1
    """,
    mode="reschedule",
    poke_interval=120,
    timeout=60 * 80,
    retries=0,
)

The timeout is 80 minutes so that a 06:00 start fails in time to matter for a 07:00 promise, matching lesson 19. The SQL is templated per task instance. retries=0 avoids a second long wait after the timeout already declared the file late.

ExternalTaskSensor waits for a task in another DAG. It matches logical dates by default, which is wrong when the two DAGs have different timetables. Prefer a dataset outlet when you own both DAGs. Use ExternalTaskSensor when the upstream DAG is cron-scheduled on the same interval and you cannot add an outlet yet. Set mode="reschedule".

Deferrable sensors (deferrable=True on sensors that support it) release the worker and hand a trigger to the triggerer process. You must run airflow triggerer. Without it, deferred tasks sit in deferred until they time out. Deferral is the right long-term wait. Reschedule mode is the right default if you are not running a triggerer yet. Poke mode is for waits of a few seconds.

Dynamic task mapping

When the landing step finds three files only at runtime, mapping creates one task instance per file without a parse-time query.

pythonCopy
@task
def list_reseller_files() -> list[str]:
    from tlp.ingest import list_files
    return list_files(**_window())

@task
def cleanse_one(path: str) -> dict:
    from tlp.cleanse import cleanse_reseller_file
    return cleanse_reseller_file(path, **_window())

cleanse_one.expand(path=list_reseller_files())

.expand() requires Airflow 2.3 or newer. The list is an XCom from list_reseller_files, so keep it to paths, not file bodies. Mapped tasks multiply concurrency. Cap them with max_active_tis_per_dag and a pool. A downstream task that needs all results can take the mapped output as a list.

Do not map over a query executed at import time. The list task runs inside the DagRun, for that interval.

Secrets

The DAG holds conn_id="warehouse". The password lives in the Connection, or in a secrets backend the Connection resolves. Airflow can delegate Connections and Variables to an external backend through airflow.secrets and provider backends such as the AWS Secrets Manager backend in apache-airflow-providers-amazon. The configuration is deployment-level, under the secrets section of airflow.cfg, not a string in the DAG.

What you need to know as an author:

• Never template a password into a bash command. It will land in the rendered template and the log.
• Never log conn.get_password().
• Variables are for non-secret config. If a value would hurt when printed, it is a secret and belongs in a Connection or the backend.
• Workers need permission to read the secret. A scheduler that can read it and a worker that cannot is an environment bug, and the task fails at runtime, not at parse.

SLA parameters versus freshness

Airflow tasks accept sla as a timedelta. The scheduler records an SLA miss against the DagRun's timetable, historically relative to the logical date or data interval end depending on version, and the behavior has well-known gaps: misses are easy to misunderstand for manually triggered and dataset-scheduled runs, and the callback story is not "page at the deadline while the task is still running." The project has treated this as a flawed mechanism slated for replacement by deadline alerts. Do not implement the 07:00 finance promise as sla=timedelta(hours=1) and consider the job done.

Keep ops.dataset_freshness written by publish_freshness. Run a small clock DAG, tlp_freshness_alarm, every five minutes, that reads the row and raises AirflowFailException when the expected business date is unpublished at 06:30 UTC. That alarm is understandable. The built-in sla field is optional extra noise unless you have tested, on your patch version, exactly when a miss is recorded.

What to monitor

Airflow can export metrics through StatsD, and OpenTelemetry export has been available since the 2.6 line via the metrics configuration. Turn one of them on in production. The UI is not an overnight alert. Page on scheduler heartbeat loss, import errors, and the freshness alarm. Do not page on every retry; page on the final failure.

airflow dags list-import-errors is the first command in an incident where "the DAG disappeared." airflow tasks state and the attempt log are the first commands when a task is red. The grid is a view of the metadata database. If the web server is down and the scheduler is up, work still runs. If the heartbeat is dead, the grid can look calm because nothing is changing.

Spark does not retire these controls

A task that submits Spark still uses a pool slot for the submitter, a timeout, retries only for submit failures you know are safe, and a dataset outlet when the Spark job has written the table. The cluster has its own concurrency limit. Reflect it as a pool, or as Spark's own fair scheduling, and pick one place to enforce it so two layers do not each assume the other did. The freshness alarm stays in Airflow. The Spark UI does not know that finance needed the table by 07:00.

Architecture and Design Diagrams

mermaidCopy
flowchart TD
    cron[Timetable 06:00 UTC] --> land[land_reseller_file]
    land --> wait[Sensor reschedule mode]
    wait --> cleanse[cleanse mapped per file]
    cleanse --> dim[load_dim_learner pool 1]
    dim --> fact[load_fact_order_line pool slots 2]
    fact --> recon[reconcile retries 0]
    recon --> pub[publish_freshness outlet]
    pub --> ds[(Dataset fact_order_line)]
    ds --> features[tlp_features_daily]
    clock[Alarm every 5 min] --> fresh[(ops.dataset_freshness)]
    pub --> fresh

The sensor is in the graph because the file is external. The feature DAG is outside the cron. It starts from the dataset update. The alarm DAG starts from the clock and reads the freshness row the publish task wrote.

mermaidCopy
flowchart LR
    subgraph slots [Deployment pools]
        scd[dim_learner_scd slots 1]
        wh[warehouse_write slots 4]
        defaultPool[default_pool]
    end
    backfill[Backfill run] --> scd
    daily[Today run] --> scd
    backfill --> wh
    daily --> wh
    sensors[Sensors in reschedule] --> defaultPool

Two runs can both want the learner dimension. The pool of one serializes them. Sensors stay out of warehouse_write.

Technical Explanation

A producer outlet and a consumer

Add the outlet to the existing publish task. Do not create a second publish that only exists to fire the dataset, or you will eventually update one and not the other.

Consumer tasks should be idempotent on the business date they read from ops.dataset_freshness, because a manual dataset update or a cleared publish task will run them again.

pythonCopy
@task
def build_features() -> dict:
    from airflow.operators.python import get_current_context
    from tlp.features import build_order_features

    ctx = get_current_context()
    # Dataset-triggered run: trust the freshness table written by the producer,
    # not date.today().
    return build_order_features(triggered_by=ctx["run_id"])

build_order_features selects the latest freshness row with recon_diff_cents = 0 and builds features for that business_date with a merge. A second trigger the same day updates in place.

Backfill with the cap in place

Before the command:

1. Deploy the corrected parse code. Backfilling old code recomputes the bug.
2. Confirm max_active_runs is 1 or 2, not the number of days.
3. Confirm load_dim_learner uses dim_learner_scd.
4. Confirm the fact merge refuses frozen dates.
5. Tell the warehouse owner which slots you will take.

Then run the CLI for the open dates only. Watch the grid. If reconcile fails on one date, that DagRun stops. Other dates continue up to max_active_runs. Fix the data, then clear the failed task. Do not mark it success.

--reset-dagruns on a range that includes a closed month plus a weak SQL predicate is how September moves. The CLI is not the control. The predicate is the control. The CLI date list is the other control. Use both.

Mapped cleanse with a cap

pythonCopy
@task(max_active_tis_per_dag=4, pool="warehouse_write", pool_slots=1)
def cleanse_one(path: str) -> dict:
    from tlp.cleanse import cleanse_reseller_file
    return cleanse_reseller_file(path, **_window())

Four files cleanse at once. A fifth waits. Each holds one warehouse slot for the staging insert. The fact load, which wants two slots, waits until enough cleanses finish if the pool is nearly full. That queue is visible. An uncapped map is a surprise full pool and a missed SLO with every task "running."

Callbacks that only notify

pythonCopy
import logging
from airflow.exceptions import AirflowFailException

log = logging.getLogger(__name__)

def notify_final_failure(context: dict) -> None:
    ti = context["ti"]
    log.error(
        "final_failure dag=%s task=%s run=%s try=%s",
        ti.dag_id,
        ti.task_id,
        context["run_id"],
        ti.try_number,
    )

@task(retries=0, on_failure_callback=notify_final_failure)
def reconcile_guard() -> None:
    raise AirflowFailException("recon_diff_cents nonzero")

The SQL operator can take the same on_failure_callback. Keep the function free of credentials pulled from the wrong place. If you send mail, do it from infrastructure that is allowed to fail independently. A broken mail server must not fail the callback in a way that hides the original task state. Log first.

Health endpoints

The web server exposes a health check that includes the scheduler heartbeat and, when configured, the triggerer. A process supervisor should hit that endpoint. A green web page that loads from cache is not the check. The metadata database being reachable is part of health, because a scheduler that cannot write task state will not keep a fresh heartbeat for long.

Manual dataset events

The UI and CLI can create a dataset event. That is a production trigger for every consumer of that URI. Use it to rerun features after a repaired publish only when the freshness row is already correct. Firing the event without writing freshness makes consumers read a stale or empty row and possibly publish an empty feature table with a new run id. The outlet on the real publish task is the path that keeps those writes together.

Practical Implementation or Design Walkthrough

Problem. Make tlp_orders_daily operable for late files, an eight-day backfill, and a feature DAG that must not start on a failed reconcile.

Steps.

1. Set the retry block from this lesson in default_args. Set retries=0 on reconcile and publish. Raise AirflowFailException from any Python guard around the diff.
2. Add SqlSensor in mode="reschedule" ahead of cleanse, with an 80-minute timeout, templated on data_interval_start.
3. Create pools dim_learner_scd (1) and warehouse_write (4). Assign slots as in the diagram.
4. Add outlets=[FACT_ORDER_LINE] on publish_freshness only.
5. Add tlp_features_daily with schedule=[FACT_ORDER_LINE] and an idempotent feature merge.
6. Add tlp_freshness_alarm on */5 * * * * that fails if the expected business date is missing at or after 06:30 UTC.
7. For the backfill, deploy the code, then:

bashCopy
airflow dags backfill tlp_orders_daily \
  --start-date 2026-03-01 \
  --end-date 2026-03-08 \
  -y

8. Watch pool slots and the alarm. Do not include a frozen date.

Expected behavior. A missing file keeps the sensor in up_for_reschedule without a dedicated worker for the whole wait, then fails. A green publish creates one dataset event and one feature run. A red reconcile creates neither. Two backfill days do not run load_dim_learner at the same time. Clearing load_fact_order_line on one day reruns the merge for that interval only.

Spark later. Replace the body of cleanse_one with a Spark submit that writes the same staging table. Keep max_active_tis_per_dag, the pool, the outlet, and the alarm. The cluster size changes. The 06:30 page does not.

Enterprise Real-World Scenarios

Late file, occupied workers, missed flash report

The reseller file landed at 06:22. The wait was a FileSensor in poke mode with a 30-second interval and a four-hour timeout, running on the warehouse queue. It held a Celery slot the fact task also needed. Two experimental mapped DAGs had filled the rest of default_pool. At 07:00 the fact task was still queued, the sensor was "running," and finance delayed the flash report. The data had been sitting in the bucket since 06:22. The fix was mode="reschedule", a timeout inside the SLO, a dedicated small pool for waits, and max_active_tis_per_dag on the experimental maps. The rerun then finished in the normal twenty minutes because the merge was idempotent and the file parse was deterministic. The incident was capacity accounting, not SQL.

Feature consumers on a failed reconcile

A feature DAG was scheduled at 07:30 UTC. Reconcile failed at 06:50 because of a conflicting price quarantine over the threshold. Someone reran features anyway so a dashboard would update. The dashboard updated from the previous good partition mixed with a partial load from a cleared fact task that had not reached publish. Moving the feature DAG to schedule=[FACT_ORDER_LINE] made the successful publish the only trigger. The partial load no longer had an outlet. The dashboard stayed on the last good business date, which ops.dataset_freshness still named. Product complained about staleness. Staleness was the honest state. The previous design had freshness theater.

Backfill beside the daily run

An eight-day cents fix deployed at 05:30 UTC, and the daily DAG also became eligible at 06:00. max_active_runs=2 allowed the daily run and one backfill day. dim_learner_scd forced the dimension applies to take turns. warehouse_write kept four slots, with the fact task taking two, so analysts kept capacity. One of the eight days had a quarantine over the threshold and failed reconcile. The other days completed. The on-call cleared nothing until the partner file for that date was replaced, then cleared cleanse_orders downstream through publish. September was not in the --start-date / --end-date window, and the merge predicate would have ignored it anyway. The freshness alarm stayed quiet for the current business date because the daily run finished by 06:20. The backfill did not borrow the alarm's meaning. Each DagRun had its own interval.

Best Practices

• Inherit a small retry count, then set retries=0 or raise AirflowFailException on business-rule failures.
• Keep callbacks side-effect free aside from a log or a page.
• Clear task instances to repeat work for an existing interval. Backfill to create intervals. Do not loop dates in a shell.
• Leave depends_on_past off unless you have tested a backfill stall. Serialize SCD with a pool.
• Update datasets only from the task that actually published, and only on success. Airflow already withholds the update when the task fails, so do not also update the dataset from a failure callback.
• Treat dataset URIs as exact identifiers. Share one constant between producer and consumer.
• On dataset-triggered DAGs, read the business date from the producer's freshness row.
• Use DatasetOrTimeSchedule only when both the clock and the dataset are genuine reasons to run.
• Put long waits in mode="reschedule" or defer them, and run a triggerer if you defer.
• Give mapped tasks max_active_tis_per_dag and a pool.
• Keep secrets in Connections or a backend. Assume rendered templates are logged.
• Alarm on freshness with a dedicated DAG. Do not treat the built-in sla timedelta as that alarm until you have tested your version.
• Page on heartbeat, import errors, and final failures. Do not page on every retry.
• When the task becomes Spark, keep pools, outlets, timeouts, and the alarm.

Common Mistakes and Anti-Patterns

Retrying reconcile until the grid is green. The mismatch is real. retries=0 and AirflowFailException.

depends_on_past=True on every task. One bad day blocks the backfill and the future. Depend on edges inside the run, and on pools across runs.

Backfill from a laptop with date.today(). Intervals are whatever the laptop thinks today is. Use airflow dags backfill so each DagRun has an interval.

A dataset URI built with string formatting and a date. You create a new dataset every day and the consumer waiting on the un-dated URI never fires. One stable URI per logical table. Put the date in the data, not in the dataset identity, unless you truly want one dataset per partition and a consumer schedule that changes at parse time, which you do not.

Poke-mode sensors on the warehouse queue. They occupy slots and steal them from the merge. Reschedule, and keep waits off the write pool.

Deferrable operators without a triggerer. Tasks sit in deferred. Run the process or do not defer.

Uncapped .expand() over every object in a bucket. The pool saturates, the metadata database grows a task instance per object, and the scheduler spends the morning deciding. Cap it, or compact the files first.

Firing a manual dataset event to "kick" consumers after a failed load. Consumers run on a lie. Repair the data, let publish_freshness succeed, and let the outlet fire.

Using sla= as the finance deadline. The semantics are easy to get wrong and weak for dataset-triggered DAGs. Write the freshness alarm explicitly.

Logging the connection URI. The password hits the log store. Log the conn_id only.

Two pools that secretly limit the same database. Each team thinks it has four slots and the database sees eight. Name one pool for that database and share it.

Troubleshooting and Design Review

Design review for an operational change:

1. What is retryable, and what raises AirflowFailException?
2. Is the backfill command's date range free of frozen periods, and does SQL agree?
3. Which task is the only outlet for each dataset URI?
4. Which pool does the task use, and how many slots remain for analysts and for the daily run?
5. Do waits free the worker?
6. What pages a human at 06:30, and which table does that check read?
7. Can a second run of each task leave the same rows?

Interview Questions and Answers

Q1. How is a retry different from a clear and from a backfill?

A retry is another attempt of the same task instance after a failure, with the same data interval, up to retries. A clear resets an existing task instance so it can run again, often after a code fix, and can include downstream tasks. A backfill creates or resets DagRuns across a date range you pass to airflow dags backfill. Retries happen automatically. Clears and backfills are operations you choose. All three re-enter the task, so the task must be idempotent.

Q2. Why raise AirflowFailException on a reconciliation mismatch?

A normal exception fails the attempt and schedules another attempt if retries remain. A cents mismatch will not heal on the second try. AirflowFailException fails the task immediately, which is what you want when default_args still carries retries for the rest of the DAG. Pair it with retries=0 on that task so the intent is visible in the definition, not only in the exception type.

Q3. What does an Airflow Dataset update mean?

It means a task with that dataset in outlets succeeded. Airflow records an event for the URI. It does not inspect the table. Consumer DAGs scheduled on that dataset become eligible once their dataset conditions are met. Failed and skipped tasks do not emit the update. The URI is a literal identifier: no globs, and case-sensitive. TechLearningPro uses one URI for analytics.fact_order_line, updated only by publish_freshness.

Q4. Why can a dataset-triggered DAG not assume data_interval_start is yesterday?

Its schedule is an event, not the daily timetable of the producer. The run's interval fields do not automatically copy the producer's business date. Read ops.dataset_freshness for the business date that just published, and make the consumer merge idempotent on that date. Using date.today() in the consumer reintroduces the wall-clock bug for late publishes.

Q5. What problem does DatasetOrTimeSchedule solve, and when is it the wrong tool?

On Airflow 2.9 it runs a DAG when any of the given datasets update or when a cron timetable fires. That fits a janitor that should run nightly and also whenever a table publishes. It is the wrong tool for a feature job that must run only after a good publish. Adding a cron "just in case" starts the job when the dataset never updated, which is the stale-dashboard failure mode.

Q6. How do pools prevent a backfill from colliding with the daily SCD load?

load_dim_learner requests the only slot in dim_learner_scd. The daily run and a backfill run both queue that task. One executes. The other stays queued until the slot frees. The fact pool separately caps heavy merges. Pools are deployment-wide, so the cap survives someone adding a second DAG that also loads the dimension, as long as that DAG uses the same pool. A pool that exists only in a diagram does nothing.

Q7. Why is mode="poke" a poor default for a file that may arrive an hour late?

Poke holds a worker slot between checks. An hour-long wait occupies a slot the merge may need, and it counts against the pool if you set one. mode="reschedule" frees the worker and asks the scheduler to check again after poke_interval. Deferrable mode frees the worker and uses the triggerer. Poke is acceptable for a few seconds. It is not acceptable as the wait for the reseller file.

Q8. What does .expand() change about concurrency?

It creates one task instance per item returned by an upstream task at runtime, which is how you cleanse a variable number of files without querying the bucket at import. Each instance is a real task instance that can run in parallel. Without max_active_tis_per_dag and a pool, a large list fills the default pool and delays other DAGs. The expanded values travel through XCom, so expand on paths and counts, not on file contents.

Q9. What is wrong with Airflow's sla timedelta as a finance deadline?

It is a scheduler feature tied to timetable semantics, historically easy to misread, weak for manual and dataset-triggered runs, and not a clear "page at 06:30 while the task is still queued" control. The finance promise is "this business date is published by 07:00 UTC with a zero diff," which is a row in ops.dataset_freshness. A small DAG on a five-minute schedule can read that row and fail. That behavior you can explain to finance. Test any use of sla= on your patch version before you depend on it, and do not let it replace the freshness row.

Q10. Which signals tell you Airflow has stopped scheduling?

A stale scheduler heartbeat, import errors for the DAG file, a full pool with tasks stuck queued, a dead triggerer if you depend on deferred tasks, and a metadata database the scheduler cannot write to. The grid looking unchanged is consistent with all of those. airflow dags list-import-errors and the health endpoint are the first checks. A green web server only means the web process answered.

Q11. Where should the warehouse password live, and what will leak it?

In a Connection or a secrets backend configured for the deployment, accessed by conn_id inside the hook. Templating the password into a command, printing the connection URI, or storing it in a Variable that a DAG renders will copy it into logs and the metadata database's rendered fields. Workers and the scheduler need the same secret access. The DAG repository should contain only the connection id.

Q12. Which operational controls stay when cleanse_orders submits Spark?

The data interval, retries for retryable submit failures, the pool and max_active_tis_per_dag, the sensor or dataset that gates the start, the outlet on publish, the freshness alarm, and the backfill command. Spark adds cluster-level concurrency. Enforce that either as an Airflow pool on the submit task or as a queue on the cluster, and do not assume both did it. The idempotent staging write remains the property that makes clear and backfill safe.

Hands-On Exercises

Exercise 1. Sensor, pool, and a failed reconcile

Problem. Extend tlp_orders_daily on local Airflow 2.9.

Tasks.

1. Add a reschedule sensor with a short timeout so you can see failure without waiting an hour.
2. Create a pool of one slot and point load_dim_learner at it. Start two DagRuns and show one dimension task queued.
3. Force a reconciliation exception with AirflowFailException and show that the task does not retry.
4. Confirm no dataset event exists for the failed run.

Acceptance. The sensor does not sit in running for the whole wait. The second SCD task waits on the pool. The failed reconcile has one attempt.

Exercise 2. Dataset consumer and a two-day backfill

Problem. Publish a dataset and backfill two business dates.

Tasks.

1. Add FACT_ORDER_LINE as an outlet on publish and as the schedule of a consumer that writes a row saying which business date it read from ops.dataset_freshness.
2. Run one successful producer day and show the consumer DagRun.
3. Backfill two dates with airflow dags backfill and max_active_runs at most 2.
4. Show that clearing publish on one date retriggers the consumer and that the feature write is still one row per business date.

Rubric.

Lesson Summary

Operating tlp_orders_daily means deciding which failures retry, which intervals to recreate, which URI signals a real publish, and which slots the warehouse can spare. Airflow 2.9 will backfill a range, schedule consumers on datasets, reschedule sensors, map tasks at runtime, and export a heartbeat. It will also retry a bad reconcile, poke until the pool is full, and treat a manual dataset click as a publish if you ask it to.

The freshness promise stays a row your alarm DAG can read. The built-in SLA timedelta does not replace that row. The idempotent merge stays the reason a clear or a backfill is allowed to touch fact_order_line at all. The next lesson records what these tasks read and write, so a publish outlet is backed by lineage and a catalog, not only by a URI string inside Airflow.

Further Reading

• Apache Airflow 2.9, Data-aware scheduling — dataset URIs, outlets, AND schedules, and the 2.9 logical expressions.
• Apache Airflow 2.9 release notes on DatasetOrTimeSchedule — when a cron and a dataset share one DAG.
• Apache Airflow, Pools and the airflow pools set CLI.
• Apache Airflow, Dynamic task mapping — .expand() from 2.3 onward.
• Apache Airflow, Sensors — poke, reschedule, and deferrable mode.
• airflow dags backfill --help on the release you run — confirm flags before a production range.
• TechLearningPro, Apache Airflow Architecture and DAGs — the DAG these controls attach to.
• Next: Data Lineage and Metadata Management.