Transforms

The gwenlake.transforms layer lets you declare what a function reads and writes with decorators. The platform registers the function from those bindings, runs it in a build and records the lineage. Datasets and models are addressed by their handle, "<project_slug>.<slug>", the same one used in SQL. Bindings are matched to the function’s parameters by keyword name.

`transform_df`: DataFrames in, DataFrame out

The function receives each Input as a pandas.DataFrame and returns the DataFrame to write to the single Output. The result is written for you, replacing the dataset’s content by default.

from gwenlake.transforms import transform_df, Input, Output

@transform_df(
    raw_data=Input("Project_A.users"),
    processed_data=Output("Project_A.users_filtered"),
)
def process(raw_data):
    df = raw_data[raw_data["age"] >= 18].copy()
    df["name_upper"] = df["name"].str.upper()
    return df

process(client)   # reads, computes, writes

With several outputs, return a dict keyed by their names:

@transform_df(
    raw=Input("Project_A.contacts"),
    valid=Output("Project_A.contacts_valid"),
    rejected=Output("Project_A.contacts_rejected"),
)
def split(raw):
    ok = raw["email"].str.contains("@")
    return {"valid": raw[ok], "rejected": raw[~ok]}

`transform`: explicit reads and writes

The lower-level form receives TransformInput and TransformOutput objects and does its own IO. Use it for append mode, for chunked processing, or for files.

from gwenlake.transforms import transform, Input, Output

@transform(
    my_input=Input("Project_A.users"),
    my_output=Output("Project_A.users_distinct"),
)
def dedupe_users(my_input, my_output):
    df = my_input.dataframe()
    my_output.write_dataframe(df.drop_duplicates(), mode="replace")

mode="replace" (default) clears the dataset first and writes a snapshot; mode="append" keeps what is there and adds the file. write_dataframes(chunks) streams an iterator of DataFrames out as part files, see Large datasets.

Large datasets

A transform_df loads each input whole. When an input does not fit, use transform and stream.

Read less

columns= restricts a read to the named top-level columns, on the binding or per call. It is the single most effective way to cut the cost of a wide dataset: a transform that uses ten of two hundred columns otherwise pays to move all two hundred. chunk_size and order_by steer the paging.

@transform_df(
    events=Input("Project_A.events", columns=["id", "ts", "user_id"], chunk_size=100_000, order_by="id"),
    out=Output("Project_A.events_slim"),
)
def slim(events):
    return events

Chunked reads and writes

In a transform, iter_dataframes() pages through the input with LIMIT/OFFSET and yields pandas.DataFrame chunks; write_dataframes() streams them back out as part-00000.parquet, part-00001.parquet, …

from gwenlake.transforms import transform, Input, Output

@transform(
    big_dataset=Input("Project_A.events"),
    result=Output("Project_A.events_clean"),
)
def transform_in_chunks(big_dataset, result):
    chunks = (
        chunk[chunk["valid"]]
        for chunk in big_dataset.iter_dataframes(chunk_size=50_000, order_by="id")
    )
    result.write_dataframes(chunks, mode="replace")

Always pass order_by= so that pages are split deterministically.

Local Parquet, batches, row count

For libraries that want a file, to_parquet() writes the whole input to a local Parquet file, in chunks, and returns the row count; with reuse=True (default) an existing complete file is kept, so a resumed run skips the download.

rows = big_dataset.to_parquet("/tmp/events.parquet", chunk_size=100_000)

batches(size) returns a callable that yields batches of size rows, ready for a training loop’s run.steps(...), see Models. input.rows (also len(input)) is the row count, counted server-side with one COUNT(*) and kept for the run, to size a loop before reading anything.

Working with files

.filesystem() on an input or an output gives a small filesystem over the dataset: ls(path=, glob=), read, write, delete, clear and open. Use it for images, PDFs and any raw file.

@transform(
    images=Input("Project_A.scans"),
    thumbnails=Output("Project_A.scans_processed"),
)
def process_files(images, thumbnails):
    src, dst = images.filesystem(), thumbnails.filesystem()
    for entry in src.ls(glob="**/*.png"):
        data = src.read(entry["filename"])
        with dst.open(f"thumbs/{entry['filename']}", "wb") as f:
            f.write(make_thumbnail(data))

