Orchestrating Tiling Jobs with Dask

This page runs a shard-per-task tiling job with Dask — building a task graph from the shard list, sizing workers by memory rather than by cores, retrying transient failures without re-running successful shards, reporting progress, and writing a manifest as results arrive so a crashed run resumes instead of restarting, for a city tiled from EPSG:25832 into EPSG:4978.

Why you hit this

A city tiling job is a few thousand independent tasks with wildly uneven cost, each needing a gigabyte or two of memory, producing files and a manifest entry. A for loop takes two days; multiprocessing.Pool runs it in two hours and loses everything when one shard raises; a shell script with xargs -P works until a worker is killed by the OOM killer and nobody notices which shards are missing. Dask handles the shape of this problem — heterogeneous tasks, memory pressure, partial failure, progress — without becoming a distributed-systems project. The shard grid it consumes comes from choosing shard sizes for city-scale tiling.

Prerequisites

  • Python 3.10+ with dask[distributed]>=2024.1, plus the tiling code itself and its dependencies available in the worker environment.
  • A shard list with an estimate of each shard’s cost — feature count is a good proxy.
  • Shared storage both the scheduler and the workers can write: a local path on a single machine, or object storage for a cluster.

Step-by-Step

1. Make the unit of work a pure function

python
import json
import os
import time
from dataclasses import dataclass, asdict
from pathlib import Path

@dataclass(frozen=True)
class ShardTask:
    quadkey: str
    source_uri: str
    out_dir: str
    feature_count: int
    params_hash: str

@dataclass(frozen=True)
class ShardResult:
    quadkey: str
    ok: bool
    triangles: int = 0
    bytes: int = 0
    seconds: float = 0.0
    worker: str = ""
    error: str = ""

def tile_shard(task: ShardTask) -> ShardResult:
    """Pure with respect to the process: reads its inputs, writes its outputs, returns a record."""
    t0 = time.perf_counter()
    out = Path(task.out_dir) / task.quadkey
    try:
        out.mkdir(parents=True, exist_ok=True)
        from twin.tiler import build_shard          # imported inside: workers import lazily
        stats = build_shard(task.source_uri, task.quadkey, out)
        return ShardResult(task.quadkey, True, stats["triangles"], stats["bytes"],
                           round(time.perf_counter() - t0, 2), os.uname().nodename)
    except Exception as exc:                        # a failed shard must not kill the run
        return ShardResult(task.quadkey, False, seconds=round(time.perf_counter() - t0, 2),
                           worker=os.uname().nodename, error=f"{type(exc).__name__}: {exc}")

Returning a result object rather than raising is the decision that shapes everything else. A task that raises makes Dask retry it, mark the graph as failed and lose the partial information; a task that returns a record lets the run continue, records why a shard failed, and leaves the retry decision to the orchestrator where it belongs. Genuinely transient failures — a dropped object-store connection — are worth retrying inside the task, and everything else is worth reporting.

Importing the tiler inside the function keeps the module importable on the scheduler, which does not need the heavy dependencies, and avoids serialising the whole library into the task graph.

2. Size the cluster by memory, not by cores

python
import psutil
from dask.distributed import Client, LocalCluster

