Asset Definitions

Added in version 2.4.

Changed in version 3.0: The concept was previously called “Dataset”.

What is an “Asset”?

An Airflow asset is a logical grouping of data. Upstream producer tasks can update assets, and asset updates contribute to scheduling downstream consumer Dags.

Uniform Resource Identifier (URI) define assets:

from airflow.sdk import Asset

example_asset = Asset("s3://asset-bucket/example.csv")

Airflow makes no assumptions about the content or location of the data represented by the URI, and treats the URI like a string. This means that Airflow treats any regular expressions, like input_\d+.csv, or file glob patterns, such as input_2022*.csv, as an attempt to create multiple assets from one declaration, and they will not work.

You must create assets with a valid URI. Airflow core and providers define various URI schemes that you can use, such as file (core), postgres (by the Postgres provider), and s3 (by the Amazon provider). Third-party providers and plugins might also provide their own schemes. These pre-defined schemes have individual semantics that are expected to be followed. You can use the optional name argument to provide a more human-readable identifier to the asset.

from airflow.sdk import Asset

example_asset = Asset(uri="s3://asset-bucket/example.csv", name="bucket-1")

What is valid URI?

Technically, the URI must conform to the valid character set in RFC 3986, which is basically ASCII alphanumeric characters, plus %, -, _, ., and ~. To identify a resource that cannot be represented by URI-safe characters, encode the resource name with percent-encoding.

The URI is also case sensitive, so s3://example/asset and s3://Example/asset are considered different. Note that the host part of the URI is also case sensitive, which differs from RFC 3986.

For pre-defined schemes (e.g., file, postgres, and s3), you must provide a meaning URI. If you can’t provide one, use another scheme altogether that don’t have the semantic restrictions. Airflow will never require a semantic for user-defined URI schemes (with a prefix x-), so that can be a good alternative. If you have a URI that can only be obtained later (e.g., during task execution), consider using AssetAlias instead and update the URI later.

# invalid asset:
must_contain_bucket_name = Asset("s3://")

Do not use the airflow scheme, which is reserved for Airflow’s internals.

Airflow always prefers using lower cases in schemes, and case sensitivity is needed in the host part of the URI to correctly distinguish between resources.

# invalid assets:
reserved = Asset("airflow://example_asset")
not_ascii = Asset("èxample_datašet")

If you want to define assets with a scheme that doesn’t include additional semantic constraints, use a scheme with the prefix x-. Airflow skips any semantic validation on URIs with these schemes.

# valid asset, treated as a plain string
my_ds = Asset("x-my-thing://foobarbaz")

The identifier does not have to be absolute; it can be a scheme-less, relative URI, or even just a simple path or string:

# valid assets:
schemeless = Asset("//example/asset")
csv_file = Asset("example_asset")

Non-absolute identifiers are considered plain strings that do not carry any semantic meanings to Airflow.

Extra information on assets

If needed, you can include an additional dictionary in an asset using the extra parameter:

example_asset = Asset(
    "s3://asset/example.csv",
    extra={"team": "trainees"},
)

This allows you to provide custom metadata about the asset, such as ownership information or the purpose of the file. The extra field does NOT affect the identity of an asset. Thus, maintaining the uniqueness of the extra value is the user responsibility. It suggested to have only one single set of extra value per asset.

For example, in the following snippet, only one of the extra dictionaries will ultimately be stored, but it does guaranteed which one will be stored.

Asset("s3://asset/example.csv", extra={"d": "e"})
Asset("s3://asset/example.csv", extra={"f": "g"})

This behavior also applies to dynamically generated assets created through AssetAlias. In the example below, the final stored extra value is not guaranteed and it might vary based on Dag processor settings.

from airflow.sdk import AssetAlias


@dag(schedule=None)
def my_dag_1():

    @task(outlets=[AssetAlias("my-task-outputs")])
    def my_task_with_outlet_events(*, outlet_events):
        outlet_events[AssetAlias("my-task-outputs")].add(
            # Asset extra set as {"from": "asset alias"}
            Asset("s3://bucket/my-task", extra={"from": "asset alias"})
        )

    my_task_with_outlet_events()


# Asset extra set as {"key": "value"}
@dag(schedule=Asset("s3://bucket/my-task", extra={"key": "value"}))
def my_dag_2(): ...


my_dag_1()
my_dag_2()

# It's not guaranteed which extra will be the one stored

Security Warnings

Secure naming of asset URIs: Asset URIs and values in the extra field are stored in cleartext in Airflow’s metadata database. These fields are not encrypted. DO NOT store sensitive information, especially credentials, in either the asset URI or the extra dictionary.
Security Implication of Asset Creation: In Airflow’s security model, granting the can_create permission on Assets is effectively equivalent to granting “trigger” permissions on all downstream Dags that depend on those assets. Because Airflow uses an “implicit trust” model for data-aware scheduling, any user who can create an Asset Event (via the API or a task) can trigger any Dag scheduled on that asset, even if the user does not have permission to view or edit the downstream Dags. Exercise caution when granting can_create on Assets in multi-tenant environments, as it allows users to influence workflows outside their direct scope.

Creating a task to emit asset events

Once an asset is defined, tasks can be created to emit events against it by specifying outlets:

from airflow.sdk import DAG, Asset
from airflow.providers.standard.operators.python import PythonOperator

example_asset = Asset(name="example_asset", uri="s3://asset-bucket/example.csv")

def _write_example_asset():
    """Write data to example_asset..."""

with DAG(dag_id="example_asset", schedule="@daily"):
    PythonOperator(task_id="example_asset", outlets=[example_asset], python_callable=_write_example_asset)

