Airflow Datasets and Assets: Replacing Cron Guesswork With Data-Aware Scheduling

The marketing team’s attribution DAG ran at 06:30. Not because anything happened at 06:30, but because the orders pipeline usually finished around 06:05 and someone added twenty-five minutes of padding in 2023.

Then orders grew. By the time I looked, the orders DAG was finishing at 06:20 on a good day and 06:50 on a Monday, and attribution was producing yesterday’s numbers roughly one Monday in three. The fix everyone proposed was moving attribution to 07:00. That is not a fix, that is a larger guess.

Cron offsets between DAGs are a distributed system held together by somebody’s estimate of somebody else’s runtime, and the estimate is never revisited. Data-aware scheduling replaces the estimate with the actual event: the orders DAG finishes, and attribution starts.

Here is how I use it in production, including the parts that took me two incidents to learn.

The shape of the thing

A producer task declares an outlets list. When that task succeeds, Airflow records an update event against those datasets. A consumer DAG sets schedule=[dataset] and gets triggered when the update lands.

In Airflow 3 the class is Asset; in Airflow 2 it was Dataset. Same mechanism, better name, plus features the older one did not have. I use the Airflow 3 spelling throughout — the Airflow 3 upgrade notes cover the rename if you are still on 2.

# dags/orders_pipeline.py  (producer, owned by the data platform team)
import pendulum
from airflow.sdk import Asset, dag, task

ORDERS_CURATED = Asset("s3://lake/curated/orders/")

@dag(
    schedule="0 5 * * *",
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    max_active_runs=1,
    tags=["platform", "producer"],
)
def orders_pipeline():

    @task
    def extract() -> int:
        ...
        return 41_882

    @task(outlets=[ORDERS_CURATED])
    def publish(row_count: int) -> None:
        # Only this task carries the outlet. Extract failing means
        # no event, which is exactly what downstream needs to know.
        if row_count == 0:
            raise ValueError("Refusing to publish an empty orders partition")
        ...

    publish(extract())

orders_pipeline()

# dags/attribution.py  (consumer, owned by marketing analytics)
import pendulum
from airflow.sdk import Asset, dag, task

ORDERS_CURATED = Asset("s3://lake/curated/orders/")

@dag(
    schedule=[ORDERS_CURATED],   # no cron, no padding
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    tags=["marketing", "consumer"],
)
def attribution():

    @task
    def build_attribution() -> None:
        ...

    build_attribution()

attribution()

The consumer no longer knows or cares when the producer runs. If the platform team moves the orders DAG to 04:00, attribution moves with it. If orders takes an extra forty minutes, attribution waits.

flowchart TD
  cron["Cron 05:00"] --> orders["orders_pipeline"]
  orders --> pub["publish task, outlet"]
  pub --> evt["Asset update event"]
  evt --> attr["attribution DAG run"]
  evt --> fin["finance_daily DAG run"]
  evt --> ml["feature_refresh DAG run"]

One event, three consumers, zero coordination meetings.

Put the outlet on the right task

This is the mistake I have made and seen most often, and it produces the exact failure datasets were supposed to prevent.

The outlet fires when the task succeeds. Not when it writes data. A task that runs, finds nothing to do, and returns cleanly emits the update anyway, and every downstream consumer wakes up and processes the same data it processed yesterday.

Two rules keep this honest.

The outlet goes on the last task that writes, not on a final no-op. If your DAG ends with a cleanup or notify task, the outlet does not belong there. It belongs on the task whose success means the data is complete and readable.

A no-op run should fail or skip, not succeed. If zero rows is genuinely a valid outcome for a day, the task should raise AirflowSkipException rather than succeed, because a skipped task does not emit the update. If zero rows is never valid, raise.

from airflow.exceptions import AirflowSkipException
from airflow.sdk import Asset, task

ORDERS_CURATED = Asset("s3://lake/curated/orders/")

