Asset Conditions and Aliases

Concepts covered: Cross-DAG dependencies

A daily revenue table needs orders and refunds together, and a fraud scorer needs either a new order batch or a new rule set. Asset schedules combine for both: & builds a schedule that waits for every asset in it, | builds one that fires on any of them, and parentheses nest the 2. Airflow keeps, per consumer DAG and per asset, a flag saying whether an event has arrived since that consumer last ran, and tests the expression against those flags on each scheduler loop.

from datetime import datetime
from airflow.sdk import Asset, dag

orders = Asset("orders_raw")
refunds = Asset("refunds_raw")
rules = Asset("fraud_rules")
START = datetime(2026, 11, 10)


@dag(schedule=(orders & refunds), start_date=START)
def daily_revenue():
    ...


@dag(schedule=(orders | rules), start_date=START)
def fraud_scores():
    ...


@dag(schedule=((orders & refunds) | rules), start_date=START)
def nightly_audit():
    ...


daily_revenue()
fraud_scores()
nightly_audit()

An AND schedule waits for 1 event per asset

daily_revenue needs 1 event from each of its 2 assets since its last run. Counts beyond 1 change nothing: 2 orders events and 0 refunds events leave it waiting. When the refunds event arrives, the run is created with all 3 events in triggering_asset_events, and both flags reset to waiting. fraud_scores fires on every single event of either asset, so 2 orders events 15 minutes apart give it 2 runs.

nightly_audit is satisfied by the pair or by a rules event alone. Over the same night it runs at 02:40 on the pair, at 03:10 on the rule set by itself, and at 04:05 on the pair again. Each run consumes every event queued for that DAG, whichever side of the expression matched. Had a rules event arrived at 03:55, after the 03:50 refunds event, it would have created a run carrying both and cleared the refunds flag, and the 04:05 orders event would then have left the pair waiting.

Time and data together

A report due at 06:00 whether the feed arrived mixes the 2 kinds of trigger. AssetOrTimeSchedule takes a timetable and an asset expression and creates a run when either side is satisfied, and the clock side fires every day whether or not the assets landed. With CronTriggerTimetable as the clock side, the 06:00 tick creates a scheduled run whose logical date is 06:00 and whose data interval starts and ends at 06:00, a point with no span to read. The asset side creates an asset-triggered run with no logical date at all. So the task branches: when triggering_asset_events holds events it reads them, and otherwise it builds the 24 hours ending at the logical date itself.

from datetime import datetime, timedelta
from airflow.sdk import Asset, dag, task
from airflow.timetables.assets import AssetOrTimeSchedule
from airflow.timetables.trigger import CronTriggerTimetable


@dag(
    schedule=AssetOrTimeSchedule(
        timetable=CronTriggerTimetable("0 6 * * *", timezone="UTC"),
        assets=(Asset("orders_raw") & Asset("refunds_raw")),
    ),
    start_date=datetime(2026, 11, 10),
)
def revenue_by_six():
    @task
    def build(triggering_asset_events, logical_date):
        if triggering_asset_events:
            return [e.source_run_id for evs in triggering_asset_events.values() for e in evs]
        return (logical_date - timedelta(days=1), logical_date)

    build()


revenue_by_six()

Aliases for names known only at run time

A producer that writes 1 directory per day, file:///lake/orders/2026-11-10/, cannot declare that path as an outlet: the path differs in every run, and the DAG file is parsed before any run exists. An AssetAlias is a fixed name a task can attach any asset to while it runs. The task declares the alias in outlets, and inside its body adds the concrete asset to outlet_events[alias]. When the task succeeds, Airflow records an event on that concrete asset, records that the alias produced it, and triggers 2 kinds of consumer: DAGs scheduled on the alias, which fire for any asset added through it, and DAGs scheduled on that exact path.

from datetime import datetime
from airflow.sdk import Asset, AssetAlias, dag, task

daily_orders = AssetAlias("daily_orders")
START = datetime(2026, 11, 10)


@dag(schedule="@daily", start_date=START)
def orders_partition():
    @task(outlets=[daily_orders])
    def write(outlet_events, data_interval_start):
        day = data_interval_start.strftime("%Y-%m-%d")
        outlet_events[daily_orders].add(Asset(f"file:///lake/orders/{day}/"))

    write()


@dag(schedule=daily_orders, start_date=START)
def any_partition_report():
    ...


@dag(schedule=Asset("file:///lake/orders/2026-11-10/"), start_date=START)
def one_partition_report():
    ...


orders_partition()
any_partition_report()
one_partition_report()

After the run for Nov 10 succeeds, the Assets page shows file:///lake/orders/2026-11-10/ as a real asset with daily_orders as its producer. any_partition_report runs for it as it runs for every day, and one_partition_report runs for this day only. Airflow records the alias-to-path resolution at the event, and the dependency lines in the UI catch up at the next parse of the file. A task that adds 4 assets through one alias writes 4 events, 1 per concrete asset, all at its success, so a consumer scheduled on the alias gets 1 run whose triggering_asset_events lists all 4, each event naming its own concrete path.

Schedule on the alias

• Fires for every asset the alias adds
• The name is fixed in the DAG file
• Fits a consumer that handles whatever partition landed

Schedule on the exact asset

• Fires for 1 concrete path only
• The path is known only once the producer has added it
• Fits 1 consumer per partition or per region

About this lesson

This section is part of the Scheduling: Advanced lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.