This is quite a lot of boilerplate. Airflow provides a shorthand for this simple but most common case of creating a Dag with one single task that emits events of one asset. The code block below is exactly equivalent to the one above:

from airflow.sdk import asset


@asset(uri="s3://asset-bucket/example.csv", schedule="@daily")
def example_asset():
    """Write data to example_asset..."""

Declaring an @asset automatically creates:

An Asset with name set to the function name.
A DAG with dag_id set to the function name.
A task inside the DAG with task_id set to the function name, and outlet to the created Asset.

The parameter names self, context, and outlet_events are reserved in an @asset function: they are populated by Airflow at runtime (with the asset itself, the execution context, and the outlet event accessor respectively) and are never treated as inlet asset references.

Attaching extra information to an emitting asset event

Added in version 2.10.0.

A task with an asset outlet can optionally attach extra information before it emits an asset event. This is different from Extra information on assets. Extra information on an asset statically describes the entity pointed to by the asset URI; extra information on the asset event instead should be used to annotate the triggering data change, such as how many rows in the database are changed by the update, or the date range covered by it.

The easiest way to attach extra information to the asset event is by yield-ing a Metadata object from a task:

from airflow.sdk import Metadata, asset


@asset(uri="s3://asset/example.csv", schedule=None)
def example_s3(self):  # 'self' here refers to the current asset.
    df = ...  # Get a Pandas DataFrame to write.
    # Write df to asset...
    yield Metadata(self, {"row_count": len(df)})

Airflow automatically collects all yielded metadata, and populates asset events with extra information for corresponding metadata objects.

This can also be done in classic operators. The best way is to subclass the operator and override execute. Alternatively, extras can also be added in a task’s pre_execute or post_execute hook. If you choose to use hooks, however, remember that they are not rerun when a task is retried, and may cause the extra information to not match actual data in certain scenarios.

Another way to achieve the same is by accessing outlet_events in a task’s execution context directly:

@asset(schedule=None)
def write_to_s3(self, context):
    context["outlet_events"][self].extra = {"row_count": len(df)}

There’s minimal magic here—Airflow simply writes the yielded values to the exact same accessor. This also works in classic operators, including execute, pre_execute, and post_execute.

Note

Asset event extra information can only contain JSON-serializable values (list and dict nesting is possible). This is due to the value being stored in the database.

Fetching information from previously emitted asset events

Added in version 2.10.0.

Events of an asset defined in a task’s outlets, as described in the previous section, can be read by a task that declares the same asset in its inlets. A asset event entry contains extra (see previous section for details), timestamp indicating when the event was emitted from a task, and source_task_instance linking the event back to its source.

Inlet asset events can be read with the inlet_events accessor in the execution context. Continuing from the write_to_s3 asset in the previous section:

@asset(schedule=None)
def post_process_s3_file(context, write_to_s3):  # Declaring an inlet to write_to_s3.
    events = context["inlet_events"][write_to_s3]
    last_row_count = events[-1].extra["row_count"]

Each value in the inlet_events mapping is a sequence-like object that orders past events of a given asset by timestamp, earliest to latest. It supports most of Python’s list interface, so you can use [-1] to access the last event, [-2:] for the last two, etc. The accessor is lazy and only hits the database when you access items inside it.

Dependency between `@asset`, `@task`, and classic operators

Since an @asset is simply a wrapper around a Dag with a task and an asset, it is quite easy to read and @asset in a @task or a classic operator. For example, the above post_process_s3_file can also be written as a task (inside a Dag, omitted here for brevity):

@task(inlets=[write_to_s3])
def post_process_s3_file(*, inlet_events):
    events = inlet_events[example_s3_asset]
    last_row_count = events[-1].extra["row_count"]


post_process_s3_file()

The other way around also applies:

example_asset = Asset("example_asset")


@task(outlets=[example_asset])
def emit_example_asset():
    """Write to example_asset..."""


@asset(schedule=None)
def process_example_asset(example_asset):
    """Process inlet example_asset..."""

In addition, @asset can be used with @task to customize the task that generates the asset, utilizing the modern TaskFlow approach described in Pythonic Dags with the TaskFlow API.

This combination allows you to set initial arguments for the task and to use various operators, such as the BashOperator:

@asset(schedule=None)
@task.bash(retries=3)
def example_asset():
    """Write to example_asset, from a Bash task with 3 retries..."""
    return "echo 'run'"

Output to multiple assets in one task

It is possible for a task to emit events for multiple assets. This is generally discouraged, but needed in certain situations, such as when you need to split a data source into several. This is straightforward with tasks since outlets is plural by design:

from airflow.sdk import DAG, Asset, task

input_asset = Asset("input_asset")
out_asset_1 = Asset("out_asset_1")
out_asset_2 = Asset("out_asset_2")

with DAG(dag_id="process_input", schedule=None):

    @task(inlets=[input_asset], outlets=[out_asset_1, out_asset_2])
    def process_input():
        """Split input into two."""

The shorthand for this is @asset.multi:

from airflow.sdk import Asset, asset

input_asset = Asset("input_asset")
out_asset_1 = Asset("out_asset_1")
out_asset_2 = Asset("out_asset_2")


@asset.multi(schedule=None, outlets=[out_asset_1, out_asset_2])
def process_input(input_asset):
    """Split input into two."""

Dynamic data events emitting and asset creation through AssetAlias

An asset alias can be used to emit asset events of assets with association to the aliases. Downstreams can depend on resolved asset. This feature allows you to define complex dependencies for Dag executions based on asset updates.

