Models

In Bauplan, models are the core unit of data manipulation. They are declarative functions written in Python or SQL that transform one or more input tables into a single output table. Models are designed to provide a straightforward way to express data transformations without dealing with containerization, data movement and runtime configuration.

Models can be chained together to form Pipelines, where downstream models depend on the outputs of upstream ones.

Anatomy of a Bauplan Model

A typical Python model consists of two decorators @bauplan.model() and @bauplan.python() and a function.

from typing import Annotated

import bauplan
import pyarrow

class SelectedInputColumns(bauplan.TableSchema):
    """Columns read from the input table."""

    col_1: bauplan.String
    col_2: bauplan.Int64
    col_3: bauplan.Float64

@bauplan.model()
@bauplan.python('3.11')
def my_model(
        data: Annotated[
            pyarrow.Table,
            bauplan.Model(
                "input_table",
                projection_schema=SelectedInputColumns,
                filter="timestamp >= '2022-12-15T00:00:00-05:00'",
            )
        ],
) -> Annotated[pyarrow.Table, SelectedInputColumns]:
    return data

Bauplan Models are fully declarative, allowing for explicit column selection and filter pushdown. In practice, this means your code only requires the name of the inputs and, optionally, the desired columns and filters.

For example, you don't need to specify whether input_table is an Iceberg Table, PyArrow Table, Pandas dataframe, or Polars dataframe. This approach makes the code fully portable, easier to reproduce across environments, and simpler to maintain.

Filters only apply to models from the lakehouse catalog

filter only works on an input that reads from the catalog. Filtering an input that is another model in the same pipeline is not supported, and the planner rejects it.

For example, my_model applies a filter on timestamp on bauplan.Model('input_table'). This is valid because input_table is not defined in the pipeline and must come from the catalog. It is not supported for a new model, other_model, to take my_model as an input and apply a filter to it.

@bauplan.model()
@bauplan.python('3.11')
def other_model(
    data: Annotated[
        pyarrow.Table,
        # filters can only be applied to models from the catalog
        bauplan.Model("my_model", filter="timestamp <= '2022-12-16'")
    ]
) -> Annotated[pyarrow.Table, SelectedInputColumns]:
    return data

Importing Python packages

Bauplan models are fully containerized, with each model running in the cloud as its own isolated environment (similar to Function-as-a-Service frameworks, like AWS Lambda). To isolate environments, Bauplan uses a optimized version of Docker containers.

Each model in Bauplan runs in its own isolated Python environment - defined by the interpreter version and uv dependencies you declare in code. They are expressed entirely in code using the decorator @bauplan.python(). This decorator specifies the Python interpreter version (for example, 3.11 or 3.12) and any additional libraries and their versions using uv.

class ColOne(bauplan.TableSchema):
    """The single column this model returns."""

    col_1: bauplan.String

@bauplan.model()
# specify the package and version - in this case Pandas 2.2.0
@bauplan.python('3.11', pip={'pandas': '2.2.0'})
def my_model(
    data: Annotated[pyarrow.Table, bauplan.Model('input_table')],
) -> Annotated[pyarrow.Table, ColOne]:
    # import the package declared in the decorator
    import pandas

    # use the package
    df = data.to_pandas()
    df = df[['col_1']]

    return pyarrow.Table.from_pandas(df)

This approach allows you to run each Bauplan model as an independent unit in a fully declarative way: all you need to run a Bauplan model deterministically in the cloud is the code.

This enhances reproducibility, prevents unintended cross-contamination between environments on the same machine, and provides the freedom to introduce new Python packages without worrying about backward compatibility. In fact, you can run a pipeline where different models use different versions of the same packages and different versions of the Python interpreter.

Each model builds its own environment, so Bauplan installs every package you declare for that model alone. Declare only what the function body imports.

from typing import Annotated
import bauplan
import pyarrow

class StepOutput(bauplan.TableSchema):
    """The columns each step passes along."""

    col: bauplan.Int64

