Dask Distributed Processing for LiDAR

A ProcessPoolExecutor is the right first step for parallel LiDAR processing, and it stops at the edge of one machine. When a statewide collection has twenty thousand tiles and each takes three minutes, a 32-core workstation needs more than a day; spreading the work over twenty machines brings it to about an hour. Dask’s distributed scheduler is the most direct way to make that jump from Python. It keeps the programming model of futures and map that you already use with a process pool, adds a scheduler that spreads tasks over any number of workers on any number of machines, and brings retries, memory limits, resource constraints and a live dashboard for free. This topic in the Batch Automation and Cloud Integration section covers how to structure PDAL work for Dask, how to size workers, and how to run the same code on a laptop, a VM fleet and Kubernetes.

# Prerequisites

• Python 3.10+ with dask[distributed] (2024.x or later), the PDAL bindings, and the same versions on every worker — in practice, one container image.
• Tiles in shared storage that every worker can read: S3 or another object store via GDAL’s /vsis3/, or a shared file system.
• A per-tile function that takes a path, runs PDAL, writes outputs to shared storage and returns a small summary. If it already works with ProcessPoolExecutor, it works with Dask.
• Memory measurements for the largest tiles, as in measuring peak memory of a PDAL pipeline, to set worker memory limits.
• Somewhere to run workers: local processes for development, VMs over SSH, a cloud VM group, or Kubernetes via the Dask Kubernetes operator.

# Core Workflow Architecture

1. List work. Build the list of tile keys from a tile index or bucket listing; sort largest first for better balance.
2. Start a cluster. LocalCluster for one machine; a remote scheduler plus workers for many. Workers use one thread each and a memory limit sized to the largest tile.
3. Submit one task per tile. client.map(process_tile, tiles, retries=2, pure=False) returns one future per tile.
4. Stream results. as_completed yields futures as they finish, for progress logging and immediate failure handling.
5. Collect failures. Exceptions stay on their futures; gather them into a retry list rather than letting one bad tile stop the batch.
6. Shut down and record. Close the client, write a manifest of outputs and failures, and keep the dashboard’s performance report if needed.

# Full Implementation

"""Process LiDAR tiles with Dask distributed: one task per tile, retries, progress, manifest."""
from __future__ import annotations

import json
import logging
import os
import time
from pathlib import Path

from dask.distributed import Client, LocalCluster, as_completed, performance_report

log = logging.getLogger("dask-lidar")

PIPELINE = {
    "pipeline": [
        {"type": "readers.las", "filename": None},
        {"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": None, "resolution": 1.0, "output_type": "idw",
         "window_size": 6, "data_type": "float32", "gdalopts": "COMPRESS=DEFLATE,TILED=YES"},
    ]
}


def process_tile(src: str, out_prefix: str) -> dict:
    """Runs on a worker. Returns a small summary, never point arrays."""
    os.environ.setdefault("OMP_NUM_THREADS", "1")
    import pdal
    spec = json.loads(json.dumps(PIPELINE))
    stem = Path(src).stem
    spec["pipeline"][0]["filename"] = src
    spec["pipeline"][-1]["filename"] = f"{out_prefix}/{stem}_dtm.tif"
    t0 = time.perf_counter()
    n = pdal.Pipeline(json.dumps(spec)).execute()
    return {"tile": stem, "ground_points": n, "seconds": round(time.perf_counter() - t0, 1),
            "output": spec["pipeline"][-1]["filename"]}