How to use AssetAlias

AssetAlias has one single argument name that uniquely identifies the asset. The task must first declare the alias as an outlet, and use outlet_events or yield Metadata to add events to it.

The following example creates an asset event against the S3 URI f"s3://bucket/my-task" with optional extra information extra. If the asset does not exist, Airflow will dynamically create it and log a warning message.

Emit an asset event during task execution through outlet_events

from airflow.sdk import AssetAlias


@task(outlets=[AssetAlias("my-task-outputs")])
def my_task_with_outlet_events(*, outlet_events):
    outlet_events[AssetAlias("my-task-outputs")].add(Asset("s3://bucket/my-task"), extra={"k": "v"})

Emit an asset event during task execution through yielding Metadata

from airflow.sdk import Metadata


@task(outlets=[AssetAlias("my-task-outputs")])
def my_task_with_metadata():
    s3_asset = Asset(uri="s3://bucket/my-task", name="example_s3")
    yield Metadata(s3_asset, extra={"k": "v"}, alias=AssetAlias("my-task-outputs"))

Only one asset event is emitted for an added asset, even if it is added to the alias multiple times, or added to multiple aliases. However, if different extra values are passed, it can emit multiple asset events. In the following example, two asset events will be emitted.

from airflow.sdk import AssetAlias


@task(
    outlets=[
        AssetAlias("my-task-outputs-1"),
        AssetAlias("my-task-outputs-2"),
        AssetAlias("my-task-outputs-3"),
    ]
)
def my_task_with_outlet_events(*, outlet_events):
    outlet_events[AssetAlias("my-task-outputs-1")].add(Asset("s3://bucket/my-task"), extra={"k": "v"})
    # This line won't emit an additional asset event as the asset and extra are the same as the previous line.
    outlet_events[AssetAlias("my-task-outputs-2")].add(Asset("s3://bucket/my-task"), extra={"k": "v"})
    # This line will emit an additional asset event as the extra is different.
    outlet_events[AssetAlias("my-task-outputs-3")].add(Asset("s3://bucket/my-task"), extra={"k2": "v2"})

Fetching information from previously emitted asset events through resolved asset aliases

As mentioned in Fetching information from previously emitted asset events, inlet asset events can be read with the inlet_events accessor in the execution context, and you can also use asset aliases to access the asset events triggered by them.

with DAG(dag_id="asset-alias-producer"):

    @task(outlets=[AssetAlias("example-alias")])
    def produce_asset_events(*, outlet_events):
        outlet_events[AssetAlias("example-alias")].add(Asset("s3://bucket/my-task"), extra={"row_count": 1})


with DAG(dag_id="asset-alias-consumer", schedule=None):

    @task(inlets=[AssetAlias("example-alias")])
    def consume_asset_alias_events(*, inlet_events):
        events = inlet_events[AssetAlias("example-alias")]
        last_row_count = events[-1].extra["row_count"]

Cross-team asset event filtering with `producer_teams`

Added in version 3.3.0.

When Multi-Team mode is enabled, asset events are filtered by team membership. By default, a consuming Dag only receives asset events produced by Dags within the same team or by global (teamless) Dags. This prevents unintended cross-team triggers.

To configure cross-team access, use the access_control parameter on the Asset definition with an AssetAccessControl instance:

from airflow.sdk import Asset, AssetAccessControl

shared_data = Asset(
    name="my_data",
    uri="s3://bucket/shared/data.csv",
    access_control=AssetAccessControl(
        producer_teams=["team_analytics", "team_ml"],
    ),
)

In this example, asset events produced by Dags belonging to team_analytics or team_ml will be accepted by any consuming Dag that schedules on shared_data, in addition to events from the consuming Dag’s own team.

`AssetAccessControl` parameters

The AssetAccessControl class accepts the following parameters:

producer_teams (list[str], default []): List of team names allowed to produce events consumed by this asset’s consumers, in addition to the consumer’s own team.
consumer_teams (list[str] | None, default None): List of team names allowed to consume events produced by this asset’s producers. See Cross-team asset event filtering with consumer_teams.
allow_global (bool, default True): Whether teamless (global) Dags can participate in cross-team event delivery. See Multi-Team for the full semantics on both consumer-side and producer-side assets.

Blocking global producers

By default, global (teamless) Dags can trigger any consumer. In strict team isolation scenarios, you may want to block teamless producers:

from airflow.sdk import Asset, AssetAccessControl

strict_data = Asset(
    name="strict_data",
    uri="s3://bucket/strict/data.csv",
    access_control=AssetAccessControl(
        producer_teams=["team_analytics"],
        allow_global=False,
    ),
)

With allow_global=False, only Dags belonging to the consumer’s own team or to team_analytics can trigger consumers of strict_data. Teamless Dag producers are blocked.

Note

The allow_global flag only affects Dag producers. Teamless API users are always restricted to triggering teamless consumers only, regardless of this setting.

Default behavior

When access_control is not specified, a default AssetAccessControl() is used (empty producer_teams, consumer_teams=None, and allow_global=True). See Multi-Team for the complete behavioral rules table. In summary, the rules depend on whether the producer and consumer have a team association:

Both have the same team: The event is always delivered.
Producer has a team, consumer has a different team: The event is blocked (unless the producer’s team is in the asset’s producer_teams).
Producer has no team (global Dag): The event is delivered to all consumers whose asset has allow_global=True (the default). Global Dags act as shared infrastructure that any team can depend on.
Consumer has no team (global Dag): The consumer accepts events from any source, regardless of the producer’s team. Teamless consumers act as shared infrastructure that any team can feed into.
Neither has a team: The event is delivered (both are global).

