Python LiDAR

Dynamic Task Mapping for LiDAR Tiles

TL;DR: In Airflow 2.3+ (and Airflow 3), a @task that returns the list of tile keys can feed process_tile.partial(out_prefix=...).expand(key=tiles), which creates one mapped task instance per tile at run time. Cap parallelism with a pool or max_active_tis_per_dag, and add a downstream @task that receives the list of mapped results to write a manifest.

# Context and Motivation

This guide is part of Airflow DAG Orchestration. Before dynamic task mapping, fanning out over LiDAR tiles in Airflow meant either generating a DAG with thousands of static tasks — slow to parse and impossible to change without redeploying — or hiding the loop inside a single task, which threw away Airflow’s per-task retries, logs and visibility. Dynamic mapping solves both: the DAG stays small, the tile list is decided when the run starts, and each tile still becomes a first-class task instance with its own state, retries and log.

It is the natural shape for recurring LiDAR processing: a nightly run that picks up newly delivered tiles, or an on-demand run over a project’s tile index.

List, expand, reduce A task named list_tiles returns tile keys at run time. The process_tile task expands into one mapped instance per key, shown as a column of small boxes with indices 0, 1, 2 through n. A write_manifest task receives the list of all mapped results after they finish. list_tiles() process_tile [0] process_tile [1] process_tile [2] process_tile [n] write_manifest() ⋮

# Prerequisites and Assumptions

  • Airflow 2.3 or newer (dynamic task mapping), including Airflow 3.x.
  • Workers — Celery, Kubernetes executor or a container-running operator — with PDAL available, or tasks that launch PDAL in containers.
  • Tiles in object storage and an index or listing to enumerate them.

# Step-by-Step Implementation

# Step 1 — List tiles in a task

A @task returns a list of tile keys, for example from an S3 prefix or a tile index filtered to new deliveries.

# Step 2 — Define the per-tile task

A @task taking one key runs the PDAL pipeline and returns a small dict.

# Step 3 — Expand

process_tile.partial(out_prefix=...).expand(key=list_tiles()) — partial fixes shared arguments, expand maps over the list.

# Step 4 — Limit concurrency

Assign the task to a pool sized to the worker capacity, and set max_active_tis_per_dag on the mapped task so one run cannot occupy every slot.

# Step 5 — Reduce

A downstream @task receiving the mapped output as a list writes the manifest; set trigger_rule="all_done" so it runs even if some tiles failed.

# Complete Working Example

python
"""Airflow DAG: one mapped task per LiDAR tile, with a manifest at the end."""
from __future__ import annotations

import json
from datetime import datetime

import boto3
from airflow.decorators import dag, task
from airflow.utils.trigger_rule import TriggerRule

BUCKET_IN, BUCKET_OUT = "lidar-in", "lidar-out"


@dag(schedule="@daily", start_date=datetime(2026, 1, 1), catchup=False,
     max_active_runs=1, tags=["lidar"])
def lidar_dtm_daily():

    @task
    def list_tiles(prefix: str = "deliveries/2026/") -> list[str]:
        s3 = boto3.client("s3")
        keys = []
        for page in s3.get_paginator("list_objects_v2").paginate(Bucket=BUCKET_IN, Prefix=prefix):
            keys += [o["Key"] for o in page.get("Contents", []) if o["Key"].endswith(".laz")]
        return sorted(keys)

    @task(pool="pdal_slots", max_active_tis_per_dag=64, retries=2)
    def process_tile(key: str, out_prefix: str) -> dict:
        import pdal
        stem = key.rsplit("/", 1)[-1].removesuffix(".laz")
        spec = {"pipeline": [
            f"/vsis3/{BUCKET_IN}/{key}",
            {"type": "filters.range", "limits": "Classification![7:7],Classification![18:18]"},
            {"type": "filters.smrf", "slope": 0.15, "window": 18, "threshold": 0.5},
            {"type": "filters.range", "limits": "Classification[2:2]"},
            {"type": "writers.gdal", "filename": f"/vsis3/{BUCKET_OUT}/{out_prefix}/{stem}.tif",
             "resolution": 1.0, "output_type": "idw", "window_size": 6, "data_type": "float32"},
        ]}
        n = pdal.Pipeline(json.dumps(spec)).execute()
        return {"tile": stem, "ground_points": n}

    @task(trigger_rule=TriggerRule.ALL_DONE)
    def write_manifest(results: list[dict | None]) -> int:
        done = [r for r in results if r]
        boto3.client("s3").put_object(Bucket=BUCKET_OUT, Key="dtm/manifest.json",
                                      Body=json.dumps(done).encode())
        return len(done)

    results = process_tile.partial(out_prefix="dtm").expand(key=list_tiles())
    write_manifest(results)


lidar_dtm_daily()

