Asset Partitions: Matching Workflow to the Right Data

Transcript

1. Asset Partitions: Matching Workflow to the Right Data Wei Lee Astronomer · wei-lee.me
2. Who here uses assets? 🙋 02 / 104
3. In Airflow Survey 2025... 03 / 104
4. The Annual Airflow Survey is Here! 04 / 104
5. Airflow 2.4 gave us datasets. 05 / 104
6. Airflow 3.0 calls them assets. 06 / 104
7. An asset is just... 07 / 104
8. A name... Asset(name="daily_sales") 08 / 104
9. A URI... Asset(uri="s3://warehouse/daily_sales") 09 / 104
10. A contract... ingest_daily_sales → daily_sales 10 / 104
11. A contract between every Dag that writes it and every Dag that reads it... ingest_daily_sales → daily_sales → sales_summary 11 / 104
12. And it says exactly one thing. 12 / 104
13. "Something changed." 13 / 104
14. It knows when it happened. 14 / 104
15. Not which part of the asset. 15 / 104
16. AIP-76: Asset Partitions Shipped in 3.2. Extended in 3.3. 16 / 104
17. Partitions touch the timetable, the scheduler and the task. 17 / 104
18. Slide deck 18 / 104
19. Assets you already have. 19 / 104
20. 365 days of data. daily_sales ┌────┬────┬────┬────┬────┬────┬────┬────┬────┬────┬────┬────┐ │ │ │ │ │ │ │ │ │ │ │ │ │ └────┴────┴────┴────┴────┴────┴────┴────┴────┴────┴────┴────┘ 01 02 03 04 05 06 07 08 09 10 11 12 … 20 / 104
21. One bad day. daily_sales ┌────┬────┬────┬────┬────┬────┬────┬────┬────┬────┬────┬────┐ │ │ │ │ │ │▓▓▓▓│ │ │ │ │ │ │ └────┴────┴────┴────┴────┴────┴────┴────┴────┴────┴────┴────┘ 01 02 03 04 05 06 07 08 09 10 11 12 … 21 / 104
22. Upstream fixed it. daily_sales ┌────┬────┬────┬────┬────┬────┬────┬────┬────┬────┬────┬────┐ │ │ │ │ │ │▓▓▓▓│ │ │ │ │ │ │ └────┴────┴────┴────┴────┴────┴────┴────┴────┴────┴────┴────┘ 01 02 03 04 05 06 07 08 09 10 11 12 … 22 / 104
23. Day 06 is good again. What gets recomputed? 23 / 104
24. Everything. 24 / 104
25. Because the event only said: something changed. 25 / 104
26. Where partitions pay off. 26 / 104
27. We only want to retry part of the data. 27 / 104
28. A report that waits for every partition to arrive. 28 / 104
29. One weekly model. Seven daily inference runs. 29 / 104
30. How do you slice your data today? 30 / 104
31. Put the information in the extra. 31 / 104
32. @task(outlets=[sales]) def ingest(*, outlet_events): outlet_events[sales].extra = {"day": "2026-03-10", "rows": 41_233} Set per emission. Describes this event. 32 / 104
33. The scheduler could not see any of it. That dict is a payload. Nothing schedules on it. 33 / 104
34. Something has to decide whether to run when partitions arrive. 34 / 104
35. The only question is what. 35 / 104
36. The workaround becomes a schedule. YOU WROTE THIS NOW YOU DECLARE IT Which slice? one run per slice Has it landed? held until ready How many arrived? the scheduler counts Go, or stop? you set the wait policy in a task · after the run started in the schedule · before the run exists 36 / 104
37. @task(outlets=[sales]) def ingest(*, outlet_events): # the scheduler never reads this outlet_events[sales].extra = {"day": "2026-03-10"} # the scheduler schedules on this outlet_events[sales].add_partitions("2026-03-10") Same object. One is a dict the scheduler never reads. One is what it schedules on. 37 / 104
38. An asset is a collection of partitions. before with partitions ┌──────────────────┐ │ │ │ daily_sales │ │ │ └─────────▲────────┘ │ one schedule ┌────┬────┬────┬────┬────┐ │ 08 │ 09 │ 10 │ 11 │ 12 │ └────┴──▲─┴────┴────┴────┘ │ │ │ one slice 38 / 104
39. 73 is the vision. AIP-73 · Expanded Data Awareness AIP-74 Introducing Data Assets AIP-75 New Asset-Centric Syntax Airflow 3.0 AIP-76 Asset Partitions Airflow 3.2 / 3.3 39 / 104
40. 74 and 75 got us here. 74 · Introducing Data Assets — datasets became assets, and the model was reworked underneath, explicitly to make room for partitions. 75 · New Asset-Centric Syntax — @asset, so a Dag can be written data-first instead of task-first. 40 / 104
41. 76 — assets have parts. That's the one this talk is about. 41 / 104
42. 01 / 04 Definitions Where the partition key comes from. DEFINITIONS MAPPINGS COMPLETENESS BACKFILLS ADOPTING IT 42 / 104
43. A partition key is a string. 43 / 104
44. 2026-03-10T09:00:00 hourly, from a schedule tw a segment, decided at runtime tw|2026-03-10 composite, both at once One Dag writes this string. Another Dag has to match it exactly. 44 / 104
45. Who decides the key, and when. PRE-DETERMINED RUNTIME Decided by the timetable the task Decided when before the run exists while the run executes Producer API CronPartitionTimetable PartitionedAtRuntime Use it when the key follows the cadence the key comes from the data 45 / 104
46. Pre-determined: put it on the cadence. @asset( uri="file://incoming/player-stats/team_b.csv", schedule=CronPartitionTimetable("15 * * * *", timezone="UTC"), ) def team_b_player_stats(): pass One partitioned event per run. 46 / 104
47. Same thing on a classic Dag. with DAG( dag_id="ingest_team_a_player_stats", schedule=CronPartitionTimetable("0 * * * *", timezone="UTC"), ): @task(outlets=[team_a_player_stats]) def ingest_team_a_stats(): pass Nothing in the task body changes. 47 / 104
48. Runtime: let the task decide. @asset( uri="file://incoming/live-region.csv", schedule=PartitionedAtRuntime(), ) def live_region_stats(self, outlet_events): outlet_events[self].add_partitions("tw") For a watermark read from the source, a late file, a key from the payload. 48 / 104
49. One run can emit many partitions. @asset( uri="file://incoming/multi-region.csv", schedule=PartitionedAtRuntime(), ) def multi_region_stats(self, outlet_events): outlet_events[self].add_partitions(["tw", "jp", "us"]) Each key gets its own event. Duplicate keys collapse to one. 49 / 104
50. Cron cadence with a runtime key. Subclass CronTriggerTimetable, set partitioned_at_runtime = True. 06:00 fires ▼ data landed? yes ──→ add_partitions("2026-03-10") ──→ downstream run no ──→ emit nothing ──→ no run 50 / 104
51. Schedule = when to run. 51 / 104
52. Partition = what to touch. 52 / 104
53. 02 / 04 Mappings How upstream slices match downstream runs. DEFINITIONS MAPPINGS COMPLETENESS BACKFILLS ADOPTING IT 53 / 104
54. So what triggers a partitioned run? 54 / 104
55. No partition key, no trigger. An event without a partition_key will not fire a PartitionedAssetTimetable. 55 / 104
56. The consumer side. with DAG( dag_id="clean_and_combine_player_stats", schedule=PartitionedAssetTimetable( assets=team_a & team_b & team_c, default_partition_mapper=StartOfHourMapper(), ), catchup=False, ): ... 56 / 104
57. Why a mapper at all? hourly_sales emits daily_summary wants 2026-03-10T09:00:00 │ │ a mapper translates ▼ 2026-03-10 The producer names its own slice. The consumer needs a different name for it. 57 / 104
58. Mapping 1:1 • ────→ • N:1 •••• ─→ • 1:N • ────→ •••• 58 / 104
59. The simple case. upstream key 2026-03-10 IdentityMapper downstream key 2026-03-10 59 / 104
60. Identity mapping. Key in. Same key out. This is the default. 60 / 104
61. Seven keys in. The same key out. 2026-03-09 2026-03-10 StartOfWeek Mapper all become 2026-03-09 2026-03-11 ... 61 / 104
62. It does not wait for all seven. The first key to arrive already produces the weekly key — and fires the run. 62 / 104
63. The rule. A mapper maps an upstream key to a downstream key. The run fires when every required upstream asset yields the same downstream key. 63 / 104
64. Time-based mappers. MAPPER 2026-03-10T09:37:51 BECOMES StartOfHourMapper 2026-03-10T09 StartOfDayMapper 2026-03-10 StartOfWeekMapper 2026-03-09 (W11) StartOfMonthMapper 2026-03 StartOfQuarterMapper 2026-Q1 StartOfYearMapper 2026 64 / 104
65. Different keys. Same hour. team_a 09:00:00 ┐ team_b 09:15:00 ├── team_c 09:30:00 ┘ StartOfHourMapper ──→ 2026-03-10T09 3 feeds, 3 cron schedules, 1 aligned hourly run. 65 / 104
66. Override one asset at a time. schedule=PartitionedAssetTimetable( assets=hourly_sales & daily_targets, default_partition_mapper=StartOfDayMapper(), partition_mapper_config={ daily_targets: IdentityMapper(), }, ) hourly_sales daily_targets 2026-03-10T09:00:00 2026-03-10 ──StartOfDay──→ ──Identity────→ 2026-03-10 2026-03-10 66 / 104
67. Composite keys, segment by segment. default_partition_mapper=ProductMapper( IdentityMapper(), # region segment, untouched StartOfDayMapper(), # timestamp segment, to a day ) ┌─────────────────────────────────────────────────────┐ tw | 2026-03-10T09:00:00 └─────────────────────────────────────────────────────┘ IdentityMapper StartOfDayMapper ▼ ▼ ┌─────────────────────────────────────────────────────┐ tw | 2026-03-10 └─────────────────────────────────────────────────────┘ 67 / 104
68. No match is not an error. It just doesn't run. 68 / 104
69. So check it yourself. 1 Write down each upstream's key format. 2 Write down what your mapper produces. 3 Confirm the strings can be equal. 69 / 104
70. A trigger no longer just means "the Dag ran". It also means "my slice is ready". 70 / 104
71. 03 / 04 Completeness When is a slice actually done? DEFINITIONS MAPPINGS COMPLETENESS BACKFILLS ADOPTING IT 71 / 104
72. Is the day done? 72 / 104
73. Many upstream. One downstream. default_partition_mapper=RollupMapper( upstream_mapper=StartOfHourMapper(), window=DayWindow(), ) 73 / 104
74. The window is an expected set. DayWindow · 24 required ┌────┬────┬────┬────┬────┬────┬────┬────┬────┬────┬────┬────┐ │✓ │✓ │✓ │✓ │✓ │✓ │✓ │✓ │✓ │· │✓ │· │ └────┴────┴────┴────┴────┴────┴────┴────┴────┴────┴────┴────┘ 00 01 02 03 04 05 06 07 08 09 10 11 … waiting on 2 of 24 — the run is held, not started 74 / 104
75. The shipped windows. WINDOW ONE DOWNSTREAM KEY NEEDS HourWindow sixty minutes DayWindow twenty-four hours WeekWindow seven days MonthWindow every day of the month QuarterWindow every day of the quarter YearWindow every day of the year SegmentWindow every key you declare 75 / 104
76. The scheduler holds the run. Nothing is occupying a worker while it waits. 76 / 104
77. And the run can see what fed it. @task def summarize(*, triggering_asset_events, dag_run=None): day = dag_run.partition_key # "2026-03-10" hours = [e.partition_key for e in triggering_asset_events[hourly_sales]] 77 / 104
78. This one fails loudly. RollupMapper( upstream_mapper=IdentityMapper(), window=DayWindow(), ) # decodes to str # requires datetime TypeError at parse time — before the Dag is ever scheduled. 78 / 104
79. What if one of them never arrives? RollupMapper( upstream_mapper=StartOfHourMapper(), window=DayWindow(), wait_policy=MinimumCount(23), ) With the default, the run waits. Forever. 79 / 104
80. When reality is late. expected arrived tw ✓ jp ✓ WaitForAll() MinimumCount(2) MinimumCount(-1) held fires fires us · — the default — at least two arrived — at most one missing 80 / 104
81. Roll up a fixed set of names. tw jp us ┐ ├── ┘ FixedKeyMapper("all_countries") ──→ all_countries default_partition_mapper=RollupMapper( upstream_mapper=FixedKeyMapper("all_countries"), window=SegmentWindow(["tw", "jp", "us"]), ) Hold until every declared region arrives, then fire once. 81 / 104
82. One upstream. Many downstream. default_partition_mapper=FanOutMapper( upstream_mapper=StartOfWeekMapper(), window=WeekWindow(), max_downstream_keys=7, ) weekly artifact 2026-03-09 ┌──→ ├──→ ├ ├──→ └──→ 2026-03-09 2026-03-10 … 2026-03-14 2026-03-15 → run → run → run → run 82 / 104
83. Cap the fan-out on purpose. → the runs are not queued. Omitted → falls back to [scheduler] Exceeded partition_mapper_max_downstream_keys, default 1000. 83 / 104
84. Which period, relative to the key. key = 2026-03-09 │ 03-03 ── 03-04 ── … ────── 03-08 ── 03-09 ── 03-10 ── … ────── 03-15 └───────────── BACKWARD ────────────┘ └──────── FORWARD ─────────┘ 84 / 104
85. You can write all three yourself. REGISTERED AS a timetable — where keys come from AirflowPlugin.timetables a partition mapper — how keys map AirflowPlugin.partition_mappers a window — what one period contains AirflowPlugin.windows 85 / 104
86. 04 / 04 Backfills Materializing the past, and everything after the first green run. DEFINITIONS MAPPINGS COMPLETENESS BACKFILLS ADOPTING IT 86 / 104
87. Backfill = materialize the past. 87 / 104
88. Who can backfill? TIMETABLE BACKFILL? CronPartitionTimetable — producer yes — the date range becomes a partitiondate range PartitionedAssetTimetable — no — listed in consumer DagNonPeriodicScheduleException 88 / 104
89. To backfill you need a time-based schedule. with DAG("ingest_hourly_sales", schedule=CronPartitionTimetable("0 * * * *")): ... # 02-18 00:00, 01:00, 02:00 ... a list of ticks with DAG("daily_sales_summary", schedule=PartitionedAssetTimetable(assets=hourly_sales)): ... # no ticks at all Backfill asks for a list of runs. Only one of these can answer. 89 / 104
90. So you backfill the producer instead. airflow backfill create \ --dag-id ingest_hourly_sales \ --from-date 2026-02-18 \ --to-date 2026-02-20 Airflow detects the Dag is partitioned and creates one run per partition. 90 / 104
91. Trigger exactly one partition. curl -X POST "http://$HOST/api/v2/dags/aggregate_regional_sales/dagRuns" \ -H "Content-Type: application/json" \ -d '{"partition_key": "tw|2026-03-10T09:00:00"}' The trigger window in the UI takes a partition key too. 91 / 104
92. and then Adopting it You do not have to take all of this at once. DEFINITIONS MAPPINGS COMPLETENESS BACKFILLS ADOPTING IT 92 / 104
93. Decide two things first. Grain what is one slice? hour, day, region, region-day Key format a string — mapper classes can transform it 93 / 104
94. You may not need any of it. If downstream reprocesses everything and that is fast enough, plain assets are the right answer. 94 / 104
95. Producer first, then consumer. First · a key on the producer — CronPartitionTimetable. Nothing downstream changes yet. Then · one consumer on PartitionedAssetTimetable. Default mapper, nothing to configure. The one real hazard: partitioned events do not trigger non-partition-aware Dags. Move a producer and its consumers together. 95 / 104
96. The rest is a menu. IF YOUR DATA… USE upstreams disagree on grain StartOf*Mapper · partition_mapper_config one slice is a region and a day ProductMapper a period is many upstream slices RollupMapper + a window …and one of them is sometimes late MinimumCount on that rollup one coarse input drives many fine runs FanOutMapper + max_downstream_keys a fixed set of names, not a time grain SegmentWindow + FixedKeyMapper 96 / 104
97. Or none of them fit. Custom PartitionMapper, custom Window, custom partition-aware timetable. 97 / 104
98. Gotcha: mappers that never agree schedule=PartitionedAssetTimetable( assets=combined_stats & team_a_stats & team_b_stats, partition_mapper_config={ combined_stats: StartOfYearMapper(), team_a_stats: StartOfHourMapper(), }, ) 2026 can never equal 2026-03-10T09. No error. No warning. No log line. 98 / 104
99. 3 takeaways. 99 / 104
100. 1 · three shapes 1:1 • ─────────→ • Identity · StartOf* · Product N:1 •••• ──────→ • RollupMapper + window (holds) 1:N • ─────────→ •••• FanOutMapper + window (multiplies) 100 / 104
101. 2 · the scheduler decides, not your task Whether a run exists. Which slice. When it may start. 101 / 104
102. 3 · the partition key travels on its own hourly_sales │ ▼ daily_summary │ ▼ weekly_rollup 2026-03-10T09 2026-03-10 2026-03-09 102 / 104
103. Wei Lee / 李唯 Apache Airflow PMC Member Senior Software Engineer @ Astronomer wei-lee.me 103 / 104
104. Questions? Wei Lee · Astronomer wei-lee.me