Running a PDAL Pipeline in Streaming Mode

TL;DR: Build a chain of per-point stages, guard on pipeline.streamable, then call pipeline.execute_streaming(chunk_size=100_000) — or pdal pipeline --stream on the command line — and confirm with ru_maxrss that peak memory sits near the chunk working set instead of tracking the input file size.

# Context and Motivation

This recipe is part of Streaming Mode Execution in PDAL, which explains the pull-loop execution model and which stages participate in it. Here the focus is narrower: taking a pipeline you already have and actually running it in streaming mode, with enough instrumentation that you can tell whether it worked.

The motivation is nearly always a memory ceiling. A worker with 4 GB of RAM cannot read a 6 GB tile conventionally, no matter how simple the filtering is, because standard execution materialises the whole cloud before the first filter runs. Retiling the input is one answer and a laborious one. Streaming is the other, and for a chain of range filters and a reprojection it takes about four lines of change. The catch is that nothing about the ordinary API tells you which mode you got, so the recipe below spends as much effort on verification as on execution.

# Prerequisites and Assumptions

The example assumes a tile in a projected metric CRS. Nothing about streaming requires that, but the reprojection stage in the chain does need to know what it is transforming from, which the spatial reprojection guide covers in detail.

# Step-by-Step Implementation

# Step 1 — Build the chain from per-point stages only

Streaming is a property of the whole chain, so the design work happens before any execution call. Keep classification, sorting and neighbourhood filters out of this pipeline.

{
  "pipeline": [
    {"type": "readers.las", "filename": "tile_0431.laz"},
    {"type": "filters.range", "limits": "Z[-30:5000]"},
    {"type": "filters.range", "limits": "Classification![7:7]"},
    {"type": "filters.reprojection", "out_srs": "EPSG:6318"},
    {"type": "writers.las", "filename": "tile_0431_clean.laz", "compression": "laszip", "forward": "all"}
  ]
}

# Step 2 — Ask the pipeline whether it streams

import pdal

pipeline = pdal.Pipeline(spec)
if not pipeline.streamable:
    raise RuntimeError("chain contains a blocking stage — cannot stream")

This costs nothing and converts the failure mode from “used 30 GB in production” into a message at startup.

# Step 3 — Choose a chunk size deliberately

The default of 10,000 points is safe but leaves throughput on the table. A hundred thousand is the value worth writing down: about 35 MB of working set on a typical 40-byte point layout, and close to the throughput ceiling.

# Step 4 — Execute and capture the count

count = pipeline.execute_streaming(chunk_size=100_000)

The returned integer is the number of points the writer accepted. There is no pipeline.arrays afterwards, because no complete array was ever assembled.

# Step 5 — On the command line, use --stream

pdal pipeline clean.json --stream --verbose 4

If any stage blocks, this exits non-zero and names it. There is no fallback, which is exactly what you want in a CI check.

# Complete Working Example

Save as stream_run.py. It runs the pipeline, records peak memory, and cross-checks the streaming result against a conventional run on a capped subset of the same file.

"""Run a PDAL pipeline in streaming mode and prove the memory claim."""
from __future__ import annotations

import argparse
import json
import logging
import resource
import time
from pathlib import Path

import pdal

LOG = logging.getLogger("stream_run")


def spec(src: Path, dst: Path, out_srs: str, count: int | None = None) -> str:
    reader = {"type": "readers.las", "filename": str(src)}
    if count is not None:
        reader["count"] = count
    return json.dumps({
        "pipeline": [
            reader,
            {"type": "filters.range", "limits": "Z[-30:5000]"},
            {"type": "filters.range", "limits": "Classification![7:7]"},
            {"type": "filters.reprojection", "out_srs": out_srs},
            {
                "type": "writers.las",
                "filename": str(dst),
                "compression": "laszip",
                "minor_version": 4,
                "dataformat_id": 6,
                "forward": "all",
            },
        ]
    })


def peak_rss_mb() -> float:
    return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024.0


def stream(src: Path, dst: Path, out_srs: str, chunk: int) -> dict:
    pipeline = pdal.Pipeline(spec(src, dst, out_srs))
    if not pipeline.streamable:
        raise RuntimeError("pipeline is not streamable; inspect the stage list")

    started = time.perf_counter()
    written = pipeline.execute_streaming(chunk_size=chunk)
    elapsed = time.perf_counter() - started

    return {
        "mode": "streaming",
        "chunk_size": chunk,
        "points": written,
        "seconds": round(elapsed, 2),
        "peak_rss_mb": round(peak_rss_mb(), 1),
    }


