optimize-performance

# Optimize dlt pipeline performance dlt runs in three stages — **extract → normalize → load** — each with its own parallelism and memory knobs. Tuning the wrong stage wastes effort, so **diagnose the bottleneck first (Step 1), fix that one stage (Step 2), then re-measure (Step 3)**. **Essential reading:** https://dlthub.com/docs/reference/performance Parse `$ARGUMENTS`: - `pipeline-name` (optional): the dlt pipeline. If omitted, infer from session context; if ambiguous, ask and stop. - `symptom` (optional): what's wrong — "slow extract", "OOM during normalize", "load takes hours", etc. > Source-specific knobs (DB backends/`chunk_size`, REST async/concurrency, file format/readers) live in a separate pipeline toolkit — see [Source-specific tuning](#source-specific-tuning-separate-toolkits) for the toolkit to install. Apply those **in addition** to the stage levers here. > > Knobs below are shown as `config.toml`; each has an env-var equivalent — see [Config forms](#config-forms). Start from dlt's defaults and change **one lever at a time**. ## Step 1: Diagnose the bottleneck stage Always measure before changing anything. Enable progress logging and read the per-stage timing: ```toml # .dlt/config.toml progress = "log" # or PROGRESS=log, or dlt.pipeline(..., progress="log") ``` **Install `psutil` first** (`pip install psutil` or `uv add psutil`) — then `progress="log"` adds a `Memory usage: … MB (…%) | CPU usage: …%` line per stage; without it dlt warns once (`psutil dependency is not installed and memory stats will not be available`) and omits that line. It's one interface across macOS/Windows/Linux, so the same profiling runs in dev and prod. Read durations from `pipeline.last_trace` or via `debug-pipeline` (trace + load packages; `debug-pipeline` ships with the calling pipeline toolkit — use it only if that toolkit is installed, otherwise read `last_trace` directly). Benchmark on a real machine — shared or resource-constrained environments give unreliable numbers. **Tune on a representative subset, not the full dataset.** For time tuning a slice (e.g. a few million of tens of millions of rows) reproduces the stage ratios and lever effects while iterating far faster. Size it in a band: big enough that fixed overhead doesn't dominate **and** that enough files/resources exist for parallelism levers to show (too small → a good lever looks useless; a tiny `.add_limit(1)` fails this floor and `.add_limit()` masks pagination), small enough to iterate. **Reduce volume only once you've confirmed the subset is representative** — it reproduces the full-run stage ratios and clears the floor (>1 file, >1 resource); if you can't confirm that, run the full dataset. Memory/OOM work and the final confirmation always run full volume. Then route to the slowest stage: | Bottleneck | Bound by | Go to | |---|---|---| | Extract slow | network / source I/O | [Extract](#extract-is-slow-io-bound) + [Source-specific tuning](#source-specific-tuning-separate-toolkits) | | Normalize slow (CPU) | CPU | [Normalize](#normalize-is-slow-cpu-bound) | | Out of memory / killed, or disk full | RAM **or** disk | [Out of memory](#out-of-memory-ram-or-disk) | | Load slow | destination I/O | [Load](#load-is-slow-destination-io-bound) | ## Step 2: Fix the bottleneck Jump to the **one** section Step 1 flagged. Memory isn't a stage — it's a separate failure mode with its own section. ### Extract is slow (I/O-bound) By default dlt extracts **one resource at a time, one item at a time** — N endpoints/tables are fetched sequentially. Parallelism overlaps those I/O waits, but only helps when multiple slow resources can overlap. **Decide before adding workers:** 1. **Count resources.** One resource → cross-resource threading won't help; look at async-within-it, page size, or another stage. Many → candidate. 2. **Measure per-resource time** (`pipeline.last_trace`, or extract them individually). To probe one resource cheaply, cap it with `.add_limit(max_items=N)` or `.add_limit(max_time=seconds)` and time a few pages — don't extrapolate from a single item, and note `.add_limit()` masks pagination (re-verify before removing it; see `adjust-endpoint`/`adjust-table`). If one resource dominates, parallelizing the rest saves nothing — fix the slow one. 3. **Confirm it's I/O, not CPU.** Threading overlaps waits, not Python compute (see [How parallelism works](#how-parallelism-works)). 4. **Size `workers`** ≈ number of slow resources to overlap. Excess workers idle or trip rate limits. | Lever | Default | Effect | |---|---|---| | `@dlt.resource(parallelized=True)` | off | run the (sync) resource's generator in a worker thread, so resources overlap | | `@dlt.defer` on a per-item fn, then `yield fn(item)` | — | run per-item work (e.g. one HTTP request per item, inside a transformer) concurrently in the extract thread pool — the sync alternative to `async` | | `async def` resource / async `@dlt.transformer` | concurrent | many awaited items in flight on one event loop — no flag needed | | `[extract] workers` | 5 | thread-pool size for parallelized sync resources | | `[extract] max_parallel_items` | 20 | max in-flight items for async resources | | yield pages, not rows | — | `yield list(islice(it, 1000))` cuts per-item overhead; orthogonal to threading | Group related resources into a `@dlt.source` (`return (r1, r2, r3)`) so dlt schedules them on a shared thread pool. Scopes for the knobs: global, per-source (`[sources.<name>.extract]`), per-resource. ### Normalize is slow (CPU-bound) Normalize runs a **process pool with one worker per extract file** — so more workers only help if **extract produced many files**. Key rule: each stage's `data_writer` rotates the files it *writes* (= the next stage's input), so **rotate extract output to parallelize normalize; rotate normalize output to parallelize load**. | Lever | Default | Effect | |---|---|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | `[normalize] workers` | 1 (serial) | process pool; raise toward CPU-core count | | `[normalize] start_method = "spawn"` | — | recommended on Linux when resources use threads | | `[data_writer] file_max_items` / `file_max_bytes` | none | rotate stage outputs so normalize and load each get many files to split across workers — **the lever that parallelizes both**. Global `[data_writer]` rotates **both** stages' outputs at once; scope it to target one — `[sources.data_writer]` (extract output → parallelizes normalize), `[normalize.data_writer]` (normalize output → parallelizes load, see [Load](#load-is-slow-destination-io-bound)) | | `[normalize.data_writer] disable_compression = true` | off (gzip) | trade disk for CPU when CPU-bound | Note: the extract writer is scoped under **`[sources.data_writer]`** (or global **`[data_writer]`**) — there is no `[extract.data_writer]` scope; dlt **silently ignores** that one (no error, no effect). Running *out of memory* during normalize is a different problem with a different fix set → [Out of memory](#out-of-memory-ram-or-disk). ### Out of memory (RAM or disk) A pipeline that dies on a constrained machine fails one of two ways — the **failure mode tells you which**, and the fixes diverge. Don't guess; read how it died: | Signal | RAM exhaustion | Disk exhaustion | |---|---|---| | How it dies | **killed** — exit **137 / SIGKILL**, pod `OOMKilled`, **no Python traceback** | Python exception with traceback: `OSError: [Errno 28] No space left on device` | | Killed by | the OS / k8s OOM-killer (RSS hit the cgroup/container limit) | dlt/Python writing intermediate, load-package, or staging files | | Confirm mid-run | the `progress="log"` memory line (needs `psutil` — see Step 1); for the whole picture see [Tracking memory](#tracking-memory-dev-vs-production) below | `df -h` on the working volume and `$DLT_DATA_DIR`; `du -sh ~/.dlt` and the staging path | With `progress="log"` on, tie a RAM spike to its stage: extract spiking = pulling the source into memory; normalize spiking = buffering items. A traceback that **isn't** `Errno 28` is a different bug → use `debug-pipeline`. ##### Tracking memory (dev vs production) The `progress="log"` line reports **main-process RSS**, which is complete for the default serial run but undercounts once `[normalize] workers > 1` (the process pool runs in child processes). How to see the whole picture depends on where you can reach: - **Dev / staging, with a shell** — `docker stats` / `kubectl top pod` / `top` (sorted by RSS) / cgroup `memory.current` vs `memory.max`. These measure the **whole container**, so they already include the normalize pool. - **Production, no interactive shell** (the agent isn't attached) — don't rely on ad-hoc commands. Read memory from things emitted by the run itself: - **dltHub platform logs** — if the pipeline runs on the dltHub platform, read its logs directly: `dlthub job logs <name>` (latest run), `dlthub job runs logs <name> [run#]` (a specific run), or `-f` to stream. The `progress="log"` memory lines show up here. Ref: https://dlthub.com/docs/hub/pipeline-operations/monitoring - **From inside the run** — the `progress="log"` memory line lands in your logs; to capture the whole-tree peak with no external tooling, log the container's cgroup peak at the end of the run (cgroup v2: read `/sys/fs/cgroup/memory.peak`; v1: `memory.max_usage_in_bytes`) — it accounts for all processes in the container. - **The OOM-kill itself is a signal** — exit 137 / `OOMKilled` in the pod status or orchestrator events confirms RAM was the cause even with no live metrics. #### **RAM — bound peak memory:** | Lever | Default | Effect | |---|---|---| | `[data_writer] buffer_max_items` | 5000 | items in RAM before flush; **lower to cut memory** (scope per stage: `[sources.data_writer]` for extract, `[normalize.data_writer]` for normalize; `[data_writer]` for both) | | yield pages / stream input | — | stream the source (see [Extract](#extract-is-slow-io-bound)) so the **extract** side stays bounded — the load and destination still scale with data (see below) | | `[data_writer] file_max_items` / `file_max_bytes` | none | rotate files (global scope) so less is held in a single file at once | Gotcha: `buffer_max_items = 1` forces single-item writes and **disables multithreading** — don't. And **don't cut dlt's `workers` to fix memory** — that usually just slows the run; peak memory is bounded by the levers above and by the destination (next), not by dlt's worker count. **Cap the destination's own memory — dlt's buffers don't reach it.** Streaming and file rotation bound *dlt's* in-flight memory, but the destination loads with its **own** memory and thread/connection settings — often sized to the machine's total RAM — in or beside the same process. When extract is already fully streamed but the run still OOMs on a constrained box, suspect the destination: dlt's `buffer_max_items`/rotation won't touch it. To fix, cap the destination itself, not dlt's `workers`: - **Memory limit** — set the destination's own max-memory/heap setting below the container/cgroup limit. - **Threads / connections** — reduce the destination's concurrent thread or connection count (each holds its own buffers). - **Insert / batch size** — lower the destination's write batch size. Check the destination's own config for these knobs (they live in the destination, not in dlt). Capping destination memory and threads typically drops peak RSS several-fold versus running the destination at its RAM-sized defaults — often the single change that keeps a streamed load inside its limit. **Many-object loops — use a fresh pipeline per object.** Reusing one `pipeline` object across many `run()` calls grows its in-memory state (schema deltas, load packages, cursor values) run-over-run, which can OOM a long job. Create a **fresh pipeline per object** and drop it when done: ```python for obj in objects: pipe = dlt.pipeline(pipeline_name=f"load_{obj}", destination=..., dataset_name=...) pipe.run(source_for(obj)) del pipe ``` Caveat: trades away cross-run convenience, and `pipeline.default_schema.tables` keeps every table ever loaded under a given pipeline name — track the object list explicitly. #### **Disk** Stop intermediate/staging files filling the volume: see [Constrained disk / environment](#constrained-disk--environment) — point `DLT_DATA_DIR` at a bigger volume, stage to blob, set `delete_completed_jobs`, and keep compression **on** (don't `disable_compression`). ### Load is slow (destination I/O-bound) | Lever | Default | Effect | |---|---|---| | `[load] workers` | 20 | thread pool, one file per thread; I/O-bound, safe to raise toward destination capacity | Load runs **one file per thread**, so parallelism needs **many normalize-output files**: set `[normalize.data_writer] file_max_items` / `file_max_bytes` to rotate them (one file = one load job = no speedup). Smaller files also mean smaller transactions and less destination memory pressure. ## Source-specific tuning (separate toolkits) The biggest extraction wins are source-specific and live in the pipeline toolkit for your source — **not in this toolkit**. Pick the row for your source. If you arrived here from that toolkit it is already installed; otherwise install it first, then run its optimize skill: ``` uv run dlthub --non-interactive ai toolkit install <toolkit> ``` | Source | Install toolkit | Then run skill | |---|---|---| | REST / HTTP API | `rest-api-pipeline` | `optimize-rest-api-performance` | | SQL database | `sql-database-pipeline` | `optimize-sql-performance` | | Files (S3 / GCS / Azure / SFTP / local) | `filesystem-pipeline` | `optimize-filesystem-performance` | Apply source-specific tuning **in addition** to the stage levers above. These skills live in other toolkits, so always install the toolkit before invoking the skill — never assume it is present. ## Step 3: Re-measure, then repeat Re-run with `progress="log"` and compare per-stage durations against Step 1, plus the peak `Memory usage` line it logs (`last_trace` has timings, not RAM). Change **one lever at a time** so you can attribute the effect, and use `debug-pipeline` to confirm no new failures (timeouts, type errors) surfaced under higher concurrency. **If a lever regresses — slower, higher peak memory, or a new failure — roll it back to its previous value before trying the next**; changing one at a time is what makes this a clean revert. **Iterate on one stage — skip the full `run()`.** dlt exposes each stage as its own method (`pipeline.extract(source)`, `pipeline.normalize()`, `pipeline.load()`), and each consumes the previous stage's packages. So when tuning a lever, **re-run from the changed stage's input forward, up to the stage you're timing** — not the whole pipeline: | Lever | Re-run | |---|---|