open(..., "rb") returns the downloaded bytes; open(..., "wb") returns a buffer that uploads on close.

Other bindings

Binding	The function receives	Use
`Input("p.dataset")`	a DataFrame, or a `TransformInput`	a dataset to read
`Output("p.dataset")`	a `TransformOutput` (written for you in `transform_df`)	a dataset to write
`Model("p.model")`	a `TransformModel`	artifacts to load, see Running
`Connection("p.slug")`	a `TransformConnection`	a source to read, see Connections

Several outputs, one build

A function with several Outputs is registered as one transform per output, and the platform knows they are one program: running any of them runs the function once and rewrites every output, a second build of the same function is refused while one is running, and the outcome is recorded on every output. The outputs are written one after the other, in declaration order, without atomicity between them, and a failure’s message says so.

Incremental transforms

A transform that must not redo what it has already done keeps its memory in one of its own outputs, a log of what it has processed, and reads it back on the next run. An output can never be an input of the same transform (that would be a cycle in the lineage), so the @incremental decorator gives outputs a "previous" reading mode and makes writes add to the dataset instead of replacing it. Stack it above @transform or @transform_df:

import pandas as pd
from gwenlake.transforms import incremental, transform, Connection, Output

@incremental(semantic_version=1)
@transform(
    share=Connection("Ops.drop-folder"),
    pages=Output("Ops.pages"),
    log=Output("Ops.processed"),
)
def ingest(share, pages, log):
    done = set(log.dataframe("previous", schema=["path"])["path"])   # empty on the first run
    new = [f for f in share.filesystem().ls(glob="**/*.pdf") if f["path"] not in done]
    if not new:
        return
    pages.write_dataframe(parse(new))                                       # appended
    log.write_dataframe(pd.DataFrame({"path": [f["path"] for f in new]}))   # last: a failure re-processes, never forgets

The source here is a connection, read whole on every run: the log is what remembers. The same pattern works with a dataset as the source.

Reading an output

Call	Returns
`output.dataframe("previous", schema=[...])`	the output as the last successful run left it; an empty frame of `schema` when there is nothing yet
`output.dataframe("current")`	the output as it is now, re-read, this run’s writes included
`output.filesystem("previous")`	the files as they were before this run wrote, read-only (`ls`, `read`, `open`)
`output.has_previous`	whether there is anything to read as `"previous"`

"previous" is read before the run’s first write and memoized: ask for it afterwards and the library refuses rather than answer a mix of old and new rows. schema shapes the empty frame of a first run; pass a list of column names, a {name: dtype} mapping, a pyarrow.Schema or a DataFrame to copy the columns of. columns= restricts the read, which is all a log of paths needs:

done = set(log.dataframe("previous", schema=["path"], columns=["path"])["path"])

Writing

Under @incremental a write adds to the output: an append into a tabular dataset, or one more part file into a file dataset, named part-<build>-<n>.parquet so two runs never overwrite each other. Several writes in one run add up. To start the output over in one run, call output.set_mode("replace") before the first write, or pass mode="replace" to it: the first write then replaces the dataset and the following ones add to it.

@incremental()
@transform(events=Input("p.events"), daily=Output("p.daily"))
def rebuild(events, daily):
    daily.set_mode("replace")
    daily.write_dataframe(events.dataframe().groupby("day").size().reset_index(name="n"))

Starting over: `semantic_version`

Bump semantic_version when the logic changes in a way that invalidates what was written: the next build starts over, "previous" comes back empty and the first write replaces. The platform records, after each successful build, the version each output was written with; a run whose decorator says another version is a run that starts over. A run that starts over and finds nothing to write still empties the output, so the bump means what it says. Leave the version alone for an ordinary code change: the log is the memory, and it survives deployments.

What a build records

On success, the platform stamps every output of the transform with the semantic_version the code declared and, for tabular inputs, the point each one was read up to (see below). This is what the next run compares itself to.
On failure, nothing is recorded. What the run appended before failing stays in the output and is the "previous" content of the next run, which is why the log is written last: a failure between the two writes re-processes the files, it never forgets them. The build’s error message says so.

