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. Airflow 2.4 gave us datasets. 02 / 106
3. Airflow 3.0 renamed them assets. 03 / 106
4. An asset is just... 04 / 106
5. A name... Asset(name="daily_sales") 05 / 106
6. A URI... Asset(uri="s3://warehouse/daily_sales") 06 / 106
7. A contract... ingest_daily_sales → daily_sales 07 / 106
8. A contract between every Dag that writes it and every Dag that reads it... ingest_daily_sales → daily_sales → sales_summary 08 / 106
9. And it says exactly one thing. 09 / 106
10. "Something changed." 10 / 106
11. It knows when it happened. 11 / 106
12. Not which day it is about. 12 / 106
13. AIP-76: Asset Partitions Shipped in 3.2. Extended in 3.3. 13 / 106
14. One table. daily_sales ┌────┬────┬────┬────┬────┬────┬────┬────┬────┬────┬────┬────┐ │ │ │ │ │ │ │ │ │ │ │ │ │ └────┴────┴────┴────┴────┴────┴────┴────┴────┴────┴────┴────┘ 01 02 03 04 05 06 07 08 09 10 11 12 … 14 / 106
15. 365 days of data. 15 / 106
16. One bad day. daily_sales ┌────┬────┬────┬────┬────┬────┬────┬────┬────┬────┬────┬────┐ │ │ │ │ │ │▓▓▓▓│ │ │ │ │ │ │ └────┴────┴────┴────┴────┴────┴────┴────┴────┴────┴────┴────┘ 01 02 03 04 05 06 07 08 09 10 11 12 … 16 / 106
17. What do you re-run? 17 / 106
18. Everything. 18 / 106
19. Because the event only said: something changed. Your consumer woke up knowing which asset, which run, and when. Nothing else. 19 / 106
20. So make the event say more. 20 / 106
21. Put the day in the extra. 21 / 106
22. Like this. ON THE EVENT @task(outlets=[sales]) def ingest(*, outlet_events): outlet_events[sales].extra = {"day": "2026-03-10", "rows": 41_233} Set per emission. Describes this event. 22 / 106
23. Reading it back works. THIS IS SUPPORTED @task(inlets=[sales]) def report(*, inlet_events): events = inlet_events[sales] day = events[-1].extra["day"] 23 / 106
24. Something has to wait for the data. 24 / 106
25. The only question is what. 25 / 106
26. What is waiting? A TASK WAITING a worker slot a pool slot THE SCHEDULER WAITING a row held until the data lands a heartbeat a log stream a retry budget held until the data lands 26 / 106
27. A task can only wait after it exists. 27 / 106
28. The choice is two lines apart. ONE ACCESSOR @task(outlets=[sales]) def ingest(*, outlet_events): outlet_events[sales].extra = {"day": "2026-03-10"} outlet_events[sales].add_partitions("2026-03-10") Same object. One is a dict the scheduler never reads. One is what it schedules on. 28 / 106
29. When a run fired decided what it processed. 29 / 106
30. Data owns its own slices. 30 / 106
31. An asset is a collection of partitions. before with partitions ┌──────────────────┐ │ │ │ daily_sales │ │ │ └─────────▲────────┘ │ one schedule ┌────┬────┬────┬────┬────┐ │ 08 │ 09 │ 10 │ 11 │ 12 │ └────┴──▲─┴────┴────┴────┘ │ │ │ one slice 31 / 106
32. 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 32 / 106
33. 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. 33 / 106
34. 76 — assets have parts. That's the one this talk is about. 34 / 106
35. 01 / 04 Definitions Where the partition key comes from. DEFINITIONS MAPPINGS COMPLETENESS BACKFILLS ADOPTING IT 35 / 106
36. A partition key is a string. 36 / 106
37. Three shapes you will actually write. 2026-03-10T09:00:00 hourly, from a schedule tw a segment, decided at runtime tw|2026-03-10 composite, both at once Whatever writes it, both sides must agree on the format. 37 / 106
38. Not time versus not-time. THE USEFUL SPLIT That is the split everyone guesses. The one that matters is who decides the key, and when. 38 / 106
39. 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 39 / 106
40. Schedule = when to run. 40 / 106
41. Partition = what to touch. 41 / 106
42. Proof they are different things. NO SCHEDULE, STILL PARTITIONED @asset( uri="file://incoming/live-region.csv", schedule=PartitionedAtRuntime(), ) def live_region_stats(self, outlet_events): outlet_events[self].add_partitions("tw") PartitionedAtRuntime has no cadence at all — and still produces partitions. 42 / 106
43. Pre-determined: put it on the cadence. PRODUCER @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. 43 / 106
44. Same thing on a classic Dag. PRODUCER · OUTLETS 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. 44 / 106
45. Runtime: let the task decide. PRODUCER · RUNTIME @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. 45 / 106
46. One run can emit many partitions. RUNTIME FAN-OUT @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. 46 / 106
47. Cron cadence with a runtime key. PLUGIN ESCAPE HATCH Subclass CronTriggerTimetable, set partitioned_at_runtime = True. 06:00 fires │ ▼ data landed? │ ├── yes │ └── no ──→ add_partitions("2026-03-10") ──→ downstream run ──→ emit nothing ──→ no run 47 / 106
48. A partition key lives on a Dag run. 48 / 106
49. Every run knows which slice it owns. INSIDE THE TASK @task def summarize(dag_run=None): day = dag_run.partition_key # "2026-03-10" rows = load(f"s3://warehouse/sales/{day}/") write_summary(day, rows) 49 / 106
50. 02 / 04 Mappings How upstream slices match downstream runs. DEFINITIONS MAPPINGS COMPLETENESS BACKFILLS ADOPTING IT 50 / 106
51. So what triggers a partitioned run? 51 / 106
52. The consumer side. THREE ARGUMENTS with DAG( dag_id="clean_and_combine_player_stats", schedule=PartitionedAssetTimetable( assets=team_a & team_b & team_c, default_partition_mapper=StartOfHourMapper(), ), catchup=False, ): ... 52 / 106
53. No partition key, no trigger. An event without a partition_key will not fire a PartitionedAssetTimetable. 53 / 106
54. The simple case. upstream key 2026-03-10 IdentityMapper downstream key 2026-03-10 54 / 106
55. Identity mapping. Key in. Same key out. This is the default. 55 / 106
56. Seven keys in. The same key out. 2026-03-09 2026-03-10 StartOfWeek Mapper all become 2026-03-09 2026-03-11 ... 56 / 106
57. It does not wait for all seven. The first key to arrive already produces the weekly key — and fires the run. 57 / 106
58. 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. 58 / 106
59. The temporal family. IN: 2026-03-10T09:37:51 MAPPER OUTPUT StartOfHourMapper 2026-03-10T09 StartOfDayMapper 2026-03-10 StartOfWeekMapper week start StartOfMonthMapper month start StartOfQuarterMapper quarter start StartOfYearMapper 2026 59 / 106
60. Aligning three producers on the hour. THE SHIPPED EXAMPLE schedule=PartitionedAssetTimetable( assets=team_a_player_stats & team_b_player_stats & team_c_player_stats, default_partition_mapper=StartOfHourMapper(), ) Three feeds, three cron schedules, one aligned hourly run. 60 / 106
61. Different keys. Same hour. team_a 09:00:00 ┐ team_b 09:15:00 ├── team_c 09:30:00 ┘ StartOfHourMapper ──→ 2026-03-10T09 one run 61 / 106
62. Override one asset at a time. PARTITION_MAPPER_CONFIG 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 same key · one run 62 / 106
63. Composite keys, segment by segment. PRODUCTMAPPER default_partition_mapper=ProductMapper( IdentityMapper(), # region segment, untouched StartOfDayMapper(), # timestamp segment, to a day ) tw │ IdentityMapper │ ▼ tw | | 2026-03-10T09:00:00 │ StartOfDayMapper │ ▼ 2026-03-10 63 / 106
64. Chaining, and the trap in it. CHAINMAPPER ChainMapper( StartOfHourMapper(), StartOfDayMapper(input_format="%Y-%m-%dT%H"), ) 2026-03-10T09:37:51 │ └─ StartOfHourMapper ──→ 2026-03-10T09 │ └─ StartOfDayMapper ──→ 2026-03-10 input_format="%Y-%m-%dT%H" 64 / 106
65. Validate instead of transform. ALLOWEDKEYMAPPER default_partition_mapper=AllowedKeyMapper(["tw", "jp", "us"]) tw jp eu ──→ ──→ ──→ tw jp ✗ in the list · passes through not in the list · no downstream run 65 / 106
66. Collapse everything onto one key. tw jp us ┐ ├── ┘ FixedKeyMapper("all_countries") ──→ FIXEDKEYMAPPER all_countries Every upstream key becomes the same downstream key. 66 / 106
67. No match is not an error. It just doesn't run. 67 / 106
68. Airflow ships a Dag that never fires. ON PURPOSE with DAG( dag_id="player_odds_quality_check_wont_ever_to_trigger", schedule=PartitionedAssetTimetable( assets=(combined_stats & team_a_stats & Asset.ref(name="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. 68 / 106
69. Why it is silent. "No match" is a legitimate everyday state. A key that does not apply to this consumer must not raise — otherwise every unrelated partition in your deployment would throw. 69 / 106
70. 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. BEFORE YOU DEPLOY The one thing that does fail loudly is coming up in Part 3. 70 / 106
71. A trigger stops meaning "the Dag ran". It starts meaning "my slice is ready". 71 / 106
72. 03 / 04 Completeness When is a slice actually done? DEFINITIONS MAPPINGS COMPLETENESS BACKFILLS ADOPTING IT 72 / 106
73. Is the day done? 73 / 106
74. Something has to count to 24. 74 / 106
75. Let it be the scheduler. 75 / 106
76. Many upstream. One downstream. ROLLUPMAPPER default_partition_mapper=RollupMapper( upstream_mapper=StartOfHourMapper(), window=DayWindow(), ) 76 / 106
77. 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 77 / 106
78. 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 78 / 106
79. The scheduler holds the run. Nothing is occupying a worker while it waits. 79 / 106
80. This one fails loudly. RollupMapper( upstream_mapper=IdentityMapper(), window=DayWindow(), ) AT DAG PARSE # decodes to str # requires datetime TypeError at parse time — before the Dag is ever scheduled. 80 / 106
81. What if one of them never arrives? A THIRD ARGUMENT RollupMapper( upstream_mapper=StartOfHourMapper(), window=DayWindow(), wait_policy=MinimumCount(23), ) With the default, the run waits. Forever. 81 / 106
82. When reality is late. expected arrived tw ✓ jp ✓ WaitForAll() MinimumCount(2) MinimumCount(-1) held fires fires WAIT_POLICY us · — the default — at least two arrived — at most one missing 82 / 106
83. Now FixedKeyMapper makes sense. SEGMENT ROLLUP default_partition_mapper=RollupMapper( upstream_mapper=FixedKeyMapper("all_countries"), window=SegmentWindow(["tw", "jp", "us"]), ) Hold until every declared region arrives, then fire once. 83 / 106
84. One upstream. Many downstream. FANOUTMAPPER 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 84 / 106
85. Cap the fan-out on purpose. MAX_DOWNSTREAM_KEYS → Exceeded the runs are not queued, and an audit row is written: event="partition fan-out exceeded" Omitted → falls back to [scheduler] partition_mapper_max_downstream_keys, default 1000. 85 / 106
86. Which period, relative to the key. WINDOW.DIRECTION key = 2026-03-09 │ 03-03 ── 03-04 ── … ────── 03-08 ── 03-09 ── 03-10 ── … ────── 03-15 └───────────── BACKWARD ────────────┘ └──────── FORWARD ─────────┘ 86 / 106
87. Three shapes. That's the whole library. 1:1 • ─────────→ • Identity · StartOf* · Product N:1 •••• ──────→ • RollupMapper + window (holds) 1:N • ─────────→ •••• FanOutMapper + window (multiplies) 87 / 106
88. 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 88 / 106
89. 04 / 04 Backfills Materializing the past, and everything after the first green run. DEFINITIONS MAPPINGS COMPLETENESS BACKFILLS ADOPTING IT 89 / 106
90. Backfill = materialize the past. 90 / 106
91. Who can backfill? THE SPLIT THAT MATTERS TIMETABLE BACKFILL? CronPartitionTimetable — producer yes — the date range becomes a partitiondate range PartitionedAssetTimetable — no — listed in consumer DagNonPeriodicScheduleException 91 / 106
92. A consumer has no schedule to walk. 92 / 106
93. Re-drive it by re-emitting upstream. 93 / 106
94. Backfilling a producer. ORDINARY FLAGS 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. 94 / 106
95. Trigger exactly one partition. BY HAND 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. 95 / 106
96. What operators can see. next run assets · THE PAYOFF daily_sales_summary ✓ 21 received · 3 required, not yet arrived · 2026-03-10 hours 09, 11, 22 status: held, not started "Waiting on 3 of 24" is a fact you read, not a guess. 96 / 106
97. and then Adopting it You do not have to take all of this at once. DEFINITIONS MAPPINGS COMPLETENESS BACKFILLS ADOPTING IT 97 / 106
98. Decide two things first. Grain what is one slice? hour, day, region, region-day Key format the exact string, written down — both sides must agree BEFORE ANY CODE 98 / 106
99. You may not need any of it. If downstream reprocesses everything and that is fast enough, plain assets are the right answer. 99 / 106
100. Only two moves have an order. 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. 100 / 106
101. The rest is a menu. REACH FOR WHAT YOUR DATA NEEDS IF YOUR DATA… REACH FOR 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 always 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 101 / 106
102. Or none of them fit. PLUGIN ATTRIBUTES Custom PartitionMapper, custom Window, custom partition-aware timetable. Shipped examples: a namespace-stripping mapper, a business-day window, a cron timetable that defers its key to runtime. 102 / 106
103. extra describes the event. partition_key schedules the run. 103 / 106
104. The model is simple. The leverage is large. 104 / 106
105. Wei Lee Apache Airflow PMC Member Senior Software Engineer @ Astronomer wei-lee.me 105 / 106
106. Questions? Wei Lee · Astronomer wei-lee.me