When Multi-Team mode is disabled, access_control is ignored and all asset events are delivered to all consuming Dags, preserving backward compatibility.

Asset partitions

Added in version 3.2.0.

Asset events can include a partition_key to make it _partitioned__. This lets you model the same asset at partition granularity (for example, 2026-03-10T09:00:00 for an hourly partition).

To produce partitioned events on a schedule, use CronPartitionTimetable in the producer Dag (or @asset). This timetable creates asset events with a partition key on each run.

from airflow.sdk import CronPartitionTimetable, asset


@asset(
    uri="file://incoming/player-stats/team_b.csv",
    schedule=CronPartitionTimetable("15 * * * *", timezone="UTC"),
)
def team_b_player_stats():
    pass

Partitioned events are intended for partition-aware downstream scheduling, and do not trigger non-partition-aware Dags.

Pre-determined vs runtime partitioning

Both kinds attach a partition key to a Dag run — the difference is when and by whom the key is decided:

Pre-determined partitioning — the partition key is worked out before the task runs, using the timetable’s schedule cadence and partition mappers to match upstream keys to downstream keys and trigger partition-based Dag runs. CronPartitionTimetable uses this kind as a producer; PartitionedAssetTimetable uses it as a consumer.
Runtime partitioning — the partition key is deferred to task runtime: the producing task records key(s) via outlet_events[self].add_partitions(...). PartitionedAtRuntime uses this kind and never schedules on its own (can_be_scheduled=False); a schedulable timetable can also defer to runtime by subclassing CronTriggerTimetable and setting partitioned_at_runtime = True (see the custom plugin example below).

A timetable uses one kind or the other, not both: it either resolves partitions ahead of the run or defers them to task runtime.

Practical rule: use CronPartitionTimetable when the partition key follows from the schedule cadence; use PartitionedAtRuntime when the key is only known once the task runs (e.g. a watermark from source data); use PartitionedAssetTimetable downstream to consume either kind.

For downstream partition-aware scheduling, use PartitionedAssetTimetable:

from airflow.sdk import DAG, StartOfHourMapper, PartitionedAssetTimetable

with DAG(
    dag_id="clean_and_combine_player_stats",
    schedule=PartitionedAssetTimetable(
        assets=team_a_player_stats & team_b_player_stats & team_c_player_stats,
        default_partition_mapper=StartOfHourMapper(),
    ),
    catchup=False,
):
    ...

PartitionedAssetTimetable requires partitioned asset events. If an asset event does not contain a partition_key, it will not trigger a downstream Dag that uses PartitionedAssetTimetable.

default_partition_mapper is used for every upstream asset unless you override it via partition_mapper_config. The default mapper is IdentityMapper (no key transformation).

Partition mappers define how upstream partition keys are transformed to the downstream Dag partition key:

IdentityMapper keeps keys unchanged.
Temporal mappers such as StartOfHourMapper, StartOfDayMapper, and StartOfYearMapper normalize time keys to a chosen grain. For input key 2026-03-10T09:37:51, the default outputs are:
- StartOfHourMapper -> 2026-03-10T09
- StartOfDayMapper -> 2026-03-10
- StartOfYearMapper -> 2026
ProductMapper maps composite keys segment-by-segment. It applies one mapper per segment and then rejoins the mapped segments. For example, with key us|2026-03-10T09:00:00, ProductMapper(IdentityMapper(), StartOfDayMapper()) produces us|2026-03-10.
AllowedKeyMapper validates that keys are in a fixed allow-list and passes the key through unchanged if valid. For example, AllowedKeyMapper(["us", "eu", "apac"]) accepts only those region keys and rejects all others.
FixedKeyMapper collapses every upstream key onto a fixed downstream key, regardless of the upstream value.
SegmentWindow declares a fixed categorical set of string keys (e.g. regions, tenants) that constitute one downstream period; paired with FixedKeyMapper inside a RollupMapper it holds the downstream run until every declared segment has arrived (see segment-rollup).

Example of per-asset mapper configuration and composite-key mapping:

from airflow.sdk import (
    Asset,
    IdentityMapper,
    PartitionedAssetTimetable,
    ProductMapper,
    StartOfDayMapper,
)

regional_sales = Asset(uri="file://incoming/sales/regional.csv", name="regional_sales")

with DAG(
    dag_id="aggregate_regional_sales",
    schedule=PartitionedAssetTimetable(
        assets=regional_sales,
        default_partition_mapper=ProductMapper(IdentityMapper(), StartOfDayMapper()),
    ),
):
    ...

You can also override mappers for specific upstream assets with partition_mapper_config:

from airflow.sdk import Asset, DAG, StartOfDayMapper, IdentityMapper, PartitionedAssetTimetable

hourly_sales = Asset(uri="file://incoming/sales/hourly.csv", name="hourly_sales")
daily_targets = Asset(uri="file://incoming/sales/targets.csv", name="daily_targets")

with DAG(
    dag_id="join_sales_and_targets",
    schedule=PartitionedAssetTimetable(
        assets=hourly_sales & daily_targets,
        # Default behavior: map timestamp-like keys to daily keys.
        default_partition_mapper=StartOfDayMapper(),
        # Override for assets that already emit daily partition keys.
        partition_mapper_config={
            daily_targets: IdentityMapper(),
        },
    ),
):
    ...

If transformed partition keys from all required upstream assets do not align, the downstream Dag will not be triggered for that partition.

