Partitioned Dag runs and asset events in Apache Airflow®

Airflow 3.2 introduced the concept of partitioned Dag runs and partitioned asset events. A partitioned Dag run is a Dag run with a partition_key attached to it, and a partitioned asset event is an asset event that has a partition_key attached to it. Any string can be a partition key, with time-based partition keys being the most common. Partition keys can be used in tasks in a partitioned Dag run to partition data, for example in a SQL statement.

In this guide, you’ll learn:

• When to use partitioned Dag runs and asset events.
• How to create a partitioned Dag run.
• How to create a partitioned asset event.
• How to schedule a Dag based on partitioned asset events.

Assumed knowledge

To get the most out of this guide, you should have an existing knowledge of:

• Airflow scheduling concepts. See Schedule Dags in Airflow.
• Basic asset-based scheduling. See Basic asset-based scheduling in Apache Airflow®.

When to use partitions

Situations in which you should consider using partitioned Dag runs and asset events are:

• When you want to process data from a specific time period in every Dag run. For example, if you have a Dag that runs once a day and should always process the data from the previous day.
• When you have a Dag that is scheduled based on asset events, within which you want to process data from a specific time period. You can use the partition_key to partition the data inside of the downstream Dag run. The partition_key propagates from the upstream Dag run to the downstream Dag run, and you can adjust its grain with partition key mappers.
• When you want to process data for a specific segment in manual or API-triggered Dag runs. For example, if your Dag generates a report for a specific department, you can set a different partition_key for each Dag run in the Trigger Dag run config or the API request body.

Create a partitioned Dag run

A partitioned Dag run is a Dag run with a partition_key attached to it. There are four ways to create a partitioned Dag run:

• By running a Dag manually in the Airflow UI and attaching a partition_key in the Trigger Dag run config.

  

• By running a Dag using the Airflow REST API and attaching a partition_key in the request body.

• By fulfilling the asset schedule condition of a Dag that uses the PartitionedAssetTimetable timetable.

• By using the CronPartitionTimetable timetable. Note that only Dag runs of the type scheduled and backfill are partitioned, manual runs are not, unless you are providing a partition_key in the Trigger Dag run config.

CronPartitionTimetable

The CronPartitionTimetable is a timetable that creates partitioned Dag runs with an automatic partition_key attached that is based on the run_after timestamp of each scheduled and backfilled Dag run.

1
from airflow.sdk import CronPartitionTimetable
2
3
@dag(
4
    schedule=CronPartitionTimetable("0 0 * * *", timezone="UTC"),
5
)

You can offset the partition key by providing the run_offset parameter to the CronPartitionTimetable instance. The offset is relative to the cron expression, for example if your Dag runs once every hour, a run_offset of -12 will offset the partition key by 12 hours. So the Dag run with the run id 2026-03-16T09:00:00 will have a partition key of 2026-03-15T21:00:00.

1
@dag(
2
    schedule=CronPartitionTimetable("0 * * * *", timezone="UTC", run_offset=-12),
3
)

Create a partitioned asset event

A partitioned asset event is an asset event that has a partition_key attached to it. There are four ways to create a partitioned asset event:

• By updating an asset manually in the Airflow UI and providing a partition_key in the Asset Event creation dialog.

  

• By updating an asset using the Airflow REST API and providing a partition_key in the request body.

• By updating an asset using the outlets parameter of a task in a Dag that is scheduled using a CronPartitionTimetable timetable.

• (Airflow 3.3+) By updating an asset using the outlets parameter of a task and using the .add_partitions method on the asset event object inside the task to add one or more partition_key values to the asset event. See Create partitions in a task.

The example below shows a Dag that is scheduled using a CronPartitionTimetable timetable. In every scheduled or backfilled run of this Dag, successful completion of the my_task_partitioned_upstream task will create a partitioned asset event for the my_partitioned_asset asset. The partition_key will be the run_after timestamp of the Dag run.

1
from airflow.sdk import dag, task, Asset, CronPartitionTimetable
2
3
4
my_asset = Asset("my_partitioned_asset")
5
6
7
@dag(
8
    schedule=CronPartitionTimetable("0 0 * * *", timezone="UTC"),
9
)
10
def my_dag_partitioned_upstream():
11
    @task(outlets=[my_asset])
12
    def my_task_partitioned_upstream(**context):
13
        pass
14
15
    my_task_partitioned_upstream()
16
17
18
my_dag_partitioned_upstream()

Create partitions in a task

In Airflow 3.3+ you can assign partition_key values at runtime from within your task code, using the .add_partitions method of the asset event fetched from the Airflow context.

The example Dag below contains one task (my_task) that updates one asset my_partition_at_runtime_asset, creating 3 asset events at runtime, each with a different partition key. In this example the task chooses 3 random values from the list of departments.

