leeghwater
Your dlt pipeline, the same on your laptop and in a Databricks job. One command to run it, its secrets from a secret scope, and the things Databricks asks for kept out of your code.
Early. It has run as a serverless job and from a laptop against a real workspace; what was tried and what was not is under Tried, and not yet.
leeghwater is a package your dlt project imports. Your sources stay ordinary dlt code. leeghwater gives the project a command line, and for each run it makes the dlt pipeline from what the run names: a local file on a laptop, the schema your bundle deployed in a job.
Why
A dlt pipeline on Databricks tends to run only on a cluster. Every change is a deploy and a wait, and the secrets are in one place on a laptop and another in a job. So nobody runs the pipeline locally, and finding a typo takes ten minutes. With leeghwater the same command runs the pipeline on a laptop into a local file and unchanged in a job, with its secrets from one place.
A project is two files
The app, which is the project's console script:
# src/my_ingest/cli.py
from leeghwater import App
app = App(pipelines="my_ingest.pipelines", secret_scopes=["ingest"])
# pyproject.toml
[project.scripts]
ingest = "my_ingest.cli:app"
And a pipeline: a function marked @pipeline, anywhere in the package the app names.
# src/my_ingest/pipelines/github.py
from datetime import date
import dlt
import leeghwater
from leeghwater import pipeline
@dlt.source
def github_source(since: date, access_token: str = dlt.secrets.value): ...
@pipeline
def github(since: date = date(2024, 1, 1)):
"""Load GitHub issues."""
p = leeghwater.create_pipeline("github")
return leeghwater.run(p, github_source(since=since))
create_pipeline and run stand in for dlt.pipeline() and pipeline.run(), and a
pipeline never calls those two itself. Where a run loads, into which schema and through
which staging schema is the run's to say, not the code's: create_pipeline sets them on
the dlt destination it makes, and a plain dlt.pipeline() gets none of it. What you get
back is dlt's own pipeline and dlt's own load info, and anything else the two dlt calls
take is passed on.
ingest list # the pipelines and their parameters
ingest run github --since 2024-06-01 # run one
ingest doctor # what would be set up, checked, nothing run
Start a project
data-product-template is the way to start: a copier
template for a data product — dlt to land the data, run by leeghwater; dbt to shape it; one
job; the schemas as bundle resources; every task behind mise run.
uvx copier copy gh:kostavo-oss/data-product-template my-product
To add leeghwater to a project you have: uv add leeghwater, write the two files above, and
pass the schemas' deployed names from the bundle as shown under In a job.
Where a run loads
| Where you run it | It loads into | Secrets come from |
|---|---|---|
| A laptop | a local DuckDB file, <pipeline>.duckdb |
the environment, .dlt/secrets.toml |
A laptop, with --profile <name> |
that workspace, as you | the same, then the secret scopes |
| Databricks | this workspace, as the run's identity | the environment, then the secret scopes |
The first row needs nothing but Python and dlt[duckdb]. It tries your source, not the
destination: DuckDB ignores what is Databricks' own, such as table properties.
A project that loads somewhere else says so once, in the app:
App(..., destination=..., local_destination=...).
The options of every run
A pipeline's own parameters are options of its command: since above is --since, with
the type the function gives it. Text, whole and decimal numbers, dates and times in ISO
format, and yes or no as a word (--full true), so a bundle variable can set it.
Every run also takes these:
| Option | What it sets |
|---|---|
--profile <name> |
Load into a workspace from a laptop, with a profile from ~/.databrickscfg. Refused on Databricks. |
--catalog <name> |
The catalog to load into. |
--schema <name> |
The schema to load into, used exactly as given. |
--staging-schema <name> |
The schema dlt stages through, by its whole name. |
--warehouse <id> |
The SQL warehouse to load through. |
--secret-scope <name> |
A secret scope to read, in place of the app's. May be repeated. |
--secret <name>=<scope>/<key> |
Where one secret is, when it isn't under dlt's name. May be repeated. |
--key-vault <url> |
An Azure Key Vault to read, after the scopes. May be repeated. |
--limit <n> |
At most n pages from each resource; 0, the default, for all. |
--log-level <level> |
DEBUG, INFO, WARNING, ERROR or CRITICAL, for dlt. |
--set <key>=<value> |
Any dlt config value, by dlt's name: --set sources.github.page_size=50. |
The schema, staging schema, catalog and warehouse are set on the destination that
create_pipeline makes, in the call, which is what dlt takes first. --set and
--log-level go to dlt as config, the way an environment variable does. Nothing in dlt is
patched.
A pull-request environment runs with --limit 1: enough for dlt to make every table with
its real columns. Not with nothing: dlt takes a table's columns from the rows it sees, so a
source that gives no rows makes no table.
In a job
The same command is the job's entry point. Nothing in the task names leeghwater.
python_wheel_task:
package_name: my-ingest
entry_point: ingest
parameters:
- run
- github
- --catalog
- ${var.catalog}
- --schema
- ${resources.schemas.github.name}
- --staging-schema
- ${resources.schemas.github_staging.name}
- --warehouse
- ${var.warehouse_id}
Why the schemas are passed by reference
A bundle names your schemas, and a target may change the name it deploys: a prefix, or a
suffix per developer, such as github__anna. Left alone, dlt gets this wrong twice:
- it normalizes a dataset name, and
github__annabecomesgithub_anna: a schema nobody deployed; - it derives its staging dataset as
<name>_staging, sogithub__anna_staging, while the bundle deployedgithub_staging__anna. dlt then makes a schema of its own, outside the bundle's grants and its cleanup.
--schema turns the normalizing off for the run, and --staging-schema gives dlt the
staging schema's whole name. On Databricks a run that names a schema and no staging schema
stops, and says which line of the job is missing. That holds for a pipeline that never
merges too: a staging schema that is there and unused costs nothing, one that is missing
costs a schema nobody governs.
A run ends the way it went
A run prints what was set up, then a report of the load. It ends with an error when:
- the pipeline raises;
- a load job failed. dlt raises on that by itself; a project that turned that off
(
load.raise_on_failed_jobs) still gets a failed run; - a run into Databricks names no warehouse. dlt would otherwise load through the first warehouse on the workspace's list;
- the pipeline made no load with
leeghwater.run(): a plaindlt.pipeline()loads wherever dlt's own config says, with no schema, catalog or limit from the run.
A serverless job retries a failed task once by itself unless the task sets
disable_auto_optimization: true: set it, so a load is retried when you say so, with
max_retries.
Secrets
A source asks dlt for a secret: access_token: str = dlt.secrets.value. dlt looks in the
environment, then in .dlt/secrets.toml, then in the secret scopes the app or the run
names. Three ways to give it one, and when to use each:
-
By dlt's name, in a scope. The path dlt asks for, joined with
-:databricks secrets put-secret ingest sources-github-access_token
The default. One secret per value, and no code.
-
A whole section in one secret. A secret named
sources-github(orsources,destination-<name>,destination,dlt_secrets_toml) is read as a piece ofsecrets.toml. It must spell out its own heading:[sources.github] access_token = "..."
Without the heading its values are found as every source's. Use it when a source has several secrets that change together.
-
A secret that exists under another name. Point at it, as
<scope>/<key>:app = App(..., secrets={"sources.github.access_token": "platform-shared/github-token"})
or per run, so dev and prod can differ in the bundle:
--secret sources.github.access_token=platform-shared/github-token. Use it when somebody else fills the scope. -
An Azure Key Vault, for a team whose secrets live there and not in a scope. Needs the extra,
uv add "leeghwater[keyvault]", and a vault named in the app or per run:app = App(..., key_vaults=["https://kv-ingest.vault.azure.net"])
A vault is asked after the scopes. Its secrets are named like a scope's, with dashes where a Key Vault allows no underscore:
sources-github-access-token. WhoeverDefaultAzureCredentialfinds reads it:az loginon a laptop; a service principal's variables or a managed identity elsewhere. A Databricks job has no identity in Azure by itself, so there a vault is reachable only through a service principal whose secret comes from somewhere else — in practice a Databricks scope — and only when the job's network reachesvault.azure.net. A Key Vault-backed Databricks scope needs none of this: it is a scope, read as in 1.
A scope or a vault is listed once in a run and only the secrets that are there are read. Only values a source marks as secret are looked up. No value is printed, logged, written to a file or put in the environment. To make, change or rotate a secret, use the Databricks CLI or the Azure portal.
In a job a secret is read through dbutils, so Databricks redacts it: if your own code
prints one, the job's output shows [REDACTED].
In a notebook, or without an app
import leeghwater
setup = leeghwater.prepare(secret_scopes=["ingest"], catalog="main", warehouse="<id>")
print("\n".join(setup.lines()))
p = leeghwater.create_pipeline("github", schema="github")
leeghwater.run(p, github_source())
prepare() is everything the app does before a run, and returns what it decided;
create_pipeline and run go by the last prepare(), or by a setup= you pass them.
Importing leeghwater changes nothing and does not import dlt.
Commands of your own
@app.command()
def backfill(day: str):
"""Load one day again."""
...
ingest backfill 2024-06-01 is prepared as a pipeline is, with the app's own settings, and
finds the same secrets.
Troubleshooting
- Deleting the DuckDB file does not start a pipeline over. dlt keeps a pipeline's state
outside the database, under
~/.dlt/pipelines/<name>/. Delete that folder too. import dltgives Delta Live Tables. Databricks has a module of its own nameddlt. In a serverless job a plainimport dltgets the right one. In a notebook or on a cluster, callleeghwater.prepare()beforeimport dlt. The first lines of a run say what was done about the name.- The pipeline and its schema have the same name, and DuckDB refuses. DuckDB names its
database after the file,
<pipeline>.duckdb, and can't tell a schema of that name from it. Locally, give the schema another name or leave it out. - A load from a job is refused a connection to
…storage.cloud.databricks.com. dlt uploads its files to a volume before it copies them into a table. A workspace whose serverless compute has a limited outbound network refuses that upload. ingest doctorsays where a run would load and checks the sign-in, each secret scope and the warehouse, without running anything.--jsonfor a machine.
Tried, and not yet
On 2026-10-07 a project made by leeghwater init was deployed to a workspace as a bundle
and run, and the same project was run from a laptop with a profile.
Held, for real:
- a bundle made from the former
leeghwater initvalidates, deploys and destroys, and the job's entry point is the project's own command; - the job is told the deployed names of the bundle's schemas, and finds the
config.tomlits wheel carries; - a pipeline that raises fails its task;
- a secret scope is read as the job's identity and from a laptop, and a secret is redacted in the job's output;
- from a laptop, a load into Unity Catalog; and with schemas named with two underscores, a load into exactly those, with no third schema made.
Not yet:
- a full load from a serverless job. The workspace it was tried on limits the outbound network of serverless compute, and dlt's upload to storage was refused. Everything before the upload ran;
- a notebook and a classic cluster, which is where the name
dltis said to bite; - the job's
ingesttask with leeghwater installed from PyPI, which 0.1.0 makes possible.
When not to use it
- Lakeflow Connect has a connector for your source and it does what you need.
- dlt runs somewhere else — Airflow, GitHub Actions — and only loads into Databricks. dlt does that alone.
- The data needs Spark to move it. dlt is Python on one machine, and leeghwater doesn't change that.
leeghwater is a library for one kind of code, dlt ingestion. It will not:
- deploy, schedule, or order one pipeline after another. That is a job's, and a bundle's;
- add a file format for pipelines, or patch anything in dlt;
- make, change or rotate secrets. That is the Databricks CLI's, and caland's;
- change tables, or set table properties and clustering. dlt's own
databricks_adapterdoes the second, in your source; - change anything on a cluster: no init script, no file outside the run's own directories.
When you no longer need it
When dlt reads Databricks secret scopes by itself and runs in a job with nothing around
it — the right dlt imported, the bundle's schema names used as they are — use dlt
alone. Today dlt has no provider for secret scopes, and its Databricks
page gives workarounds for the import.
Where it fits
leeghwater is one of the Kostavo tools: small tools for the ugly gaps on Databricks, one gap each. Its gap is the pipeline that only runs on a cluster: every change a deploy and a wait, and the secrets in one place on a laptop and another in a job. Each works alone. Kostavo is the company behind them: it builds a governance platform for Databricks workspaces, and the tools are complete without it.
Community project, not affiliated with or endorsed by Databricks or dltHub.
Named after
Jan Adriaanszoon Leeghwater — the millwright who drained the Beemster with windmills, dry in 1612. He moved water from where it lay to where it was wanted.
License
Metadata
Release files for leeghwater 0.1.1
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| leeghwater-0.1.1.tar.gz | 50.3 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| leeghwater-0.1.1-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 89.3 kB
Release files / leeghwater-0.1.1.tar.gz
| Download URL | leeghwater-0.1.1.tar.gz |
|---|---|
| Size | 50.3 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
cab6ca9e745e88d95de8201a8bee966b86c7e62d30d23a91df97bc5c11737b1b
|
|
BLAKE2b-256 checksum How to use checksums |
334189028f5a9b6045adf64b63a612857574c49af730718c04e625ecca3b8516
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
uv/0.13.0 {"installer":{"name":"uv","version":"0.13.0","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
|
Release files / leeghwater-0.1.1-py3-none-any.whl
| Download URL | leeghwater-0.1.1-py3-none-any.whl |
|---|---|
| Size | 39.0 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
a405c9c3b231061bcf3e3bb16aee4e6a33bee453960bdd26782e74659814478c
|
|
BLAKE2b-256 checksum How to use checksums |
1dd6b4ec9ee3b626c84eda38528207553856c22a69a8d3aefbe6b7052cd17759
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
uv/0.13.0 {"installer":{"name":"uv","version":"0.13.0","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
|