The same applies when a mapper cannot transform a key. For example, if an upstream event has partition_key="random-text" and the downstream mapping uses DailyMapper (which expects a timestamp-like key), no downstream partition match can be produced, so the downstream Dag is not triggered for that key.

Inside partitioned Dag runs, access the resolved partition through dag_run.partition_key. When the consumer’s partition mapper can resolve the key to a datetime, that value is also available as dag_run.partition_date, so templates can use {{ partition_date | ds }}. This covers the StartOf*Mapper family (which decode the key directly), IdentityMapper (which carries the producer’s partition_date through), and composite mappers — RollupMapper, ChainMapper and FanOutMapper — whose effective child mapper is temporal (they delegate the anchor to that child). Mappers whose key carries no temporal meaning (ProductMapper, AllowedKeyMapper and custom mappers that do not implement to_partition_date) leave partition_date None even when the resulting key is date-shaped, so those consumers should keep parsing partition_key.

You can also trigger a DagRun manually with a partition key (for example, through the Trigger Dag window in the UI, or through the REST API by including partition_key in the request body):

curl -X POST "http://<airflow-host>/api/v2/dags/aggregate_regional_sales/dagRuns" \
  -H "Content-Type: application/json" \
  -d '{
    "logical_date": "2026-03-10T00:00:00Z",
    "partition_key": "us|2026-03-10T09:00:00"
  }'

Rollup mappers

Added in version 3.3.0.

The mappers shown above match upstream keys to a single downstream key one-for-one. For a coarser downstream period made up of many upstream events — an hourly upstream that drives a daily summary, daily inputs that compose a weekly report — use RollupMapper. RollupMapper composes an upstream mapper (which normalizes each upstream key to the downstream granularity) with a Window that declares the full set of upstream keys required for one downstream key. The scheduler holds the Dag run until every upstream key in the window has arrived; partial windows stay pending on the next-run-assets view so operators can see progress.

The shipped windows are HourWindow (sixty minutes per hour), DayWindow (twenty-four hours per day), WeekWindow (seven days per week), MonthWindow, QuarterWindow, and YearWindow. Pair each window with an upstream mapper that decodes to the same temporal grain — for example StartOfHourMapper with DayWindow.

Each upstream asset event carries a fine-grained partition key such as 2026-03-10T09:00:00 (second precision). StartOfHourMapper normalizes that key to the hour boundary 2026-03-10T09, which is the format it encodes each expected member in — the same strings the scheduler matches against the twenty-four required members of DayWindow.

The following hourly-to-daily example produces a daily summary once all twenty-four upstream hourly partitions for a calendar day have arrived:

from airflow.sdk import (
    DAG,
    Asset,
    CronPartitionTimetable,
    DayWindow,
    PartitionedAssetTimetable,
    RollupMapper,
    StartOfHourMapper,
    task,
)

hourly_sales = Asset(uri="file://incoming/sales/hourly.csv", name="hourly_sales")

# Producer: emits one partitioned event per hour (key looks like 2026-03-10T09:00:00).
with DAG(
    dag_id="ingest_hourly_sales",
    schedule=CronPartitionTimetable("0 * * * *", timezone="UTC"),
):

    @task(outlets=[hourly_sales])
    def ingest():
        pass

    ingest()

# Consumer: fires once a day's twenty-four hourly partitions are all in.
with DAG(
    dag_id="daily_sales_summary",
    schedule=PartitionedAssetTimetable(
        assets=hourly_sales,
        default_partition_mapper=RollupMapper(
            upstream_mapper=StartOfHourMapper(),
            window=DayWindow(),
        ),
    ),
    catchup=False,
):

    @task
    def summarize(dag_run=None):
        # dag_run.partition_key is the day, e.g. "2026-03-10".
        print(dag_run.partition_key)

    summarize()

A misconfigured RollupMapper — e.g. pairing an identity-decoding upstream mapper with a DayWindow — raises TypeError at Dag parse so the misconfiguration surfaces immediately instead of silently holding every downstream run forever. The error is a type mismatch: an identity-decoding mapper’s expected_decoded_type is str, but temporal windows such as DayWindow require datetime; RollupMapper detects the mismatch at construction time and raises before the Dag is scheduled.

DayWindow always enumerates twenty-four hourly steps. With an upstream mapper configured for a local timezone that observes daylight-saving time, the spring-forward day has only twenty-three real hours (one window member never has a matching event, so the run is held indefinitely) and the fall-back day has twenty-five (the repeated hour is dropped). Use a UTC-based upstream mapper for any rollup that crosses a DST boundary; see the DayWindow class docstring for the full discussion.

Wait policies

RollupMapper accepts an optional wait_policy argument that decides when the downstream Dag run fires given the expected window vs the upstream keys that have actually arrived.

WaitForAll (the default) holds the run until every expected upstream key in the window has arrived.
MinimumCount (n) fires early once at least n of the expected keys have arrived — useful to tolerate a slow or missing upstream partition rather than holding the run indefinitely.

from airflow.sdk import (
    DAG,
    Asset,
    FixedKeyMapper,
    MinimumCount,
    PartitionedAtRuntime,
    PartitionedAssetTimetable,
    RollupMapper,
    SegmentWindow,
    asset,
)


@asset(
    uri="file://incoming/player-stats/multi-region.csv",
    schedule=PartitionedAtRuntime(),
)
def multi_region_player_stats(self, outlet_events):
    outlet_events[self].add_partitions(["us", "eu", "apac"])