def run(tiles: list[str], out_prefix: str, scheduler: str | None = None) -> dict:
    if scheduler:
        client = Client(scheduler)
    else:
        cluster = LocalCluster(n_workers=os.cpu_count() // 2, threads_per_worker=1,
                               memory_limit="6GiB")
        client = Client(cluster)
    log.info("dashboard: %s", client.dashboard_link)

    done, failed = [], []
    with performance_report(filename="dask-report.html"):
        futures = client.map(process_tile, tiles, out_prefix=out_prefix,
                             retries=2, pure=False, key=[f"tile-{Path(t).stem}" for t in tiles])
        for i, fut in enumerate(as_completed(futures), 1):
            if fut.status == "error":
                failed.append({"key": fut.key, "error": repr(fut.exception())})
            else:
                done.append(fut.result())
            if i % 100 == 0 or i == len(futures):
                log.info("%d/%d finished, %d failed", i, len(futures), len(failed))
    client.close()
    manifest = {"done": done, "failed": failed}
    Path("manifest.json").write_text(json.dumps(manifest, indent=2))
    return manifest


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
    keys = [l.strip() for l in Path("tile_list.txt").read_text().splitlines() if l.strip()]
    run([f"/vsis3/lidar-in/{k}" for k in keys], "/vsis3/lidar-out/dtm",
        scheduler=os.environ.get("DASK_SCHEDULER_ADDRESS"))

# Code Breakdown

One task per tile. Tiles are independent and large, so each task is minutes of work — ideal granularity for Dask. Splitting a tile into smaller tasks would require moving points between workers, which is exactly what to avoid.

Summaries, not arrays. process_tile writes outputs directly to storage and returns a few numbers. Returning point arrays would ship gigabytes through the network and pile them up in the client.

threads_per_worker=1 and OMP_NUM_THREADS=1. PDAL work is CPU-bound and mostly single-threaded; one thread per worker process isolates memory and avoids oversubscription, the same reasoning as in threads vs processes for PDAL workloads.

memory_limit. Dask pauses a worker at 80 percent of its limit and restarts it at 95 percent by default. Setting the limit just above the measured peak of the largest tile turns a runaway tile into a restarted worker and a retried task instead of a machine-wide out-of-memory kill.

retries=2, pure=False. Transient failures — a network blip, a spot instance reclaimed — are retried automatically. pure=False ensures tasks are not deduplicated by argument hash, which would matter if the same tile were deliberately resubmitted.

Named keys. key=[f"tile-..."] makes the dashboard and error reports show tile names instead of opaque hashes.

performance_report. Writes an HTML snapshot of the dashboard — task stream, worker memory, profiles — for the run, which is invaluable when a batch was slower than expected.

# Parameter Reference Table

# Costs and When Not to Use Dask

Dask is not free of overhead. A scheduler has to run somewhere, workers need a shared view of the data, and the whole setup has more moving parts than a single process pool. For a batch that finishes overnight on one machine, that overhead buys little. Dask earns its place when at least one of three things is true: the batch no longer fits in an acceptable time on one machine; the work is elastic, so paying for forty machines for two hours is better than one machine for three days; or the operational features — retries, memory-based restarts, a dashboard, performance reports — would otherwise have to be built by hand.

It is also worth comparing with managed batch services. A queue-based system such as AWS Batch runs one container per tile with no scheduler for you to operate, at the cost of a slower start-up per job and less interactive visibility. Dask is the better fit for iterative work — tune, rerun, inspect — and for teams already working in Python notebooks; a managed queue is often better for fixed, repeated production runs.

# Validation and Integrity Checks

• Manifest completeness. Every input tile appears exactly once in done or failed.
• Outputs exist. For each done entry, the output object exists and is non-empty; list the output prefix and compare.
• Failure reasons. Group failures by exception type; a single dominant reason (for example out-of-memory) indicates a sizing problem rather than bad tiles.
• Idempotent reruns. Resubmitting failed tiles must overwrite partial outputs cleanly, which depends on deterministic output names; see making tile outputs idempotent.

m = json.loads(Path("manifest.json").read_text())
seen = {d["tile"] for d in m["done"]} | {f["key"].removeprefix("tile-") for f in m["failed"]}
assert seen == {Path(k).stem for k in keys}, "tiles missing from manifest"

# Deployment Options

The same client code runs against three kinds of cluster, and moving between them is mostly a matter of where the scheduler address comes from.

Local. LocalCluster starts a scheduler and workers as processes on the current machine. It is the right place to develop the per-tile function, measure memory, and run small batches — and it already gives you the dashboard, retries and memory limits.