1
import random
2
3
from airflow.sdk import (
4
    Asset,
5
    dag,
6
    task
7
)
8
9
data_ready = Asset("my_partition_at_runtime_asset")
10
11
DEPARTMENTS = [
12
    "Engineering",
13
    "Product",
14
    "Legal",
15
    "Sales",
16
    "Marketing",
17
    "Finance",
18
    "HR",
19
    "Operations",
20
]
21
22
23
@dag
24
def my_partition_at_runtime_dag():
25
26
    @task(outlets=[data_ready])
27
    def my_task(**context):
28
        context["outlet_events"][data_ready].add_partitions(
29
            random.sample(DEPARTMENTS, 3)
30
        )
31
32
    my_task()
33
34
my_partition_at_runtime_dag()

In the Asset Events tab of the task instance you can see 3 asset events for the my_partition_at_runtime_asset were created, each with a different partition key value.

A Dag scheduled to run based on partitioned asset events of the my_partition_at_runtime_asset asset will start 3 runs, one for each partition key.

Schedule on a partitioned asset

To schedule a Dag on a partitioned asset, you set its schedule parameter to an instance of PartitionedAssetTimetable.

1
from airflow.sdk import dag, PartitionedAssetTimetable, Asset
2
3
@dag(schedule=PartitionedAssetTimetable(assets=Asset("my_partitioned_asset")))

This Dag will run whenever the my_partitioned_asset is updated by a partitioned asset event. It will not run based on regular asset events produced for the my_partitioned_asset asset.

You can modify the grain of the partition key by providing a partition key mapper to the PartitionedAssetTimetable instance. For example, to partition the data by day, you can use the StartOfDayMapper to normalize the partition key to the day in the format YYYY-MM-DD. See Partition key mappers for more information on the available partition key mappers.

1
from airflow.sdk import dag, PartitionedAssetTimetable, Asset, StartOfDayMapper
2
3
my_partitioned_asset = Asset("my_partitioned_asset")
4
5
@dag(
6
    schedule=PartitionedAssetTimetable(
7
        assets=my_partitioned_asset,
8
        partition_mapper_config={my_partitioned_asset: StartOfDayMapper()}
9
    )
10
)
11
def my_dag():
12
13
    @task 
14
    def my_task(**context):
15
        print(context["dag_run"].partition_key) # will print the partition key in the format `YYYY-MM-DD`
16
17
    my_task()
18
19
my_dag()

Combined partitioned asset schedules

You can combine multiple assets in a single PartitionedAssetTimetable instance to create a composite asset schedule using the same logical expressions (AND (&) plus OR (|)) as when creating a conditional asset schedule with regular assets.

1
from airflow.sdk import dag, PartitionedAssetTimetable, Asset
2
3
my_combined_asset_one = Asset("my_combined_asset_one")
4
my_combined_asset_two = Asset("my_combined_asset_two")
5
6
7
@dag(
8
    schedule=(
9
        PartitionedAssetTimetable(assets=(my_combined_asset_one & my_combined_asset_two))
10
    )
11
)
12
def my_combined_partitioned_asset_dag():
13
    @task
14
    def my_task(**context):
15
        partition_key = context["dag_run"].partition_key
16
        print(partition_key)
17
18
    my_task()
19
20
21
my_combined_partitioned_asset_dag()

If one of the assets is updated with a partitioned asset event, a pending run of this Dag will be created. Pending runs are visible in the Airflow UI by clicking on its schedule.

If several assets have been updated with a partitioned asset event of different partition keys, several pending runs are created, each with a different partition key. Each pending run has a visualization of which assets have been updated for that specific partition key and which assets are still pending.

A run of this Dag is only triggered when both my_combined_asset_one and my_combined_asset_two are updated with a partitioned asset event sharing the same partition key. Note that queued partitioned asset events for other partition keys are not reset by this.

You can use different partition key mappers for each asset in the PartitionedAssetTimetable, see Partition key mappers.

Partition keys

Partition keys are strings attached to partitioned Dag runs and asset events. You can use them in tasks in a partitioned Dag run to partition data, for example in a SQL statement.

Accessing partition keys

The partition_key can be accessed inside any task in a partitioned Dag run from within the Airflow context or by using Jinja templating.

1
# from airflow.sdk import task
2
# from airflow.providers.standard.operators.bash import BashOperator
3
4
@task
5
def print_partition_key(**context):
6
    print(context["partition_key"])
7
8
9
BashOperator(
10
    task_id="print_partition_key",
11
    bash_command="echo {{ partition_key }}",
12
)

One of the most common use cases is to use the partition key in a SQL statement to partition the data used in a specific Dag run. For example, in a Dag that runs once a day, you can use the partition key to select the data for the previous day.