@bauplan.model()
@bauplan.python('3.11', pip={'pandas': '2.2.0'})
def step_1(
    data: Annotated[pyarrow.Table, bauplan.Model('input_table')],
) -> Annotated[pyarrow.Table, StepOutput]:
    import pandas

    df = data.to_pandas()

    ...

    return pyarrow.table(df[['col']])

@bauplan.model()
@bauplan.python('3.11', pip={'pandas': '1.5.3'})
def step_2(
    data: Annotated[pyarrow.Table, bauplan.Model('step_1')],
) -> Annotated[pyarrow.Table, StepOutput]:
    import pandas

    df = data.to_pandas()

    ...

    return pyarrow.table(df)

Materializing tables with models

To write the output of a model as a table into the data catalog, use the materialization_strategy parameter in the @bauplan.model() decorator.
The parameter accepts four values: REPLACE, APPEND, NONE (default), and OVERWRITE_PARTITIONS.

Strategy	Description	Example
`"NONE"`	Streams model output in memory as an Arrow table without persisting to object storage.	Running will not write a table.
`"REPLACE"`	Fully overwrites the table on each run.	Running a pipeline twice on a model that writes 1,000 rows will result in a final table with 1,000 rows, as each run replaces the previous table.
`"APPEND"`	Appends new rows to an existing table.	Running a pipeline twice with a model that writes 1,000 rows results in a final table with 2,000 rows, as each run adds data to the existing table.
`"OVERWRITE_PARTITIONS"`	Deletes rows matching `overwrite_filter`, then appends the new rows.	See details below.

Regardless of the materialization_strategy parameter, running models with the dry run flag - bauplan run --dry-run - will prevent data from being materialized into the lakehouse catalog.

Overwrite partitions

OVERWRITE_PARTITIONS deletes all rows matching an overwrite_filter expression, then appends the new rows produced by the model. This is useful when you need to reprocess specific partitions without affecting the rest of the table.

class ReprocessedRows(bauplan.TableSchema):
    """The reprocessed rows; `year` is the declared partition column."""

    year: bauplan.Int64

@bauplan.model(
    materialization_strategy='OVERWRITE_PARTITIONS',
    partitioned_by=['year'],
    overwrite_filter='year = 2024',
)
@bauplan.python('3.11')
def my_model(
    input: Annotated[pyarrow.Table, bauplan.Model('upstream')],
) -> Annotated[pyarrow.Table, ReprocessedRows]:
    return input.select(['year'])

Requirements:

partitioned_by must be set on the model decorator.
Every column referenced in overwrite_filter must be a declared partition column.
The filter supports =, !=, <, >, <=, >=, AND, OR, IN, and NOT IN with column references and literal values. Function calls (YEAR(), CAST(), DATE_TRUNC(), etc.) and BETWEEN are not supported.

Using SQL

Many transformations, especially filtering, joining, and basic aggregations, are easier in SQL than in Python.

There are two ways to use SQL in a model. A SQL model lives in its own .sql file as a single, pure SQL query executed by a SQL engine of Bauplan's choice. Inside the Python function of a Python model, SQL can be executed using an embedded SQL engine of your choice such as DuckDB or DataFusion.

In either approach, Bauplan treats the model's output the same - as an Arrow Table - so downstream models don't need to know which language produced it. This also gives you flexibility in handling the query results: a Python model can run SQL directly in its function body, you can pass a SQL model as an input to a separate Python model, or both.

SQL file
Python with DuckDB

The schema lives in models.py:

models.py
class CountByColOne(bauplan.TableSchema):
    """Row counts grouped by `col_1`."""

    col_1: bauplan.String
    count: bauplan.Int64

The query lives in a .sql file. The optional output_schema header ties the query to that schema, the way the return annotation does in a Python model. If output_schema is not specified, then no extra type validation is asserted on the SQL results.