Failed mapped instances contribute nothing to results, so the manifest lists successes only; failures are visible per instance in the grid view and can be cleared and retried individually, as described in retrying failed tiles in Airflow.

Two limits on fan-out A queue of 2,000 mapped tile tasks feeds into a pool of 64 slots matching worker capacity. At most 64 tile tasks run at any time; the rest wait as scheduled. The max_active_tis_per_dag setting on the task prevents one DAG run from taking slots other DAGs need. 2,000 mapped tasks scheduled, waiting pool ≤ 64 running pdal_slots = worker capacity max_active_tis_per_dag keeps one run from taking every slot

# Mapping Over Several Arguments

Sometimes each tile needs more than its key — a per-tile CRS override, a buffer geometry, or an output name that does not derive from the input. Two options exist. expand_kwargs maps over a list of dicts, so list_tiles can return [{"key": ..., "epsg": ...}, ...] and each instance receives its own keyword arguments. Alternatively, expand over several lists produces the cross product, which is almost never what you want for tiles — mapping key and epsg separately would pair every tile with every EPSG code. For tile work, build explicit records upstream and use expand_kwargs.

Keep the records small and serialisable. Put shapes and large per-tile metadata in object storage or a tile index, and pass only an identifier through XCom; the mapped task can then fetch what it needs at run time. This also keeps the rendered DAG grid readable, because each instance’s map index can be labelled with the tile name through the map_index_template option in Airflow 2.9 and later.

# Key Parameter Table

Setting Where Typical Purpose
.partial(...) mapped task shared args Fixed across instances
.expand(key=...) mapped task list from upstream One instance per element
pool task pdal_slots Limit by worker capacity
max_active_tis_per_dag task 32–128 Cap per DAG across runs
max_map_length airflow.cfg [core] default 1024 Raise for large tile lists
trigger_rule reduce task all_done Run even if some tiles fail

# Verification

  • Mapped count. The grid view shows as many mapped instances as tiles listed; compare with the listing size.
  • Manifest vs outputs. Manifest entries match the objects under the output prefix.
  • Concurrency. During a run, running instances never exceed the pool size.

# Gotchas and Edge Cases

max_map_length. Airflow limits how many instances a single expand may create (1024 by default). Statewide tile lists exceed it; raise the setting, or batch tiles into groups and map over groups.

Large XCom payloads. The tile list and every mapped result pass through XCom, stored in the metadata database by default. Keep results tiny, and for very large lists store the list in object storage and pass its key.

Scheduler load. Tens of thousands of mapped instances per run stress the scheduler and database. Grouping tiles — for example 10 per task — keeps instance counts manageable while retaining useful granularity.

Group tiles when lists are huge Two options for 20,000 tiles. Mapping one instance per tile creates 20,000 task instances, heavy for the scheduler and above the default map length. Mapping over groups of 10 creates 2,000 instances, each processing ten tiles sequentially with its own summary, retaining per-group retries and visibility. 1 tile per instance 20,000 instances over max_map_length, heavy DB 10 tiles per instance 2,000 instances manageable, still retryable pick the group size that keeps instances in the low thousands

Heavy PDAL in Airflow workers. Running PDAL directly in Celery workers ties Airflow’s workers to PDAL’s dependencies. KubernetesPodOperator or AWS Batch operators, mapped the same way, keep Airflow light and PDAL in its own image.

# Frequently Asked Questions

What is dynamic task mapping in Airflow?

A feature, introduced in Airflow 2.3, that creates one task instance per element of a list produced at run time. For LiDAR, the list is the set of tiles to process, so each tile becomes its own task with retries and logs.

How do I limit how many tiles process at once?

Put the mapped task in an Airflow pool sized to your worker capacity, and set max_active_tis_per_dag on the task. Instances beyond the limit wait in the scheduled state.

What if I have more tiles than the maximum map length?

Raise max_map_length in the core configuration, or group tiles into batches and map over the batches. Grouping also reduces load on the scheduler and metadata database.

Can the reduce task run if some tiles fail?

Yes. Set its trigger rule to all_done. Failed mapped instances contribute no value to the collected results, so the reduce task sees only successful outputs.

Related in Airflow DAG Orchestration for PDAL Workflows

↑ Airflow DAG Orchestration for PDAL Workflows overview --- title: "Airflow DAG Orchestration for PDAL Workflows" description: "Orchestrating multi-stage LiDAR processing with Apache Airflow —... Orchestrating PDAL Workflows with Airflow --- title: "Orchestrating PDAL Workflows with Airflow" description: "A worked Airflow DAG that runs a PDAL LiDAR pipeline per tile with... Retrying Failed Tiles in Airflow --- title: "Retrying Failed Tiles in Airflow" description: "Make per-tile LiDAR tasks in Airflow fail cleanly and retry sensibly: retries...