bauplan_sdk_types
Binary data type corresponding to the Arrow data type Binary.
Bases:
FieldType
Boolean data type corresponding to the Arrow data type Bool.
Bases:
FieldType
Date data type corresponding to the Arrow data type Date32 (days).
Bases:
FieldType
Date data type corresponding to the Arrow data type Date64 (milliseconds).
Bases:
FieldType
Floating point data type corresponding to the Arrow data type Float64.
Bases:
FieldType
Integer data type corresponding to the Arrow data type Int32.
Bases:
FieldType
Integer data type corresponding to the Arrow data type Int64.
Bases:
FieldType
A reference to another Bauplan model (a DAG node) as a data dependency. This identifies the model by catalog identifier, which may be qualified ('namespace.name') or may be bare ('name'). A bare model name will be prefixed with the default namespace.
bauplan.Model(
name: str,
projection_schema: Optional[type[TableSchema]] = None,
filter: Optional[LiteralString] = None,
)
As a simple example, consider two models: (1) a leaf model, leaf_model that reads
from the catalog and (2) a downstream model, my_model, that reads from the leaf
model. In the code below, leaf_model specifies the catalog table it depends on
using Model('src_table'); then, my_model specifies its dependency on the output
of leaf_model using Model('bauplan.leaf_table'). If leaf_model wasn't defined
to have the identifier "bauplan.leaf_table", then its default identifier (its
function name, "leaf_model") would be used instead.
from typing import Annotated
import bauplan
#! my_col: bauplan.Int64
@bauplan.model(name='bauplan.leaf_table')
def leaf_model(
catalog_table: Annotated[pyarrow.Table, bauplan.Model('src_table')],
) -> Annotated[pyarrow.Table, MySchema]:
return catalog_table.select(['my_col'])
@bauplan.model()
def my_model(
leaf_data: Annotated[
pyarrow.Table,
bauplan.Model(name='bauplan.leaf_table')
],
) -> Annotated[pyarrow.Table, MySchema]:
return leaf_data
There are two parameters supported to provide "pushdown" support for projections and selections.
The projection_schema parameter specifies a schema, by identifier, to
use as a projection (select column names to read into result table). Specifying a
projection is important, because it lets the system avoid reading data that the model
doesn't use. For example:
@bauplan.model(name='bauplan.leaf_table')
def leaf_model(
catalog_table: Annotated[
pyarrow.Table,
bauplan.Model('src_table', projection_schema=MySchema),
],
) -> Annotated[pyarrow.Table, MySchema]:
# `projection_schema` is the equivalent of `select(['my_col'])` with validation
return catalog_table
The filter parameter specifies a string-based SQL-like predicate that can be used
to filter table rows from the input table. The predicate must be a string literal;
usage would look like:
bauplan.Model('src_table', filter='bar > 1')
ModelMaterializationStrategy = Literal['NONE', 'REPLACE', 'APPEND', 'OVERWRITE_PARTITIONS']
Represents a parameter that can be used to "template" values passed to a model during a run or
query with, e.g., bauplan run --parameter interest_rate=2.0.
bauplan.Parameter(
param_name: str,
)
Syntax for accessing a parameter uses the init method as an Annotation to
communicate the proper behavior to the Bauplan control plane:
proj_param: Annotated[float, Parameter('interest_rate')]
String data type corresponding to the Arrow data type String.
Bases:
FieldType
A schema field that contains metadata for a table column.
bauplan.TableField(
doc: Optional[str] = None,
lineage: Optional[FieldType | str] = None,
)
A table schema is a collection of table column definitions and is a required base class when specifying a model output schema or a projection schema. A model output schema is specified in the return annotation of a model. A projection schema may be used to project a set of columns from a model input or an expectation input.
For example:
class NewTableSchema(TableSchema):
first: Annotated[Int64, TableField(doc='first column')]
second: Annotated[Float64, TableField(doc='second column')]
References to schema columns can be made by using "index syntax" and passing it the
lineage parameter of TableField, for example:
class DerivedSchema(TableSchema):
third: Annotated[Int64, TableField(lineage=NewTableSchema['first'])]
Time data type corresponding to the Arrow data type Timestamp('us').
Bases:
FieldType
Time data type corresponding to the Arrow data type Timestamp('us', tz='UTC').
Values are absolute instants; Iceberg stores these as UTC and does not preserve
the timezone they were written from.
Bases:
FieldType
Time data type corresponding to the Arrow data type Timestamp('ns').
Bases:
FieldType
Time data type corresponding to the Arrow data type Timestamp('ns', tz='UTC').
Values are absolute instants; Iceberg stores these as UTC and does not preserve
the timezone they were written from.
Bases:
FieldType
Decorator that defines a Bauplan expectation.
def expectation() -> Callable: ...
An expectation is a function from one (or more) dataframe-like object(s) to a boolean: it is commonly used to perform data validation and data quality checks when running a pipeline. Expectations take as input the table(s) they are validating and return a boolean indicating whether the expectation is met or not. Additionally, assert statements in the function body halt the pipeline immediately on failure.
Example
from typing import Annotated
import bauplan
import pyarrow
from bauplan.standard_expectations import expect_column_no_nulls
class AnomalySchema(bauplan.TableSchema):
'''The columns of `join_dataset` this expectation reads.'''
anomaly: Annotated[
bauplan.Bool,
bauplan.TableField(
doc=(
"An example column of some arbitrary datatype, "
"but is expected to not have any null values. "
)
)
],
@bauplan.expectation()
@bauplan.python('3.11')
def test_joined_dataset(
data: Annotated[
pyarrow.Table,
bauplan.Model('join_dataset', projection_schema=AnomalySchema),
],
) -> bool:
# your data validation code here
# ...
# use assertions to stop the pipeline in critical scenarios
assert data.num_rows > 0
return expect_column_no_nulls(data, 'anomaly')
Decorator that specifies a Bauplan model.
def model(
name: Optional[str] = None,
partitioned_by: Optional[Union[str, list[str], tuple[str, ...]]] = None,
materialization_strategy: Optional[ModelMaterializationStrategy] = None,
cache_strategy: Optional[ModelCacheStrategy] = None,
overwrite_filter: Optional[str] = None,
internet_access: Optional[bool] = None,
) -> Callable: ...
A model is a function that defines data transformation logic that takes many Arrow
tables as input and returns a single Arrow table as output. A model can be referenced
by type annotations on function parameters as explicit data dependencies. See
documentation for bauplan.Model for more details on referencing declared models.
Consider the following code example that defines two models, functions decorated with
bauplan.model:
Example
from typing import Annotated
import bauplan
class IotSchema(bauplan.TableSchema):
'''Schema for result of `source_scan`.'''
...
@bauplan.model(materialization_strategy='NONE')
def source_scan(
data: Annotated[
pyarrow.Table,
bauplan.Model('iot_kaggle', filter="motion='false'")
],
) -> Annotated[pyarrow.Table, IotSchema]:
# your code here; schema of output should match `IotSchema`
return data
Decorator that defines a Python environment for a Bauplan function (e.g. a model or expectation). It is used to specify which Python version a model or expectation should run on and which Python packages should be available.
def python(
version: Optional[str] = None,
pip: Optional[dict[str, str]] = None,
) -> Callable: ...