def equivalence_check(src: Path, out_srs: str, sample: int = 200_000) -> None:
    """Run both modes over the same capped subset and compare counts."""
    plain = pdal.Pipeline(spec(src, Path("/tmp/_plain.laz"), out_srs, count=sample))
    plain_n = plain.execute()

    streamed = pdal.Pipeline(spec(src, Path("/tmp/_stream.laz"), out_srs, count=sample))
    streamed_n = streamed.execute_streaming(chunk_size=25_000)

    if plain_n != streamed_n:
        raise AssertionError(
            f"mode mismatch: standard kept {plain_n}, streaming kept {streamed_n}"
        )
    LOG.info("equivalence check passed on %d sampled points", sample)


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("src", type=Path)
    ap.add_argument("dst", type=Path)
    ap.add_argument("--out-srs", default="EPSG:6318")
    ap.add_argument("--chunk", type=int, default=100_000)
    ap.add_argument("--check", action="store_true", help="cross-check against standard mode first")
    args = ap.parse_args()

    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

    if args.check:
        equivalence_check(args.src, args.out_srs)

    result = stream(args.src, args.dst, args.out_srs, args.chunk)
    LOG.info("done: %s", json.dumps(result))
    print(json.dumps(result, indent=2))


if __name__ == "__main__":
    main()

Running it against a 2.1 GB tile on a worker with a 1 GB memory limit succeeds, which is the whole point:

python stream_run.py tile_0431.laz tile_0431_clean.laz --chunk 100000 --check

# Key Parameter Table

# Verification

Three assertions turn “it ran” into “it streamed and produced the right answer”.

Peak memory is flat across input sizes. Run the same command against a 200 MB tile and a 2 GB tile. Peak RSS should differ by a few megabytes, not by an order of magnitude.

/usr/bin/time -v python stream_run.py small.laz small_out.laz 2>&1 | grep "Maximum resident"
/usr/bin/time -v python stream_run.py big.laz big_out.laz 2>&1 | grep "Maximum resident"

Counts match standard mode. That is what --check does above, on a capped subset so the comparison is affordable.

The output header is complete. pdal info tile_0431_clean.laz --metadata should report the new CRS, a point count equal to the returned integer, and a bounding box consistent with the reprojected extent.

# Gotchas and Edge Cases

Checking streamable and then calling execute(). The guard passes, the run is conventional, and nothing warns you. Grep your own code for execute( in any function that also mentions streaming.

A chunk size larger than the file. Perfectly legal and completely pointless: one chunk means one buffer holding everything, which is standard mode with extra steps. On small tiles the memory graph will look wrong for exactly this reason.

writers.gdal in the chain. Rasterization accumulates cells over the whole input, so the pipeline will not stream. Split it into a streaming point-domain pass and a second rasterizing pass — the approach described in splitting a blocking pipeline into two passes.

Reading from S3 without tuning GDAL. Streaming from an object store works well, but the default virtual-filesystem settings issue a directory listing on every open. Set GDAL_DISABLE_READDIR_ON_OPEN=EMPTY_DIR before the run — the S3 and cloud storage I/O guide covers the rest of the environment.

# Frequently Asked Questions

Why does execute_streaming return a number instead of arrays?

Because no complete array ever exists. Streaming keeps one chunk in memory at a time, so there is nothing to hand back except the total the writer accepted. Code that needs pipeline.arrays needs standard execution and the memory that comes with it.

Does the --stream flag fall back to standard mode?

No. It fails and names the blocking stage. That refusal is the useful behaviour — a silent fallback is how a job ends up needing thirty gigabytes in production having passed on a small tile in testing.

How much memory should a streaming run actually use?

Roughly chunk_size × bytes-per-point × live buffers, plus a few tens of megabytes for PDAL, GDAL and PROJ themselves. At 100,000 points and a 40-byte layout that is about 35 MB — and it should not move when the input grows.

Can I stream directly from S3?

Yes, and it is one of the better reasons to stream. A LAZ object read through the GDAL virtual filesystem arrives in ranges, and a streaming pipeline consumes them as they land, so neither the object nor the decoded cloud is ever fully resident.

# Related

• Streaming Mode Execution in PDAL — the parent guide to the execution model
• Which PDAL Filters Break Streaming Mode — the capability rules behind streamable
• Splitting a Blocking Pipeline into Two Passes — what to do when one stage cannot stream
• Memory Management in PDAL Pipelines — the buffer model this is an alternative to
• PDAL Pipeline Architecture and Execution — the section overview