Reading what was added to a tabular input

Inside a build, a tabular input also answers dataframe("added"): the rows appended since the last successful run, read from the table’s own snapshot history. Before the run starts, the platform pins where each tabular input stands; "added" reads up to that point and never “now”, so a row landing while the run is going is read once, by the next run. dataframe("previous") is the table as the last run saw it.

from gwenlake.transforms import incremental, transform, Input, Output

@incremental(semantic_version=1)
@transform(
    events=Input("Project_A.events"),
    daily=Output("Project_A.events_daily"),
)
def aggregate(events, daily):
    new = events.dataframe("added")          # everything on the first run
    if new.empty:
        return
    daily.write_dataframe(new.groupby("day").size().reset_index(name="n"))

iter_dataframes("added", chunk_size=...) streams the same rows. Both take columns= and order_by=. Three cases have no delta, and the library or the platform says so instead of quietly answering with the whole table:

a file dataset, a connection, or a run outside a build: nothing keeps their history, so "added" and "previous" are refused and the input is read whole with dataframe();
a table rewritten in between: a snapshot upload, a delete or an upsert changes rows the previous run had already seen, so the rows added since do not describe what changed. The read fails with a message that says to read the table whole or bump semantic_version;
a snapshot expired by the dataset’s maintenance before the next run: same answer. Keep the maintenance window longer than the interval between two runs of the transform.

"added" is a transform feature: a @transform_df body receives its inputs as whole DataFrames. Under @incremental, the DataFrame a @transform_df returns is added to its output rather than replacing it, and a parameter named after an output receives the output handle, so the body may read its "previous" or set_mode it first.

Options accepted for compatibility

@incremental also accepts snapshot_inputs, require_incremental and allow_retention, so a script written against the same decorator elsewhere runs as is. They do nothing here: every input is already read whole unless "added" is asked for, the platform never falls back to a full rebuild on its own, and nothing deletes from an input behind a transform. @incremental is not for @train: a model is replaced by its training, never appended to.

Dependencies and settings

A build runs the entrypoint with uv run from the repository root, so the repository’s pyproject.toml and uv.lock (or PEP 723 inline metadata in the script) decide what is installed. Three settings on the transform add to that without a commit: packages (extra package specs), requirements (a requirements file in the repository) and python (an interpreter version). A fourth, parameters, is a JSON object the run receives as TRANSFORM_PARAMETERS: one transform attached to several datasets reads its chunk size or its model name from there. See Builds.

Running

Call the decorated function with a client to run it from your machine. In a repository the platform builds, use run as the module’s entrypoint, so a build runs only the transforms it targets:

from gwenlake.transforms import run

if __name__ == "__main__":
    run(client, process, split)

The transforms layer is synchronous. It uses the sync Gwenlake client even in an async program.

Get started

Concepts

Workspace

Data

Code

Build

Models

Inference

Agents

Client

Changelog

`transform_df`: DataFrames in, DataFrame out

`transform`: explicit reads and writes

Large datasets

Read less

Chunked reads and writes

Local Parquet, batches, row count

Working with files

Other bindings

Several outputs, one build

Incremental transforms

Reading an output

Writing

Starting over: `semantic_version`

What a build records

Reading what was added to a tabular input

Options accepted for compatibility

Dependencies and settings

Running

​transform_df: DataFrames in, DataFrame out

​transform: explicit reads and writes

​Large datasets

​Read less

​Chunked reads and writes

​Local Parquet, batches, row count

​Working with files

​Other bindings

​Several outputs, one build

​Incremental transforms

​Reading an output

​Writing

​Starting over: semantic_version

​What a build records

​Reading what was added to a tabular input

​Options accepted for compatibility

​Dependencies and settings

​Running

`transform_df`: DataFrames in, DataFrame out

`transform`: explicit reads and writes

Large datasets

Read less

Chunked reads and writes

Local Parquet, batches, row count

Working with files

Other bindings

Several outputs, one build

Incremental transforms

Reading an output

Writing

Starting over: `semantic_version`

What a build records

Reading what was added to a tabular input

Options accepted for compatibility

Dependencies and settings

Running