@task(outlets=[ORDERS_CURATED])
def publish_partition(partition_date: str) -> None:
    rows = write_partition(partition_date)

    if rows == 0:
        # Skipped tasks do not emit asset events. Consumers stay put.
        raise AirflowSkipException(f"No orders for {partition_date}; nothing published")

    # Attach context so consumers can make decisions, not just wake up.
    from airflow.sdk import Metadata
    yield Metadata(ORDERS_CURATED, {"rows": rows, "partition": partition_date})

The metadata payload is the difference between “something changed” and “here is what changed.” A consumer that receives the partition date can process exactly that partition instead of re-deriving it from a schedule it no longer has.

Conditional scheduling is what makes this work past two teams

A single upstream dependency is easy. Real marts wait on three or four things, and some of them are alternatives rather than requirements. Airflow supports boolean expressions over assets.

import pendulum
from airflow.sdk import Asset, dag

ORDERS = Asset("s3://lake/curated/orders/")
CUSTOMERS = Asset("s3://lake/curated/customers/")
PRODUCTS = Asset("s3://lake/curated/products/")
FX_RATES = Asset("s3://lake/reference/fx_rates/")
FX_MANUAL = Asset("s3://lake/reference/fx_rates_manual_override/")

@dag(
    # Needs all three curated sources, plus rates from either source.
    schedule=(ORDERS & CUSTOMERS & PRODUCTS) & (FX_RATES | FX_MANUAL),
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
)
def revenue_marts():
    ...

revenue_marts()

That expression replaces what would otherwise be four ExternalTaskSensor instances, each holding assumptions about the upstream DAG’s schedule and task IDs. ExternalTaskSensor is the pattern datasets exist to kill: it couples you to another team’s internal task naming, it needs matching logical dates, and it holds a worker slot while it waits unless you deferred it, per deferrable operators.

You can also mix a schedule with assets, which handles the very common “run when upstream lands, but at minimum once a day” requirement:

from airflow.timetables.assets import AssetOrTimeSchedule
from airflow.timetables.trigger import CronTriggerTimetable

@dag(
    schedule=AssetOrTimeSchedule(
        timetable=CronTriggerTimetable("0 8 * * *", timezone="UTC"),
        assets=(ORDERS & CUSTOMERS),
    ),
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
)
def daily_summary():
    ...

That is a floor, not a guess. It runs when the data is ready, and if the data never arrives, it runs anyway at 08:00 so the failure is visible in the right DAG instead of as an absence.

Naming: the URI is a public interface

The moment another team schedules on your asset URI, it is an API. I have watched a rename land at 2 p.m. and three downstream DAGs simply stop triggering — no error, no failed task, just DAGs that never run again. The consumers are watching a URI that nothing emits anymore.

The convention I enforce:

<system>://<domain>/<layer>/<entity>[/<grain>]

s3://lake/curated/orders/
snowflake://analytics/marts/fct_orders
postgres://crm/public/customers

Rules that came out of getting this wrong:

• Point at the physical thing. The URI should name the table or prefix a consumer would actually read. Abstract names like orders_ready drift from reality.
• Define assets in one shared module, imported by both producers and consumers. A typo in a string literal on the consumer side creates a brand new asset that nothing produces, and the DAG waits forever with no error.
• Never rename. Add the new asset, emit both for a deprecation window, migrate consumers, then remove the old outlet.

# common/assets.py — imported by every DAG in the repo
from airflow.sdk import Asset

ORDERS_CURATED = Asset("s3://lake/curated/orders/")
CUSTOMERS_CURATED = Asset("s3://lake/curated/customers/")
FCT_ORDERS = Asset("snowflake://analytics/marts/fct_orders")

That single file is the cross-team contract, and it belongs under CODEOWNERS review. It is the orchestration-layer version of the producer obligations described in data contracts for pipeline teams.

What datasets do not solve