1
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
2
3
SQLExecuteQueryOperator(
4
    task_id="execute_query",
5
    conn_id="my_snowflake_conn",
6
    sql="""
7
    SELECT * FROM my_table 
8
    WHERE 
9
        my_timestamp >= DATEADD(day, -1, '{{ partition_key }}'::DATE) 
10
        AND my_timestamp < '{{ partition_key }}'::DATE
11
    ;""",
12
)

In Airflow 3.3+ you can also directly access partition_date in partitioned Dag runs to format the date directly, for example with ds to convert the full key into yyyy-mm-dd format.

1
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
2
3
SQLExecuteQueryOperator(
4
    task_id="execute_query",
5
    conn_id="my_snowflake_conn",
6
    sql="""
7
    SELECT * FROM my_table 
8
    WHERE 
9
        my_timestamp >= DATEADD(day, -1, '{{ partition_date | ds }}') 
10
        AND my_timestamp < '{{ partition_date | ds }}'
11
    ;""",
12
)

Partition key mappers

Partition key mappers are used to modify the partition key of a Dag run. You can use them to change the grain of the partition key, to map composite keys segment by segment, to create one-to-many or many-to-one patterns, or to validate that keys are in a fixed allow-list.

One-to-one mappers

The following one-to-one partition key mappers are available:

• IdentityMapper: keeps keys unchanged. Default mapper.
• Temporal mappers change the grain of the partition key:
  • StartOfHourMapper: normalizes time keys to the hour in the format YYYY-MM-DDTHH, for example the partition key 2026-03-16T09:37:51 is mapped to 2026-03-16T09.
  • StartOfDayMapper: normalizes time keys to the day in the format YYYY-MM-DD (2026-03-16T09:37:51 -> 2026-03-16)
  • StartOfWeekMapper: normalizes time keys to the week in the format YYYY-MM-DD (W%V) (2026-03-16T09:37:51 -> 2026-03-16 (W12))
  • StartOfMonthMapper: normalizes time keys to the month in the format YYYY-MM (2026-03-16T09:37:51 -> 2026-03)
  • StartOfQuarterMapper: normalizes time keys to the quarter in the format YYYY-Q<n> (2026-03-16T09:37:51 -> 2026-Q1). The quarters are based on the calendar year, Q1 starts in January, Q2 in April, Q3 in July, and Q4 in October.
  • StartOfYearMapper: normalizes time keys to the year in the format YYYY (2026-03-16T09:37:51 -> 2026).
• ProductMapper: maps composite keys segment by segment, applying one mapper per segment and then rejoining the mapped segments. For example, with the key Finance|2026-03-16T09:00:00, ProductMapper(IdentityMapper(), StartOfDayMapper()) produces Finance|2026-03-16. See Composite partition keys.
• AllowedKeyMapper: validates that keys are in a fixed allow-list and passes the key through unchanged if valid. For example, AllowedKeyMapper(["Marketing", "Finance", "Sales"]) accepts only those department keys and rejects all others.

You can also change the default partition key mapper for all assets in the PartitionedAssetTimetable by providing a default_partition_mapper parameter.

1
from airflow.sdk import dag, PartitionedAssetTimetable, Asset, StartOfDayMapper
2
3
@dag(
4
    schedule=PartitionedAssetTimetable(
5
        assets=Asset("my_partitioned_asset"),
6
        default_partition_mapper=StartOfDayMapper(),
7
    )
8
) 

To override the default partition key mapper for a specific asset, you can set the partition_mapper_config parameter of the PartitionedAssetTimetable instance to a dictionary of asset instances and partition key mappers.

1
from airflow.sdk import dag, PartitionedAssetTimetable, Asset, StartOfDayMapper, StartOfWeekMapper
2
3
@dag(
4
    schedule=PartitionedAssetTimetable(
5
        assets=Asset("my_partitioned_asset"),
6
        default_partition_mapper=StartOfDayMapper(),
7
        partition_mapper_config={
8
            Asset("my_partitioned_asset"): StartOfWeekMapper(),
9
        },
10
    )
11
) 

You can use different partition key mappers for each asset in the PartitionedAssetTimetable.

1
from airflow.sdk import dag, PartitionedAssetTimetable, Asset, StartOfQuarterMapper, StartOfWeekMapper
2
3
@dag(
4
    schedule=PartitionedAssetTimetable(
5
        assets=(my_combined_asset_one & my_combined_asset_two),
6
        partition_mapper_config={
7
            my_combined_asset_one: StartOfQuarterMapper(),
8
            my_combined_asset_two: StartOfWeekMapper(),
9
        },
10
    )
11
) 

Composite partition keys

Composite partition keys are partition keys that are composed of multiple segments, separated by | delimiters. For example, the partition key Finance|2026-03-16T09:00:00|Revenue is a composite partition key with three segments: Finance, 2026-03-16T09:00:00, and Revenue.

