Dynamic Task Mapping

The Problem: Unknown Number of Items

# ❌ Without dynamic mapping — hardcoded file list
process_file_1 = PythonOperator(task_id="process_file_1", ...)
process_file_2 = PythonOperator(task_id="process_file_2", ...)
process_file_3 = PythonOperator(task_id="process_file_3", ...)
# What if there are 100 files? Or the count changes daily?

The Solution: .expand()

# ✅ With dynamic mapping — files are discovered at runtime
@task()
def get_files():
    """Discover files to process."""
    import boto3
    s3 = boto3.client('s3')
    response = s3.list_objects_v2(Bucket='data-bucket', Prefix='input/')
    return [obj['Key'] for obj in response['Contents']]

@task()
def process_file(file_key: str):
    """Process a single file. This runs once per file!"""
    print(f"Processing {file_key}")
    # Process the file...
    return {"file": file_key, "status": "done"}

@task()
def summarize(results: list):
    """Aggregate all results."""
    print(f"Processed {len(results)} files total")

# Magic: expand() creates one task instance per file
files = get_files()
results = process_file.expand(file_key=files)
summarize(results)

graph TD
    A["get_files()"] --> B["process_file[0]<br/>file_1.csv"]
    A --> C["process_file[1]<br/>file_2.csv"]
    A --> D["process_file[2]<br/>file_3.csv"]
    A --> E["process_file[N]<br/>file_N.csv"]
    B --> F["summarize()"]
    C --> F
    D --> F
    E --> F
    style A fill:#017cee,stroke:#015bb5,color:#fff
    style B fill:#4CAF50,stroke:#388E3C,color:#fff
    style C fill:#4CAF50,stroke:#388E3C,color:#fff
    style D fill:#4CAF50,stroke:#388E3C,color:#fff
    style E fill:#4CAF50,stroke:#388E3C,color:#fff
    style F fill:#9C27B0,stroke:#7B1FA2,color:#fff

This content is reserved for Premium Members.