def size_cluster(memory_per_task_gb=2.0, reserve_gb=4.0, max_workers=None):
    total_gb = psutil.virtual_memory().total / 1e9
    cores = os.cpu_count() or 4
    by_memory = max(1, int((total_gb - reserve_gb) // memory_per_task_gb))
    workers = min(cores, by_memory, max_workers or cores)
    return {"workers": workers, "threads_per_worker": 1,
            "memory_limit": f"{(total_gb - reserve_gb) / workers:.1f}GB",
            "bound_by": "memory" if by_memory < cores else "cores"}

spec = size_cluster()
print(spec)
cluster = LocalCluster(n_workers=spec["workers"],
                       threads_per_worker=spec["threads_per_worker"],
                       memory_limit=spec["memory_limit"],
                       processes=True)
client = Client(cluster)
print(client.dashboard_link)

Tiling tasks are memory-bound, not CPU-bound: a shard holding three hundred buildings with textures peaks at a couple of gigabytes, so a 16-core machine with 32 GB runs twelve workers, not sixteen. Setting memory_limit per worker lets Dask spill and, if a worker still exceeds it, restart that worker rather than let the kernel kill the process — which is the difference between losing one shard and losing the run.

One thread per worker is deliberate. The tiling code holds the GIL in places and uses native libraries with their own threading, so multiple threads per worker produce contention rather than throughput.

Worker count from memory rather than cores A table of machines. A sixteen core machine with thirty-two gigabytes and two gigabytes per task runs fourteen workers bound by memory. The same cores with sixty-four gigabytes run sixteen, bound by cores. A thirty-two core machine with sixty-four gigabytes runs thirty, bound by memory. Setting workers to the core count on a memory-bound machine invites the out-of-memory killer. machineper-task memory workersbound by 16 cores, 32 GB2 GB14memory 16 cores, 64 GB2 GB16cores 32 cores, 64 GB2 GB30memory Reserve a few gigabytes for the scheduler, the OS and the page cache before dividing.
The per-task memory figure comes from a pilot run; everything else follows from it arithmetically.

3. Submit the graph, largest shards first

python
from dask.distributed import as_completed

def submit_shards(client, tasks, retries=2, priority_by_cost=True):
    ordered = sorted(tasks, key=lambda t: -t.feature_count) if priority_by_cost else list(tasks)
    futures = []
    for i, task in enumerate(ordered):
        futures.append(client.submit(
            tile_shard, task,
            key=f"shard-{task.quadkey}-{task.params_hash[:8]}",   # stable, deduplicating key
            retries=retries,
            priority=len(ordered) - i,                            # bigger shards start earlier
        ))
    return futures

tasks = [ShardTask(**row) for row in json.loads(Path("build/shard_plan.json").read_text())]
futures = submit_shards(client, tasks)
print(f"{len(futures):,} shard tasks submitted")

Submitting the expensive shards first is the scheduling detail that shortens the whole run. With uniform submission order, the last task to start might be the 400-building historic centre shard, and every worker waits for it; with cost-descending order that shard starts in the first wave and the small ones fill the tail. On a city job the difference is routinely 20–30% of wall clock.

The key matters too: a deterministic key that includes the parameters’ hash means resubmitting the same work is deduplicated by Dask, and a key that changes when the parameters change avoids reusing a result computed with different settings.

What crosses the wire for one shard The scheduler sends a small task record holding the quadkey, the source URI, the output directory and a parameter hash. The worker reads the features for that shard from shared storage, writes the tile files there, and returns a small result record with counts, timing and any error. No geometry travels through the scheduler in either direction. schedulerholds the graph workerone shard at a time shared storage:features in shared storage:tiles out task: ~200 B result: ~200 B Passing geometry as a task argument is the most common way to make a Dask job slow.
Small tasks in, small records out, and the bulk data moves directly between the worker and storage.

4. Write the manifest as results arrive

python
def run_with_manifest(client, futures, manifest_path, flush_every=25):
    manifest = {}
    path = Path(manifest_path)
    if path.exists():
        manifest = json.loads(path.read_text())

    done = failed = 0
    pending_writes = 0
    for future, result in as_completed(futures, with_results=True, raise_errors=False):
        if isinstance(result, ShardResult):
            manifest[result.quadkey] = asdict(result)
            done += result.ok
            failed += not result.ok
        else:                                    # a task that died despite the try/except
            manifest[str(future.key)] = {"ok": False, "error": repr(result)}
            failed += 1
        pending_writes += 1
        if pending_writes >= flush_every:
            path.write_text(json.dumps(manifest, indent=1))
            pending_writes = 0
        if (done + failed) % 100 == 0:
            print(f"{done + failed:>6,}/{len(futures):,}  ok {done:,}  failed {failed:,}")
    path.write_text(json.dumps(manifest, indent=1))
    return manifest, done, failed

manifest, done, failed = run_with_manifest(client, futures, "build/manifest.json")
print(f"finished: {done:,} ok, {failed:,} failed")

Flushing the manifest every twenty-five results is what makes a long run survivable. A four-hour job that writes its manifest at the end loses everything to a scheduler restart; one that flushes incrementally loses at most twenty-five shards’ bookkeeping, and the resume logic in resuming failed tiling runs from checkpoints picks up from there.

as_completed with raise_errors=False is the other half: results arrive as they finish rather than in submission order, and a task that died in a way the function could not catch — a worker killed mid-execution — surfaces as an exception object instead of ending the loop.

5. Watch for the failure modes Dask makes visible

python
def cluster_health(client):
    info = client.scheduler_info()
    workers = info["workers"]
    return {
        "workers": len(workers),
        "threads": sum(w["nthreads"] for w in workers.values()),
        "memory_gb": round(sum(w["memory_limit"] for w in workers.values()) / 1e9, 1),
        "tasks_processing": sum(len(w.get("processing", {})) for w in workers.values()),
        "spilled_gb": round(sum(w["metrics"].get("spilled_bytes", {}).get("disk", 0)
                                for w in workers.values()) / 1e9, 2),
        "restart_counts": {addr: w["metrics"].get("event_loop_interval", 0) for addr, w in
                           list(workers.items())[:3]},
    }

print(cluster_health(client))

Three numbers from that snapshot tell you whether the run is healthy. Spilled bytes above zero means workers are exceeding their memory limit and paging to disk, which is ten to a hundred times slower than the work itself — the fix is fewer workers, not more. A task-processing count well below the worker count means the scheduler is starved, usually because the task list is exhausted and a few long tasks are finishing. And repeated worker restarts mean the per-task memory estimate is too low.

Worker occupancy with and without cost-ordered submission Two timelines of eight workers. With shards submitted in arbitrary order, the largest shard starts late and seven workers idle while it finishes, leaving a long tail. With shards submitted largest first, the expensive shards run in the first wave and the small ones fill the gaps, so the run ends when the work ends. arbitrary long tail largest first time — the green line is where the cost-ordered run finishes, the red where the other does
The tail is set by the largest task, so the only way to shorten a run is to start the largest tasks first.

6. Move to a real cluster without changing the code

python
def make_client(mode="local", **kw):
    if mode == "local":
        spec = size_cluster(**kw)
        return Client(LocalCluster(n_workers=spec["workers"], threads_per_worker=1,
                                   memory_limit=spec["memory_limit"], processes=True))
    if mode == "ssh":
        from dask.distributed import SSHCluster
        return Client(SSHCluster(hosts=kw["hosts"],
                                 worker_options={"n_workers": kw.get("per_host", 8),
                                                 "nthreads": 1,
                                                 "memory_limit": kw.get("memory_limit", "3GB")}))
    if mode == "kubernetes":
        from dask_kubernetes.operator import KubeCluster
        cluster = KubeCluster(name="twin-tiler", image=kw["image"],
                              n_workers=kw.get("workers", 32),
                              resources={"requests": {"memory": "3Gi", "cpu": "1"},
                                         "limits": {"memory": "4Gi", "cpu": "1"}})
        return Client(cluster)
    raise ValueError(mode)

# the task function, the submission and the manifest logic are identical in all three
client = make_client("local")

The portability is the reason to use Dask rather than multiprocessing for this. The same tile_shard, the same submission and the same manifest run on a laptop, on four machines over SSH and on Kubernetes; only the client construction changes. What does not travel for free is the environment — the workers need the tiling code and its native dependencies, which is what the container image is for — and shared storage, since a worker on another machine cannot write to a local path.

Expected Output & Verification

text
{'workers': 14, 'threads_per_worker': 1, 'memory_limit': '2.0GB', 'bound_by': 'memory'}
http://127.0.0.1:8787/status
1,742 shard tasks submitted
   100/1,742  ok 98  failed 2
   200/1,742  ok 197  failed 3
 …
 1,742/1,742  ok 1,736  failed 6
finished: 1,736 ok, 6 failed
{'workers': 14, 'threads': 14, 'memory_gb': 28.0, 'tasks_processing': 14, 'spilled_gb': 0.0}

Verify completeness against the plan, which is the check that catches the silent partial build:

python
def verify_run(plan_path, manifest_path, out_dir):
    plan = {row["quadkey"] for row in json.loads(Path(plan_path).read_text())}
    manifest = json.loads(Path(manifest_path).read_text())
    ok = {k for k, v in manifest.items() if v.get("ok")}
    failed = {k: v.get("error", "") for k, v in manifest.items() if not v.get("ok")}
    missing = plan - set(manifest)
    on_disk = {p.parent.name for p in Path(out_dir).glob("*/tileset.json")}
    return {
        "planned": len(plan), "ok": len(ok), "failed": len(failed), "never_ran": len(missing),
        "on_disk": len(on_disk),
        "ok_without_output": sorted(ok - on_disk)[:5],
        "output_without_ok": sorted(on_disk - ok)[:5],
        "error_kinds": sorted({e.split(":")[0] for e in failed.values()}),
    }

result = verify_run("build/shard_plan.json", "build/manifest.json", "build/tiles/city/shards")
print(result)
assert not result["ok_without_output"], "manifest claims shards that have no output"
assert result["never_ran"] == 0, f"{result['never_ran']} planned shards never ran"

Three discrepancies matter and each has a distinct cause. A shard marked ok with no output on disk means the task wrote to a path the verification does not read — usually a worker with a different working directory. Output with no ok entry means a manifest flush was lost, which is harmless but should reconcile. And a planned shard that never ran means a future was never submitted or its result never arrived, which is the case that produces a hole in the city.

Then look at the error kinds rather than the count: six failures all reporting InvalidGeometry is one data problem in six shards, while six different exception types is a pipeline that is unwell.

Performance Notes

  • Per-task overhead in Dask is a few milliseconds, negligible against a shard that takes seconds. It is not negligible for tasks under about 50 ms, so do not make the unit of work a single building.
  • Keep task arguments small. A ShardTask of a few hundred bytes serialises instantly; passing a GeoDataFrame as an argument serialises it to every worker and is the most common way to make a Dask job slower than a loop.
  • Read inputs inside the task, from shared storage. Workers should pull the features for their own shard rather than receive them.
  • Watch the spill metric, not the CPU graph. A cluster at 100% CPU with spilling is doing less work than one at 70% without.
  • Cap concurrency against object storage. Thirty workers each opening eight connections will hit rate limits; a semaphore or a smaller worker count is cheaper than the retries.

Common Errors

Workers die with no error and the run stalls. The kernel’s OOM killer, because memory_limit was unset or larger than the machine allows per worker. Set it explicitly, as in step 2, so Dask restarts the worker cleanly instead.

KilledWorker for one particular shard. That shard genuinely needs more memory than a worker has. Give it its own submission with a larger resource requirement, or split it a level deeper.

The dashboard shows all workers idle with tasks pending. The task graph depends on a large object being transferred, or the scheduler is blocked serialising results. Return small records from tasks, never geometry.

Every task re-runs on a resubmitted graph. The keys were non-deterministic, so Dask treats them as new work. Use a stable key including the parameter hash.

Results arrive but the manifest is empty. as_completed was iterated without with_results=True, so the loop yielded futures whose results were never collected.

Frequently Asked Questions

Why Dask rather than Airflow or a queue?

Different granularity. Airflow orchestrates the pipeline — classify, mesh, tile, publish — as a handful of long steps; Dask parallelises the thousands of tasks inside one of those steps. Most twin pipelines end up with both, with Airflow calling a script that builds a Dask graph.

Is a task graph better than a process pool here?

For a flat map over shards, a pool is adequate and simpler. Dask earns its place through memory-aware scheduling, retries with stable keys, the live dashboard and the ability to move to several machines unchanged — all of which matter once a run takes hours.

How do I keep the workers’ environment in sync?

A container image, pinned by digest, used for both local and cluster runs. Mismatched environments produce failures that look like data problems, and the digest belongs in the run summary alongside the manifest.

Back to 3D Tiles Batch Tiling Pipelines.