You can use the ProductMapper partition key mapper to map composite keys segment by segment, applying one mapper per segment and then rejoining the mapped segments. For example, with the key Finance|2026-03-16T09:00:00, ProductMapper(IdentityMapper(), StartOfDayMapper()) produces Finance|2026-03-16.

1
from airflow.sdk import dag, task, PartitionedAssetTimetable, Asset, ProductMapper, IdentityMapper, StartOfDayMapper, AllowedKeyMapper
2
3
@dag(
4
    schedule=PartitionedAssetTimetable(
5
        assets=Asset("my_partitioned_asset"),
6
        partition_mapper_config={
7
            Asset("my_partitioned_asset"): ProductMapper(IdentityMapper(), StartOfDayMapper(), AllowedKeyMapper(["Revenue", "ARR"])),
8
        },
9
    )
10
) 
11
def my_composite_dag():
12
13
    @task
14
    def my_task(**context):
15
        partition_key = context["dag_run"].partition_key
16
        print(partition_key) # prints the partition key in the format `Finance|2026-03-16|Revenue`
17
18
    my_task()
19
20
21
my_composite_dag()

One-to-many: FanOutMapper

The FanOutMapper, introduced in Airflow 3.3, creates multiple downstream Dag runs based on one partitioned asset event.

The FanOutMapper uses an upstream_mapper to normalize the upstream key to its period start, a window instance that enumerates the elements in that period, and an optional downstream_mapper to format each element.

The available windows are HourWindow, DayWindow, WeekWindow, MonthWindow, QuarterWindow, and YearWindow. Each window provides a default downstream_mapper, except HourWindow and custom windows, for which you need to set a downstream_mapper explicitly.

For example, if your asset my_asset gets updated once a day and you’d like to create 24 Dag runs, each Dag run processing the data for one hour in the day, you can use the FanOutMapper with the upstream_mapper=StartOfDayMapper(), which normalizes the often irregular partition key (for example: 2026-06-22T03:19:23) to the start of the day (2026-06-22) and then use the DayWindow to fan out to create a set of 24 Dag runs, with hourly partition keys (2026-06-22T00, 2026-06-22T01, …, 2026-06-22T23).

1
from airflow.sdk import dag, PartitionedAssetTimetable, Asset, FanOutMapper, StartOfDayMapper, DayWindow
2
3
@dag(
4
    schedule=PartitionedAssetTimetable(
5
        assets=Asset("my_asset"),
6
        default_partition_mapper=FanOutMapper(
7
            upstream_mapper=StartOfDayMapper(),
8
            window=DayWindow(),
9
        ),
10
    )
11
)

Many-to-one: RollupMapper

The RollupMapper, introduced in Airflow 3.3, is the inverse of the FanOutMapper. It maps multiple upstream partition keys to one downstream partition key, creating a single downstream Dag run once all expected upstream keys have arrived. It takes an upstream_mapper to normalize each upstream key to the downstream grain and a window instance that declares the full set of upstream keys required for a given downstream key.

For example, if my_asset gets updated every hour and you want a downstream Dag to run as soon as all 24 partitions for a day have been updated, you can use the RollupMapper with the StartOfDayMapper as upstream_mapper and the DayWindow as the window.

1
from airflow.sdk import dag, PartitionedAssetTimetable, Asset, RollupMapper, StartOfDayMapper, DayWindow
2
3
@dag(
4
    schedule=PartitionedAssetTimetable(
5
        assets=Asset("my_partitioned_asset"),
6
        default_partition_mapper=RollupMapper(
7
            upstream_mapper=StartOfDayMapper(),
8
            window=DayWindow(),
9
        ),
10
    )
11
)

The Airflow UI displays how many and which of the partitions have arrived.

By default the downstream Dag waits for all keys in the window to arrive. This is the wait_policy=WaitForAll(). You can also set wait_policy=MinimumCount(n=5), which means the downstream Dag runs as soon as 5 partition keys for the window have arrived.

SegmentWindow

The RollupMapper also supports categorical rollups, where a fixed set of string segment keys roll up into one downstream partition key. Pair a FixedKeyMapper, which collapses every upstream key onto one fixed downstream key, with a SegmentWindow, which declares the set of segment keys that make up the downstream partition. The downstream Dag run is triggered once every segment key in the window has arrived. For example, in the code sample below, the Legal and Sales segment keys roll up into one legal_and_sales downstream Dag run.

1
from airflow.sdk import dag, PartitionedAssetTimetable, Asset, RollupMapper, FixedKeyMapper, SegmentWindow
2
3
@dag(
4
    schedule=PartitionedAssetTimetable(
5
        assets=Asset("my_partitioned_asset"),
6
        default_partition_mapper=RollupMapper(
7
            upstream_mapper=FixedKeyMapper("legal_and_sales"),
8
            window=SegmentWindow(["Legal", "Sales"]),
9
        ),
10
    )
11
)