# Consumer: fires once at least two of the three declared region partitions arrive.
with DAG(
    dag_id="segment_region_stats_early_rollup",
    schedule=PartitionedAssetTimetable(
        assets=Asset.ref(name="multi_region_player_stats"),
        default_partition_mapper=RollupMapper(
            upstream_mapper=FixedKeyMapper("all_regions"),
            window=SegmentWindow(["us", "eu", "apac"]),
            wait_policy=MinimumCount(2),
        ),
    ),
    catchup=False,
):
    ...

MinimumCount(-1) is the relative spelling of the same threshold — “at most one missing” — and is equivalent to MinimumCount(2) for a three-member window. Pass WaitForAll explicitly when you want to document intent rather than relying on the default.

Segment (categorical) rollup

Added in version 3.3.0.

For categorical partitioning — regions, tenants, experiment variants — compose a RollupMapper from two primitives:

SegmentWindow(["us", "eu", "apac"]) declares the fixed set of string keys that constitute one downstream period; to_upstream returns the full set regardless of the downstream anchor.
FixedKeyMapper("all_regions") collapses every upstream key onto the single downstream partition key "all_regions".

The scheduler holds the downstream Dag run until every declared segment has arrived from the upstream producer, then fires once. All the segment events accumulate into one AssetPartitionDagRun; the fired run’s partition_key is the value passed to FixedKeyMapper. This composition only makes sense under WAIT_FOR_ALL semantics (the default).

from airflow.sdk import (
    DAG,
    Asset,
    FixedKeyMapper,
    PartitionedAtRuntime,
    PartitionedAssetTimetable,
    RollupMapper,
    SegmentWindow,
    asset,
    task,
)


@asset(
    uri="file://incoming/player-stats/multi-region.csv",
    schedule=PartitionedAtRuntime(),
)
def multi_region_player_stats(self, outlet_events):
    # Emit one event per region in a single run.
    outlet_events[self].add_partitions(["us", "eu", "apac"])


# Consumer: fires once all three region partitions have arrived.
with DAG(
    dag_id="segment_region_stats_rollup",
    schedule=PartitionedAssetTimetable(
        assets=Asset.ref(name="multi_region_player_stats"),
        default_partition_mapper=RollupMapper(
            upstream_mapper=FixedKeyMapper("all_regions"),
            window=SegmentWindow(["us", "eu", "apac"]),
        ),
    ),
    catchup=False,
):

    @task
    def aggregate_all_regions(dag_run=None):
        # dag_run.partition_key is the downstream key once all segments arrive.
        print(dag_run.partition_key)

    aggregate_all_regions()

Construction validates both components: SegmentWindow raises ValueError for an empty list, non-string items, or empty-string keys; duplicate entries are silently deduplicated. FixedKeyMapper raises ValueError if its argument is not a non-empty string. Pass a distinct FixedKeyMapper key when one consumer Dag rolls up more than one asset, so each rollup uses a distinct bucket and they do not collide on the same (target_dag_id, partition_key).

For a segment set that must be computed at runtime, do not encode it here — evaluate completeness in a consumer-side task instead (the scheduler must not run user code to decide a partition set).

Setting partition keys at runtime

When the partition key is not known ahead of time (for example, a watermark discovered from the source data, a late-arriving file, or a backfill request), let the producing task decide it while it runs. Schedule the producer with PartitionedAtRuntime() and record the key(s) on the emitted event with outlet_events[self].add_partitions(...):

from airflow.sdk import PartitionedAtRuntime, asset


@asset(
    uri="file://incoming/player-stats/live-region.csv",
    schedule=PartitionedAtRuntime(),
)
def live_region_player_stats(self, outlet_events):
    # The key is only known once the task runs.
    outlet_events[self].add_partitions("us")

Inside an @asset function, self (the emitted Asset) and outlet_events (the outlet event accessor) are reserved parameter names that Airflow populates at runtime. Pass a single key, or a list to fan out to several partitions in one run. Each key produces its own asset event, and duplicate keys collapse to a single event:

@asset(
    uri="file://incoming/player-stats/multi-region.csv",
    schedule=PartitionedAtRuntime(),
)
def multi_region_player_stats(self, outlet_events):
    outlet_events[self].add_partitions(["us", "eu", "apac"])

When a runtime run emits exactly one partition key, the producing dag_run.partition_key is back-filled to that key. Downstream Dags consume these events the same way as timetable-produced partitions, through PartitionedAssetTimetable.

Fan-out mappers

Added in version 3.3.0.

FanOutMapper is the mirror of RollupMapper: instead of holding many upstream events until one downstream run can fire, a single upstream event fans out to one downstream Dag run per window member. It composes an upstream_mapper (which normalizes the upstream key to the window anchor) with a Window that enumerates the downstream period, and an optional downstream_mapper that converts each window member into a downstream partition key string.

For temporal windows (WeekWindow, MonthWindow, etc.) a default downstream_mapper is applied automatically — for example WeekWindow defaults to StartOfDayMapper, so each of the seven daily members is encoded as a YYYY-MM-DD string. For SegmentWindow there is no default-table entry, so an explicit downstream_mapper is required.

The following example fans a weekly model artifact out to seven daily inference runs — one Dag run per day in the week:

from airflow.sdk import (
    DAG,
    Asset,
    CronPartitionTimetable,
    FanOutMapper,
    PartitionedAssetTimetable,
    StartOfWeekMapper,
    WeekWindow,
    task,
)

weekly_model_artifact = Asset(uri="file://artifacts/models/weekly.bin", name="weekly_model_artifact")

