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 the **CPU quota you actually have** — read it from the cgroup (`cpu.max` on v2, `cpu.cfs_quota_us ÷ cpu.cfs_period_us` on v1), never from `cpu_count()`, which reports visible CPUs, not your allotment | | `[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 runs out of resources on a constrained machine is short of **RAM** or of **disk**, and the fixes diverge. Don't guess; read how it died. Start from the table, then read the two exceptions under it — RAM does not always announce itself as a kill: | Signal | RAM exhaustion | Disk exhaustion | |---|---|---| | How it dies | usually **no Python traceback** (two exceptions below). Exit **137 / SIGKILL** or pod `OOMKilled` on k8s/Docker. On the **dltHub platform, three signatures — all RAM**: `Runner killed (SIGKILL), exit code: 137` with an out-of-memory note (fast climb), `Runner failed with exit code: 247` after a silent stall, or **no kill at all** (see below) | 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 | **A job can run out of RAM without dying — and that looks like a hang, not a crash.** It sits with memory parked at the tier's limit, unchanging on every sample, emitting **no progress lines at all**, never killed, until the platform job's `execute.timeout` (its wall-clock limit) expires. Progress output stopped **and** memory at the ceiling is RAM exhaustion: cancel it and tune — don't wait for an exit code and don't raise the timeout. The same config can fail either way on different runs, so never key the diagnosis on the exit code alone. With `progress="log"` on, tie a RAM spike to its stage: extract spiking = pulling the source into memory; normalize spiking = buffering items. **A traceback can still be a memory problem — read *whose* memory ran out.** Once the destination has its own memory cap, *its* OOM arrives as a normal Python exception naming the limit (e.g. `_duckdb.OutOfMemoryException: Out of Memory Error: could not allocate block of size …`), wrapped by dlt as `DatabaseTransientException`, retried `max_retry_count` times, then `LoadClientJobRetry` and exit 1. That is the cap you set doing its job — an anonymous kill or stall converted into an attributable error. Fix it by raising the cap (keeping `[load] workers × file size` inside the box) or shrinking the files. A traceback that is neither `Errno 28` nor a destination out-of-memory 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). > **It is a floor, not a peak — it is sampled at progress ticks, not continuously.** A spike *inside* a stage (parsing a whole payload, materializing a dataframe) happens between two log lines and never appears. In practice it understates the true peak by **2–3×**, and the gap is worst exactly where it matters — in a stage that allocates in bursts, such as load. **Never conclude "memory is fine" from this line**, and never size an instance down on the strength of it; only an OOM kill, or a peak you sample yourself (the runner exposes no cgroup peak counter — see below), gives you the real high-water mark. > **The percentage is not your process's share** — dlt prints `rss` for the current process but takes the percentage from `psutil.virtual_memory().percent`, i.e. **system-wide** memory in use, so on a busy laptop an alarming percentage can sit next to a few-hundred-MB process. Never read it as "how close am I to the limit". > > **On the dltHub runner it does tell you the tier**, because `/proc/meminfo` there is namespaced to the instance — so "system" *is* your instance, `psutil.virtual_memory().total` is the tier's memory, and `MB ÷ pct` off this line recovers it. Use either to confirm which tier a run actually got: `dlthub deploy --show-manifest` only shows the size you **declared**, and prints no `require` block at all on the default tier. 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. **On the dltHub runner neither peak file exists** — it mounts cgroup **v1** with only `memory.limit_in_bytes` and `memory.usage_in_bytes`, so there is no kernel high-water mark: sample `memory.usage_in_bytes` **and** RSS from a ~1 s thread and keep both maxima. They agree while dlt does its own Python work, then diverge once a memory-mapping destination (duckdb) loads, where RSS runs well above what the cgroup charges. **RSS is the alarm; the cgroup figure is what gets you killed** — a run peaking "over" the tier in RSS can still finish. - **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. Cutting `[extract] workers` / `[normalize] workers` to fix memory usually just slows the run — those stages are bounded by the buffer levers above. **`[load] workers` is the exception**: load holds one whole file per thread, so its peak scales with `workers × file size` — see [Load](#load-is-slow-destination-io-bound). **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. Only once **all** the levers in this section are in place — buffers lowered, files rotated, destination capped, and the fresh-pipeline-per-object pattern below where it applies — is a genuinely undersized runner a possibility: see [Step 4](#step-4-last-resort--raise-the-instance-size-dlthub-platform) (dltHub platform only). **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 ```