Fixed VMs. Start dask scheduler on one host and dask worker tcp://scheduler:8786 --nworkers 16 --nthreads 1 --memory-limit 6GiB on each worker host, all from the same container image. This suits an on-premises rack or a set of long-lived cloud VMs, and needs nothing beyond SSH and Docker.

Kubernetes or managed services. The Dask Kubernetes operator creates a scheduler and worker pods from a custom resource, and supports adaptive scaling so worker pods appear and disappear with the task queue. Managed Dask offerings follow the same model. This is the most elastic option and the one that pairs naturally with spot or preemptible nodes, because Dask’s retries absorb lost workers.

Whichever you choose, the image is the unit of reproducibility: client, scheduler and workers must run the same Python, PDAL, GDAL and PROJ versions, which is exactly what pinning PDAL versions with conda-lock provides.

# Performance Tuning

• Largest tiles first. Sort the task list by size descending, so the longest tasks start early and the batch does not end with one worker grinding through a giant tile; see load balancing uneven LiDAR tiles.
• Resources for heavy tiles. Give workers a resource such as MEM=16 and submit big tiles with resources={"MEM": 16}, so at most one runs per worker slot sized for it.
• Data locality. Run workers in the same cloud region as the bucket; cross-region reads cost both time and egress fees.
• Scheduler overhead. Tens of thousands of multi-minute tasks are trivial for the scheduler. Millions of sub-second tasks are not — keep tasks coarse.
• Adaptive scaling. On cloud clusters, cluster.adapt(minimum=0, maximum=40) adds and removes workers with the queue length, so you pay only while tiles remain.

# Common Errors and Troubleshooting

KilledWorker. A task caused its worker to die repeatedly, usually by exceeding the memory limit. Raise the limit for big tiles with resources, or split the tile.

Tasks pile up on one worker. Work stealing moves queued tasks between workers, but tasks already running cannot move. Largest-first ordering reduces the effect.

ModuleNotFoundError on workers. Workers run a different environment from the client. Use one container image for client, scheduler and workers.

Silent GDAL credential failures. /vsis3/ reads fail on workers that lack credentials. Rely on instance roles or service accounts rather than local credential files, and test with a single task first.

Client runs out of memory. Results are too large. Return summaries only, and write data to storage from the worker.

# Frequently Asked Questions

When should I use Dask instead of ProcessPoolExecutor?

When one machine is not enough, or when you want retries, memory limits and a live dashboard without writing them yourself. The per-tile function is the same; only the executor changes.

How many Dask workers per machine?

Usually one worker per core with one thread each, limited by memory: divide available memory by the peak memory of your largest tiles, and use the smaller of that and the core count.

Can Dask split one large tile across workers?

Not usefully for PDAL neighbourhood filters, which need all points of a tile in one process. Split large tiles into smaller buffered tiles instead, and give each its own task.

Does Dask work with spot or preemptible instances?

Yes, and well. When a worker disappears, the scheduler reassigns its unfinished tasks to other workers, and with retries set each tile simply runs again. Because tile outputs are written with deterministic names, a rerun overwrites any partial output from the lost worker.

How do I watch a long batch?

Open the dashboard link printed by the client: the task stream shows every tile as it runs, the memory panel shows workers approaching their limits, and the progress bars show completion by task group. For unattended runs, log progress from the as_completed loop and keep the performance report.

How do I run the same code on Kubernetes?

Deploy a Dask cluster with the Dask Kubernetes operator or Helm chart using your PDAL image, set the scheduler address, and connect the client to it. The processing code does not change.

# Related

• Processing LiDAR Tiles with Dask Distributed — a step-by-step first deployment
• Dask vs ProcessPoolExecutor for PDAL — when the extra machinery pays off
• Tracking Tile Progress and Failures in Dask — manifests, retries and reports
• AWS Batch Processing — the managed-queue alternative
• PDAL Docker Containers — one image for client and workers