> The complete documentation index is at [llms.txt](/llms.txt). 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](/concepts/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. ```python 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. :::note[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. ```python type:ignore @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. ```python #! from typing import Annotated 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. ```python 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](#overwrite-partitions). | 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. ```python #! from typing import Annotated 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. The schema lives in `models.py`: ```python title="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. ```sql title="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](/common-scenarios/sql-files) for the other header options and for including `.sql` files in subdirectories below the project root. ```python title="models.py" #! from typing import Annotated 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: ```sh type:ignore bauplan run --preview head -p ``` 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: ```sh type:ignore bauplan run --dry-run --preview head -p ``` ## Best practices 1. **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. 2. **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 ```python 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"]) ) ``` 3. **Model environment Management** - Include only minimal dependencies. - Pin specific dependency versions, using `==`, for consistency. 4. **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.