Apache Airflow / Intermediate: runtime work, control flow and resource budgets / Lesson 10

Assets, events and data-aware scheduling

Schedule from declared data updates while preserving separate evidence for availability, freshness and interval coverage.

About 5 min readContent reviewed 2026-10-28

On this page

Lesson goals

Explain asset identity and successful producer update events.
Review multi-asset conditions and event-triggered context.

An asset names a data dependency

An Airflow asset is a logical identity used to describe data produced or consumed by workflows. A producer task can declare an outlet and emit an update event when it succeeds under the asset contract. The event is orchestration evidence that the producer reported an update, not an automatic checksum or proof of business completeness. Use a stable identity that all producers and consumers agree on. Renaming a URI can disconnect the scheduling relationship even when the underlying file remains in the same storage system. Do not include passwords or sensitive information in asset URIs or metadata because those identifiers can be visible in the interface.

Declare matching producer and consumer identities

The fragment creates one synthetic Asset and uses it as an outlet and a downstream DAG schedule. It is illustrative framework code and does not write an object or execute a consumer. The URI is a logical identifier, not an instruction for Airflow to fetch its contents automatically. The producing task would need to publish a complete artifact before reporting success, and the consumer would need its own read and verification logic. Asset metadata should carry a concise version or partition reference where appropriate rather than a large dataset. Review the exact Airflow 3.1 asset API and provider behavior before independent use.

python Illustrative asset identity and consumer schedule

from airflow.sdk import Asset, DAG, task

parcel_manifest = Asset('https://assets.training.invalid/parcel-manifest')

@task(outlets=[parcel_manifest])
def publish_manifest_reference():
    return {'partition': '2026-10-01', 'version': 1}

with DAG(dag_id='lab_asset_consumer', schedule=[parcel_manifest], catchup=False):
    @task
    def acknowledge_update():
        return 'validate-event-partition-before-reading'
    acknowledge_update()

This fragment declares relationships; no artifact is published and the producer task is not placed in a complete producer DAG.

Multiple updates do not imply matching partitions

A consumer scheduled on several assets can require their update conditions to be satisfied under documented AND or OR expressions. That scheduling condition does not prove the updates describe the same business date, schema version or completeness boundary. Orders for October first and rates for October second can both be recently updated while still forming an invalid join. Carry partition identity and version in the data contract and validate them before calculation. Repeated updates can also be coalesced by scheduling semantics rather than guaranteeing one independent run for every event. Design the consumer around documented event behavior instead of assuming a general-purpose message queue.

Event-triggered runs need explicit business inputs

Asset-triggered and manually triggered runs without a logical date can lack data-interval keys in Airflow 3. Code that blindly indexes data_interval_start can therefore fail even when the scheduling event is valid. Obtain the relevant triggering-event information through public context APIs and validate the business partition it identifies. If a workflow supports both interval schedules and asset triggers, define a clear input-normalization step for each supported path. Do not silently substitute the current date when context is absent; that creates plausible but wrong output. Missing required business context should be an explicit validation failure with enough diagnostic identity to investigate.

Validate business compatibility after readiness

The offline fixture compares the partition values carried by two synthetic asset updates. Both say they are available, but their dates differ, so the business join is rejected. It models a validation rule only and does not implement asset event scheduling. A complete acceptance test should also cover matching dates, repeated version updates and a missing producer event. Decide whether a later correction retriggers calculation and how it replaces the prior report safely. This lesson emits no event, creates no asset record and runs no DAG; only the supplied compatibility comparison is evaluated locally.

python Offline asset-partition compatibility check

updates = {'orders': {'partition': '2026-10-01'}, 'rates': {'partition': '2026-10-02'}}
compatible = updates['orders']['partition'] == updates['rates']['partition']
assert not compatible
print(compatible)

Expected output

False

Two updates can be ready while their business partitions are incompatible.

Common errors and fixes

Two updated assets are joined for different dates.

Why it happens: Scheduling readiness was mistaken for partition compatibility.
How to fix it: Validate partition and version before processing.
Prevent it next time: Include mismatched-asset fixtures in acceptance checks.

An asset-triggered task raises KeyError for data_interval_start.

Why it happens: The code assumed interval keys exist for every trigger type.
How to fix it: Normalize supported public context and event metadata explicitly.
Prevent it next time: Test manual, interval and asset paths separately.

A renamed asset no longer triggers its intended consumer.

Why it happens: Producer and consumer logical identities diverged.
How to fix it: Coordinate a reviewed identity migration across both definitions.
Prevent it next time: Treat asset identity as a versioned integration contract.

Check your understanding

1. Does an asset URI make Airflow automatically download the named object?

Show worked solution

No. It identifies a scheduling dependency. Producers still need to create valid data and consumers need explicit authorized read logic. The URI may be logical rather than directly fetchable, so keep the actual artifact reference and integrity contract clear instead of relying on the scheduler to interpret arbitrary storage semantics.

2. Should missing interval context fall back silently to today in an asset consumer?