Data-Aware Scheduling in Airflow with Datasets

The Problem Datasets Solve

Before Airflow 2.4, if DAG B needed to wait for DAG A to finish producing data, you had two options. Both had significant drawbacks.

Option 1: TriggerDagRunOperator. DAG A explicitly triggers DAG B when it finishes. This creates tight coupling — DAG A needs to know about DAG B, and if you add a DAG C that also consumes the same data, you have to modify DAG A to trigger C as well. The producer knows about all its consumers, which violates basic separation of concerns.

Option 2: Sensors. DAG B runs on a schedule and uses an ExternalTaskSensor or S3KeySensor to wait for DAG A’s output. The sensor polls repeatedly — every 30 seconds, every minute — checking whether the data exists. Meanwhile, it occupies a worker slot, doing nothing productive. With many sensors across many DAGs, you can easily have dozens of worker slots consumed by tasks that are just sleeping and checking.

Think of the sensor approach like calling a restaurant every 30 seconds to ask “Is my table ready?” versus getting a text notification when it is. Both get you to the table, but one wastes your time and the restaurant’s phone line.

Datasets flip the model. Instead of consumers polling for data, producers announce when data is ready, and Airflow automatically triggers the consumers. The scheduler itself manages the dependency, not a running task.

What Is a Dataset?

A dataset in Airflow is a logical identifier — a URI string — that represents a piece of data your pipeline produces or consumes. It is not a connection to a database or a file handle. It is a name, a label, that the scheduler uses to track data dependencies between DAGs.

from airflow.datasets import Dataset

# These are just URIs -- Airflow does not validate or access them
orders_dataset = Dataset("s3://data-lake/orders/daily/")
user_profiles  = Dataset("postgresql://warehouse/public.user_profiles")
ml_features    = Dataset("gs://ml-bucket/features/latest.parquet")

The URI can be anything meaningful to your team. Airflow does not parse it, connect to it, or verify that the resource exists. The URI is purely a coordination signal — a name that producers and consumers agree on.

This is an important design choice. Airflow datasets are not data catalogs, data lineage tools, or data quality frameworks. They solve one specific problem: triggering DAG runs based on data availability rather than time.

Building a Producer DAG

A producer DAG is a regular DAG that declares it produces (updates) one or more datasets. You mark a task’s outputs with the outlets parameter:

from datetime import datetime
from airflow.decorators import dag, task
from airflow.datasets import Dataset

# Define the dataset -- a URI that consumers will reference
orders_daily = Dataset("s3://data-lake/orders/daily/")

@dag(
    dag_id="produce_orders",
    start_date=datetime(2026, 1, 1),
    schedule="@daily",
    catchup=False,
    tags=["producer"],
)
def produce_orders():

    @task(outlets=[orders_daily])
    def extract_and_load(**context):
        """Extract orders and load to S3. Declares orders_daily as an outlet."""
        from airflow.providers.amazon.aws.hooks.s3 import S3Hook

        hook = S3Hook("aws_default")
        ds = context["ds"]

        # Your extraction logic here
        data = fetch_orders_from_source(ds)

        # Upload to S3
        hook.load_string(
            string_data=data.to_csv(),
            key=f"orders/daily/{ds}/orders.csv",
            bucket_name="data-lake",
        )
        # When this task succeeds, Airflow marks orders_daily as "updated"

    extract_and_load()

produce_orders()

The critical detail: when a task with outlets=[orders_daily] succeeds, Airflow’s scheduler records that the dataset has been updated. This update event is what triggers consumer DAGs. The task does not need to call any special API — success is the signal.

Building a Consumer DAG

A consumer DAG declares that it should run whenever one or more datasets are updated. Instead of a time-based schedule like "@daily", you pass dataset references:

from datetime import datetime
from airflow.decorators import dag, task
from airflow.datasets import Dataset

# Reference the same dataset the producer declares
orders_daily = Dataset("s3://data-lake/orders/daily/")

@dag(
    dag_id="consume_orders",
    start_date=datetime(2026, 1, 1),
    schedule=[orders_daily],  # Triggered by dataset update, not time
    catchup=False,
    tags=["consumer"],
)
def consume_orders():

    @task
    def build_report(**context):
        """Runs automatically whenever orders_daily is updated."""
        triggering_events = context["triggering_dataset_events"]
        print(f"Triggered by: {triggering_events}")

        # Your reporting logic here
        from airflow.providers.amazon.aws.hooks.s3 import S3Hook
        hook = S3Hook("aws_default")
        # Read the data that was just produced, build reports...

    build_report()

consume_orders()

Notice that the consumer DAG has no cron schedule. It does not run every hour or every day. It runs only when the scheduler detects that orders_daily has been updated — that is, when the producer DAG’s extract_and_load task has succeeded.

Multiple Dataset Dependencies

A consumer can wait for multiple datasets. The DAG triggers only when all listed datasets have been updated since the last consumer run:

orders_daily = Dataset("s3://data-lake/orders/daily/")
users_daily  = Dataset("s3://data-lake/users/daily/")

@dag(
    dag_id="build_user_order_report",
    schedule=[orders_daily, users_daily],  # Both must be updated
    ...
)
def build_user_order_report():
    @task
    def join_and_report():
        # Only runs when BOTH orders AND users data are fresh
        ...

This is a common pattern for analytics DAGs that join data from multiple sources. The report only runs when all upstream data is available, ensuring consistency.

How Datasets Replace Sensors

Let us compare the old sensor-based approach with datasets to understand the concrete improvement.

The Sensor Approach (Before Datasets)

# Old way: Consumer DAG uses a sensor to wait for producer
from airflow.sensors.external_task import ExternalTaskSensor

@dag(schedule="0 3 * * *")  # Runs at 3 AM, hoping producer is done by then
def old_consumer():

    wait_for_orders = ExternalTaskSensor(
        task_id="wait_for_orders",
        external_dag_id="produce_orders",
        external_task_id="extract_and_load",
        mode="reschedule",       # Release worker slot between checks
        poke_interval=300,       # Check every 5 minutes
        timeout=7200,            # Give up after 2 hours
    )

    @task
    def build_report():
        ...

    wait_for_orders >> build_report()

Problems with this approach:

• The consumer runs at 3 AM regardless of whether the producer has finished
• If the producer is late, the sensor waits (possibly for hours), wasting resources
• If the producer finishes early, the consumer still waits until 3 AM to start
• The poke_interval and timeout values are guesses — too aggressive wastes resources, too conservative adds latency
• You need to know the exact task_id and dag_id of the producer, creating tight coupling

The Dataset Approach (Airflow 2.4+)

# New way: Consumer triggers automatically when data is ready
orders_daily = Dataset("s3://data-lake/orders/daily/")

@dag(schedule=[orders_daily])
def new_consumer():
    @task
    def build_report():
        ...

    build_report()

The dataset approach:

• No sensor task consuming a worker slot
• No guessing when the producer finishes
• No arbitrary schedule offset to avoid races
• Consumer runs as soon as the data is available
• Producer does not need to know about the consumer

The scheduler handles the coordination internally, without any polling, without any running tasks, and without any network calls.

Combining Time-Based and Data-Aware Scheduling

Sometimes you need both. You want a DAG to run daily but also react to upstream data updates. Airflow 2.8+ introduced DatasetOrTimeSchedule for exactly this case:

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

orders_daily = Dataset("s3://data-lake/orders/daily/")

@dag(
    dag_id="hybrid_consumer",
    schedule=DatasetOrTimeSchedule(
        timetable=CronTriggerTimetable("0 6 * * *", timezone="UTC"),
        datasets=[orders_daily],
    ),
    ...
)
def hybrid_consumer():
    @task
    def process():
        """Runs at 6 AM daily OR whenever orders_daily is updated,
        whichever comes first."""
        ...

    process()

This pattern is useful for:

• Freshness guarantees — “Process new orders when they arrive, but if nothing has arrived by 6 AM, run anyway and handle the empty case”
• Graceful degradation — “Use fresh data if available, fall back to the scheduled run if the producer is down”
• SLA compliance — “We must produce a report by 8 AM. If upstream data is ready early, great. If not, run at 6 AM with whatever is available.”

Dataset URI Best Practices

Since dataset URIs are just strings, you need conventions to keep them manageable as your system grows.

Use Descriptive, Hierarchical URIs

# GOOD: Clear hierarchy, includes system and granularity
Dataset("s3://data-lake/orders/daily/")
Dataset("postgresql://warehouse/analytics.user_metrics")
Dataset("bigquery://project.dataset.revenue_daily")

# BAD: Ambiguous, no context
Dataset("orders")
Dataset("data")
Dataset("output")

A well-designed URI tells you what the data is, where it lives, and at what granularity — without needing to open the producer DAG.

One Dataset per Logical Data Product

Do not create a dataset for every file or every table partition. Create one dataset per logical data product — a collection of data that updates atomically and is consumed as a unit.

# GOOD: One dataset per logical product
orders_daily = Dataset("s3://data-lake/orders/daily/")
# The producer writes multiple files under this prefix;
# the dataset represents "today's orders are ready"

# BAD: One dataset per file -- too granular
orders_csv = Dataset("s3://data-lake/orders/daily/2026-07-12/orders.csv")
orders_meta = Dataset("s3://data-lake/orders/daily/2026-07-12/metadata.json")

Do Not Encode Dates in Dataset URIs

Dataset URIs should be stable across runs. If you embed a date, you create a new dataset every day, and your consumers will never trigger because they are listening for yesterday’s dataset.

# BAD: New dataset every day -- consumer never triggers
Dataset(f"s3://data-lake/orders/{today}/")

# GOOD: Stable URI -- same dataset, updated daily
Dataset("s3://data-lake/orders/daily/")

Document Your Datasets