# Producer: emits one partitioned event per week (key is the Monday date).
with DAG(
    dag_id="train_weekly_model",
    schedule=CronPartitionTimetable("0 0 * * 1", timezone="UTC"),
    catchup=False,
):

    @task(outlets=[weekly_model_artifact])
    def train_model():
        pass

    train_model()


# Consumer: one Dag run per day derived from the weekly upstream event.
with DAG(
    dag_id="daily_inference",
    schedule=PartitionedAssetTimetable(
        assets=weekly_model_artifact,
        default_partition_mapper=FanOutMapper(
            upstream_mapper=StartOfWeekMapper(),
            window=WeekWindow(),
            max_downstream_keys=7,
        ),
    ),
    catchup=False,
):

    @task
    def run_inference(dag_run=None):
        # dag_run.partition_key is one daily key, e.g. "2026-03-10".
        print(dag_run.partition_key)

    run_inference()

max_downstream_keys caps how many downstream Dag runs one upstream event may create. When exceeded, the runs for that event are not queued and a “partition fan-out exceeded” audit-log entry is recorded instead. Omitting it falls back to the global [scheduler] partition_mapper_max_downstream_keys config (default 1000). Set it explicitly to document intent and guard against accidental fan-out explosions.

Window direction: FORWARD and BACKWARD

Every Window supports a direction parameter that controls which period the window enumerates relative to its anchor.

Window.Direction.FORWARD (the default) — yields the period starting at the upstream key. For a weekly upstream key "2026-03-09" (Monday), WeekWindow() yields the seven days 2026-03-09 through 2026-03-15.
Window.Direction.BACKWARD — yields the trailing period ending at the key. The same "2026-03-09" key with WeekWindow(direction=Window.Direction.BACKWARD) yields the seven days ending on that Monday (2026-03-03 through 2026-03-09).

from airflow.sdk import FanOutMapper, PartitionedAssetTimetable, StartOfWeekMapper, WeekWindow, Window

PartitionedAssetTimetable(
    assets=weekly_model_artifact,
    default_partition_mapper=FanOutMapper(
        upstream_mapper=StartOfWeekMapper(),
        window=WeekWindow(direction=Window.Direction.BACKWARD),
    ),
)

Direction applies to rollup windows too — RollupMapper uses the same window classes, so DayWindow(direction=Window.Direction.BACKWARD) holds a downstream run until the twenty-four hours preceding midnight of the downstream key have all arrived.

Custom partition mappers, windows, and timetables in plugins

Custom PartitionMapper, Window, and partition-aware Timetable classes can be shipped as Airflow plugins by listing them in AirflowPlugin.partition_mappers, .windows, and .timetables respectively. Once a plugin is installed, these classes become usable in PartitionedAssetTimetable and RollupMapper without modifying core Airflow.

Custom partition mapper — strip a namespace prefix so that upstream keys like "eu::daily-sales" and "us::daily-sales" both collapse to the downstream key "daily-sales":

airflow/example_dags/plugins/custom_partition_mapper.py[source]

class PrefixStripMapper(PartitionMapper):
    """
    A partition mapper that strips a fixed namespace prefix from upstream keys.

    Upstream systems often qualify partition keys with a region or environment
    prefix — for example ``"eu::daily-sales"`` or ``"us::daily-sales"``.  A
    downstream asset that aggregates across regions only cares about the base key
    (``"daily-sales"``).  ``PrefixStripMapper`` strips the given prefix (including
    a configurable separator) so that all upstream namespaces collapse to the
    same downstream partition key.

    If the upstream key does not start with the configured prefix the key is
    returned unchanged, which is deliberate: keys that already live in the target
    namespace pass through without modification.

    This class demonstrates registering a custom :class:`PartitionMapper
    <airflow.partition_mappers.base.PartitionMapper>` subclass via the
    ``AirflowPlugin.partition_mappers`` registry. Any plugin that lists it in
    ``partition_mappers = [...]`` makes it available to
    :class:`~airflow.sdk.PartitionedAssetTimetable` and
    :class:`~airflow.partition_mappers.base.RollupMapper` without modifying core
    Airflow.

    :param prefix: The namespace prefix to strip, e.g. ``"eu"``.
    :param separator: The string that separates the prefix from the base key.
        Defaults to ``"::"`` to match a common ``"region::key"`` convention.
    """

    def __init__(
        self,
        prefix: str,
        *,
        separator: str = "::",
        max_downstream_keys: int | None = None,
    ) -> None:
        super().__init__(max_downstream_keys=max_downstream_keys)
        if not prefix:
            raise ValueError("prefix must be a non-empty string.")
        self.prefix = prefix
        self.separator = separator

    def to_downstream(self, key: str) -> str:
        full_prefix = self.prefix + self.separator
        if key.startswith(full_prefix):
            return key[len(full_prefix) :]
        return key

    def serialize(self) -> dict[str, Any]:
        data: dict[str, Any] = {"prefix": self.prefix, "separator": self.separator}
        if self.max_downstream_keys is not None:
            data["max_downstream_keys"] = self.max_downstream_keys
        return data

    @classmethod
    def deserialize(cls, data: dict[str, Any]) -> PartitionMapper:
        return cls(
            prefix=data["prefix"],
            separator=data.get("separator", "::"),
            max_downstream_keys=data.get("max_downstream_keys"),
        )


class PrefixStripMapperPlugin(AirflowPlugin):
    name = "prefix_strip_mapper_plugin"
    partition_mappers = [PrefixStripMapper]

Custom rollup window — yield only weekday period-starts from a calendar month so a downstream asset waits only for business-day upstream partitions:

airflow/example_dags/plugins/business_day_window.py[source]

class BusinessDayWindow(Window):
    """
    A calendar-month rollup window that yields only weekday (Mon–Fri) period-starts.

    The built-in :class:`~airflow.partition_mappers.window.MonthWindow` yields every
    calendar day in the month. ``BusinessDayWindow`` skips Saturdays and Sundays, so
    a monthly downstream asset only waits for the business-day upstream partitions —
    useful for financial or operational pipelines whose upstream data isn't produced
    on weekends.

    This class demonstrates registering a custom ``Window`` subclass via the
    ``AirflowPlugin.windows`` registry: any plugin that lists it in ``windows = [...]``
    makes it available to ``RollupMapper`` without modifying core Airflow.

    *Assumes FORWARD direction and a day-1 month anchor* — the standard contract for
    month-aligned temporal upstream mappers.
    """

    expected_decoded_type: ClassVar[type] = datetime

    def to_upstream(self, period_start: datetime) -> Iterable[datetime]:
        if period_start.day != 1:
            raise ValueError(
                f"BusinessDayWindow expects a period start on day 1 of the month, "
                f"got {period_start.isoformat()}."
            )
        days_in_month = calendar.monthrange(period_start.year, period_start.month)[1]
        return (day for i in range(days_in_month) if (day := period_start + timedelta(days=i)).weekday() < 5)


class BusinessDayWindowPlugin(AirflowPlugin):
    name = "business_day_window_plugin"
    windows = [BusinessDayWindow]

Custom runtime-partitioned timetable — a schedulable cron timetable that defers the partition key to task runtime, so the producing task can check whether the period’s data exists before emitting a partition:

airflow/example_dags/plugins/custom_partition_timetable.py[source]

class ScheduledRuntimePartitionTimetable(CronTriggerTimetable):
    """
    A schedulable timetable whose partition key is decided at task runtime.

    Runs fire on the given cron cadence, exactly like an ordinary
    :class:`~airflow.timetables.trigger.CronTriggerTimetable`. The partition key,
    however, is not derived from the schedule: it is set while the producing task
    runs — typically after the task checks whether the period's source data has
    arrived — by calling ``outlet_events[self].add_partitions(...)``.

    This uses runtime partitioning on a regular cron schedule: the timetable stays
    schedulable (``can_be_scheduled`` is ``True``) yet sets
    ``partitioned_at_runtime = True`` so the partition key is deferred to task
    runtime. It differs from :class:`~airflow.sdk.PartitionedAtRuntime` (which also
    defers the key to runtime but never schedules a run on its own) and from
    :class:`~airflow.timetables.trigger.CronPartitionTimetable` (which works out the
    partition key from the cadence ahead of the run). ``partitioned`` stays
    ``False``: no partition key is worked out ahead of the run.

    Registering it via the ``AirflowPlugin.timetables`` registry makes it usable
    by Dag authors without modifying core Airflow.
    """

    partitioned_at_runtime = True


class ScheduledRuntimePartitionTimetablePlugin(AirflowPlugin):
    name = "scheduled_runtime_partition_timetable_plugin"
    timetables = [ScheduledRuntimePartitionTimetable]

A producer Dag schedules on the timetable’s cron cadence and decides the partition key at runtime — emitting it only when the period’s upstream data is present. If the data has not arrived, the task simply does not call add_partitions (an empty or None key is rejected anyway), so no partitioned event — and therefore no downstream PartitionedAssetTimetable run — is produced for that period:

from airflow.sdk import DAG, Asset, task

# ScheduledRuntimePartitionTimetable is provided by the plugin registered above.
from my_plugin.plugins import ScheduledRuntimePartitionTimetable

daily_export = Asset(uri="file://exports/daily.csv", name="daily_export")

with DAG(
    dag_id="export_when_ready",
    schedule=ScheduledRuntimePartitionTimetable("0 6 * * *", timezone="UTC"),
    catchup=False,
):

    @task(outlets=[daily_export])
    def export(*, outlet_events):
        # Fires every day at 06:00 UTC. The partition key is not fixed by the
        # schedule — decide it at runtime from the data itself: find which
        # day's source file has actually landed.
        partition_key = latest_ready_day("s3://raw")  # your own check; e.g. "2026-06-23" or None
        if partition_key:
            build_export(partition_key)
            outlet_events[daily_export].add_partitions(partition_key)
        # If nothing is ready, emit nothing: with no partition key recorded,
        # no partitioned event is produced and no downstream
        # PartitionedAssetTimetable run is triggered for this period.

    export()

For complete runnable examples, see airflow-core/src/airflow/example_dags/example_asset_partition.py.

Asset Definitions

What is an “Asset”?

What is valid URI?

Extra information on assets

Security Warnings

Creating a task to emit asset events

Attaching extra information to an emitting asset event

Fetching information from previously emitted asset events

Dependency between @asset, @task, and classic operators

Output to multiple assets in one task

Dynamic data events emitting and asset creation through AssetAlias

How to use AssetAlias

Fetching information from previously emitted asset events through resolved asset aliases

Cross-team asset event filtering with producer_teams

AssetAccessControl parameters

Blocking global producers

Default behavior

Asset partitions

Pre-determined vs runtime partitioning

Rollup mappers

Wait policies

Segment (categorical) rollup

Setting partition keys at runtime

Fan-out mappers

Window direction: FORWARD and BACKWARD

Custom partition mappers, windows, and timetables in plugins

Dependency between `@asset`, `@task`, and classic operators

Cross-team asset event filtering with `producer_teams`

`AssetAccessControl` parameters