Segment-based partitioning #57695

Activity

1. dstandish

   added

   kind:metaHigh-level information important to the communityHigh-level information important to the community

   on Nov 1, 2025
2. dstandish

   added this to AIP-76 Asset Partitioningon Nov 1, 2025
3. dosubot

   added

   kind:featureFeature RequestsFeature Requests

   on Nov 1, 2025
4. phanikumv

   moved this to Backlog in AIP-76 Asset Partitioningon Dec 3, 2025
5. Lee-W

   moved this from Backlog to Todo in AIP-76 Asset Partitioningon Apr 23, 2026
6. Lee-W

   changed the title [-]Come up with plan for segment-based partitioning.[/-] [+]Segment-based partitioning[/+] on Apr 24, 2026
7. Lee-W commented on Apr 24, 2026

   

   Lee-W

   on Apr 24, 2026

   Member

   More actions

   from airflow.sdk import Asset, AllowedKeyMapper, PartitionAtRuntime, asset
   
   region_raw_stats = Asset(
       uri="file://incoming/player-stats/by-region.csv",
       name="region_raw_stats",
   )
   
   # Must be a module-level function so the timetable is serialisable:
   # the scheduler stores its dotted path and re-imports at scheduling time.
   # Lambdas, closures, and bound methods are rejected at construction.
   def list_active_regions() -> list[str]:
       ...
   
   @asset(
       uri="file://analytics/player-stats/regional-breakdown.csv",
       schedule=PartitionAtRuntime(
           assets=region_raw_stats,
           default_partition_mapper=AllowedKeyMapper(list_active_regions),
       ),
   )
   def regional_stats_breakdown():
       ...

   This could be how it works, but will affect #44146 implemented
8. anishgirianish commented on May 22, 2026

   

   anishgirianish

   on May 22, 2026

   Contributor

   More actions

   Hi, I am actively working on this. thanks
9. Lee-W

   moved this from Todo to In Progress in AIP-76 Asset Partitioningon May 29, 2026
10. Lee-W

    added this to the Airflow 3.3.0 milestone on Jun 5, 2026
11. Lee-W commented on Jun 11, 2026

    

    Lee-W

    on Jun 11, 2026

    Member

    More actions

    dynamic segement based partition can be achieved by

    # Producer emits per-segment partitions (observability) + one completion signal:
    @asset(schedule=PartitionAtRuntime())
    def regional_sales(self, outlet_events):
        for region in active_regions():            # "what's active" = business logic, on the worker
            process(region)
            outlet_events[self].add_partitions(region)
        outlet_events[batch_done].add(...)          # emit a completion event when the loop ends
    
    # Downstream triggers on the completion signal — plain asset scheduling:
    with DAG("global_report", schedule=batch_done):
        ...

    and since we don't want to allow user code in scheduler, we won't add a new timetable or partition mapper for that
12. Lee-W

    closed this as completedon Jun 11, 2026
13. github-project-automation

    moved this from In Progress to Done in AIP-76 Asset Partitioningon Jun 11, 2026