As your dataset count grows, maintain a registry — even if it is just a Python module that defines all datasets in one place:

# dags/datasets.py -- Single source of truth for all datasets
from airflow.datasets import Dataset

# --- Orders domain ---
ORDERS_DAILY = Dataset("s3://data-lake/orders/daily/")
ORDERS_HOURLY = Dataset("s3://data-lake/orders/hourly/")

# --- Users domain ---
USER_PROFILES = Dataset("postgresql://warehouse/public.user_profiles")
USER_SEGMENTS = Dataset("s3://data-lake/ml/user_segments/")

# --- ML domain ---
RECOMMENDATION_MODEL = Dataset("s3://ml-models/recommendations/latest/")

Producers and consumers import from this module, ensuring everyone uses the same URIs:

from datasets import ORDERS_DAILY

@dag(schedule=[ORDERS_DAILY])
def my_consumer():
    ...

Viewing Dataset Lineage in the UI

Airflow’s web UI includes a Datasets view (available from the top navigation in Airflow 2.4+) that shows:

• All registered datasets
• Which DAGs produce each dataset (outlets)
• Which DAGs consume each dataset (schedule dependencies)
• The history of dataset update events

This view gives you a visual data lineage graph — you can see at a glance how data flows through your system without reading any code. It is not a full-featured data lineage tool like DataHub or OpenLineage, but it covers the “which DAG produces this data and which DAGs consume it” question effectively.

Common Patterns

Fan-Out: One Producer, Many Consumers

raw_events = Dataset("s3://lake/events/raw/")

# Producer
@dag(schedule="@hourly")
def ingest_events():
    @task(outlets=[raw_events])
    def load():
        ...

# Consumer 1: Analytics
@dag(schedule=[raw_events])
def build_analytics():
    ...

# Consumer 2: ML Features
@dag(schedule=[raw_events])
def compute_features():
    ...

# Consumer 3: Alerting
@dag(schedule=[raw_events])
def check_anomalies():
    ...

All three consumers trigger independently when raw_events is updated. No coordination needed, no sensor overhead.

Chain: Producer → Transformer → Consumer

raw_orders   = Dataset("s3://lake/orders/raw/")
clean_orders = Dataset("s3://lake/orders/clean/")

# Step 1: Ingest raw data
@dag(schedule="@daily")
def ingest():
    @task(outlets=[raw_orders])
    def extract():
        ...

# Step 2: Clean and transform (triggered by raw data)
@dag(schedule=[raw_orders])
def transform():
    @task(outlets=[clean_orders])
    def clean():
        ...

# Step 3: Build reports (triggered by clean data)
@dag(schedule=[clean_orders])
def report():
    @task
    def build():
        ...

This creates a clean pipeline where each stage triggers the next automatically. Adding a new stage or branching the pipeline requires no changes to upstream DAGs.

Conditional Dataset Updates

Sometimes a task should only update a dataset when certain conditions are met. You can control this by conditionally succeeding or skipping the task:

@task(outlets=[orders_daily])
def extract_if_data_exists(**context):
    """Only updates the dataset if there is actually new data."""
    hook = PostgresHook("source_db")
    ds = context["ds"]
    count = hook.get_first(
        f"SELECT COUNT(*) FROM orders WHERE date = '{ds}'"
    )[0]

    if count == 0:
        raise AirflowSkipException("No new orders today")

    # If we get here, the task succeeds and the dataset is updated
    records = hook.get_records(
        f"SELECT * FROM orders WHERE date = '{ds}'"
    )
    upload_to_s3(records, ds)

When a task is skipped (via AirflowSkipException), the dataset is not updated, and consumers are not triggered. This prevents empty or meaningless pipeline runs.

Limitations to Be Aware Of

Datasets are powerful but have boundaries:

• No cross-Airflow-instance datasets. Datasets work within a single Airflow deployment. If you have multiple Airflow instances, they cannot share dataset events.
• No payload in events. When a dataset is updated, there is no way to attach metadata (like “here is the S3 path of the new file”). Consumers need to determine the data location themselves.
• All-or-nothing triggering. When a consumer depends on multiple datasets, it triggers when all of them have been updated. There is no “trigger when any one of these datasets updates” (though you can work around this with separate consumer DAGs).
• No conditional logic on events. You cannot say “trigger only if the dataset was updated with more than 1000 records.” The trigger is binary: updated or not.

Next Steps

Datasets represent a significant shift in how Airflow handles inter-DAG dependencies. They are simpler, more efficient, and more maintainable than sensor-based approaches. Start by identifying your most painful sensor-based patterns — the ones that waste worker slots or require fragile schedule offsets — and migrate them to datasets.

• Dynamic DAGs — Combine dataset scheduling with dynamically generated DAGs for powerful multi-tenant data platforms.
• Production Deployment — Ensure your Airflow version (2.4+) supports datasets before planning your deployment.
• Real-World Patterns — See how datasets fit into complete production pipeline designs.