my_model.sql
-- bauplan: output_schema=CountByColOne
SELECT col_1, COUNT(*) AS count
FROM input_table
WHERE timestamp >= '2022-12-15T00:00:00-05:00'
GROUP BY col_1

Bauplan discovers .sql files in the project root and takes the model name from the filename, so this query is a model named my_model. Downstream models read it like any other node, with bauplan.Model('my_model').

See SQL files for the other header options and for including .sql files in subdirectories below the project root.

models.py
class SelectedInputColumns(bauplan.TableSchema):
    """Columns read from the input table."""

    col_1: bauplan.String

class CountByColOne(bauplan.TableSchema):
    """Row counts grouped by `col_1`."""

    col_1: bauplan.String
    count: bauplan.Int64

@bauplan.model()
@bauplan.python("3.11", pip={'duckdb': '1.2.0'})
def my_model(
    data: Annotated[
        pyarrow.Table,
        bauplan.Model(
            "input_table",
            projection_schema=SelectedInputColumns,
            filter="timestamp >= '2022-12-15T00:00:00-05:00'",
        )
    ],
) -> Annotated[pyarrow.Table, CountByColOne]:
    import duckdb

    # DuckDB will convert the Arrow input into a SQL-accessible table
    output_table = duckdb.sql("""
        SELECT col_1, COUNT(*) as count
        FROM data
        GROUP BY col_1
    """).to_arrow_table()

    return output_table

Developing models

The --preview [on | off | head | tail] option can be specified with bauplan run to see a portion of data for each model in the pipeline:

bauplan run --preview head -p <path-to-project>

In addition, print statements in a model body stream back to your terminal while the run is in flight, which is the quickest way to see what a function actually received. Pair them with bauplan run --dry-run to iterate without materializing data to the data lake:

bauplan run --dry-run --preview head -p <path-to-project>

Best practices

Model Inputs

For each model that reads from the lakehouse catalog:
- Define a projection schema to avoid reading unnecessary columns.
- Define a filter when appropriate to reduce reading unnecessary data.

Semantic Layer

In column documentation, mention:

the expected data properties of a column
- do values fall in a range
- have values been filtered
if a column has a numerical type with a qualitative or categorical meaning
if a column will participate in a join or aggregation

In table documentation, from a TableSchema, mention:

if the schema is a projection schema or an output schema
what models the schema is associated with

from typing import Annotated
import bauplan

class SelectedInputColumns(bauplan.TableSchema):
    """
    Projection schema applied to `input_table` containing three columns.

    + ----------- +  SelectedInputColumns   + -------- +
    | input_table | ----------------------> | my_model |
    + ----------- +     (projection)        + -------- +
    """

    col_1: Annotated[bauplan.String, bauplan.TableField(
        doc='aggregation key in `my_model`'
    )]

class CountByColOne(bauplan.TableSchema):
    """
    Output schema for `my_model` containing two columns.
    Row counts grouped by `col_1`.
    """

    col_1: Annotated[bauplan.String, bauplan.TableField(
        doc='aggregation column for `my_model`',
        lineage=SelectedInputColumns["col_1"],
    )]
    count: Annotated[bauplan.Int64, bauplan.TableField(
        doc='row count for each unique value in `col_1`'
    )]

@bauplan.model()
@bauplan.python("3.11")
def my_model(
    data: Annotated[
        pyarrow.Table,
        bauplan.Model(
            "input_table",
            projection_schema=SelectedInputColumns,
            filter="timestamp >= '2022-12-15T00:00:00-05:00'",
        )
    ],
) -> Annotated[pyarrow.Table, CountByColOne]:
    return (
        data.group_by("col_1")
            .aggregate([([], "count_all")])
            .rename_columns(["col_1", "count"])
    )

Model environment Management
- Include only minimal dependencies.
- Pin specific dependency versions, using ==, for consistency.
Code Organization
- Group models that form a pipeline within a single file.
- Separate function bodies into external modules and call them within the Bauplan models. This keeps business logic code neatly separated from the DAG and environment declaration, making code refactoring easier.