Backfills. This is the largest gap and the one that surprises people. Running a producer for thirty historical dates does not meaningfully replay thirty downstream runs — asset events are about “this thing changed now,” not “this interval was reprocessed.” Plan backfills as an explicit multi-DAG operation, in dependency order, triggered deliberately. I keep a documented backfill runbook per subject area for exactly this reason.

Partial freshness. An asset update says the whole thing changed. If your producer wrote one partition of a hundred, the event does not convey that unless you attach metadata and your consumer reads it. Most consumers do not, so they reprocess everything.

Cross-instance dependencies. Assets are scoped to one Airflow deployment. If the producer is in another team’s Airflow, you are back to an API call, a sensor on the physical object, or a shared message. Watchers can bridge some of this by triggering assets from external events, but it is a bridge, not a native link.

Ordering guarantees among consumers. Three DAGs triggered by the same event start concurrently. If one must precede another, that is a dependency between them, expressed as its own asset.

Where teams get this wrong

Outlet on a notification task. The Slack message succeeds whether or not the load did. Downstream now triggers on the reliable delivery of a message rather than on data.

A task that succeeds on empty input. The most common cause of consumers running on stale data. Skip or fail on empty; never succeed quietly.

Duplicating the URI string in every DAG file. One typo produces a silent orphan. Import from a shared module.

Renaming a URI. Consumers stop firing with no error anywhere. There is no “asset not found” warning because an asset with no producer is a perfectly legal thing to wait on.

Keeping the cron as a safety net without thinking. A consumer with both a cron schedule and asset triggers will run on both. Sometimes that is what you want (AssetOrTimeSchedule); often it means the DAG runs twice and you have an idempotency problem you did not have before.

Assuming the event carries the interval. Asset-triggered runs may have no logical_date. Any path built from it needs a fallback, which is the single most common Airflow 3 breakage.

FAQ

Do datasets replace ExternalTaskSensor entirely?

Within one Airflow instance, yes, and you should retire the sensors. Across instances, no — you still need a sensor on the physical object or an API-triggered run. The advantage holds anyway: you stop coupling to another team’s task IDs.

What happens if the producer runs twice in an hour?

Each success emits an event and each event triggers the consumer, subject to max_active_runs. If the consumer is expensive, set max_active_runs=1 and accept that a rapid double-publish collapses into one downstream run plus one queued.

Can I trigger on an asset produced by a task that is not in Airflow?

Yes, through the REST API — you can post an asset event directly. That is how I wire in external systems: the Spark job on EMR finishes, calls the endpoint, and the Airflow-side consumers fire. It is also how you bridge a second Airflow instance.

How do I see which DAGs consume an asset?

The Assets view in the UI shows producers and consumers per asset, which is the first place I look when someone asks whether they can change a table. The graph is also in the REST API if you want to generate a dependency report.

Does asset scheduling work with dynamic task mapping?

Yes, and it composes well: a mapped set of tasks can all declare the same outlet, and the event fires when they succeed. Be careful with partial failures — see dynamic task mapping for how mapped task states aggregate.

Should every table be an asset?

No. Make an asset out of anything another team schedules on. Internal intermediate tables within one DAG do not need one, and a hundred assets nobody consumes makes the Assets view useless for the ten that matter.

What this means for your pipelines

Every cron offset between two DAGs encodes a belief about how long something else takes. Those beliefs are correct on the day they are written and decay from there, silently, until a Monday when the padding is not enough and a report goes out with yesterday’s numbers. Nobody gets paged, because nothing failed.

Data-aware scheduling turns that belief into an event. The producer says “this is ready,” the consumers respond, and the coupling is a URI instead of a clock. The cost is that you now maintain a namespace of URIs with the seriousness of an API, guard your outlets so a no-op run does not lie, and handle backfills as a deliberate operation rather than a side effect.

Start with your worst offender. Find the pair of DAGs where someone has already moved the downstream cron later at least twice — every team has one, and everyone knows which it is. Convert that single edge to an asset, watch it for two weeks, then move the next. Within a quarter the padding disappears from your schedules and the question “did upstream finish?” stops being something anyone has to check.