Base classes

After this page, you will know which base class to use for Spark, Pandas, or Polars jobs in bolt-pipeliner==0.2.6. You will also know how to register your own base class with a dotted Python path.

All class names and import behavior here are verified against bolt_pipeliner/runner.py and the modules under bolt_pipeliner/bases/.

Built-in base classes

These are the five built-in values accepted in class_name.

`class_name`	Engine	Storage target
`ETLBase`	Spark	Iceberg table at `<save_catalog>.<fixed_schema>.<layer>_<output_table_name>`
`ETLBaseDelta`	Spark	Delta table at `<save_catalog>.<layer>_<output_table_name>`
`ETLBaseParquet`	Spark	Parquet dataset at `<output_location>/<layer>_<output_table_name>.parquet`
`ETLBaseParquetPandas`	Pandas	Parquet dataset directory at `<output_location>/<layer>_<output_table_name>`
`ETLBaseParquetPolars`	Polars	Parquet dataset directory at `<output_location>/<layer>_<output_table_name>`

Example job entries:


bronze:
  - module: bronze_orders
    class_name: ETLBase
    input_tables:
      orders: flatfile_orders
    output_table_name: orders_bronze
 
silver:
  - module: silver_orders
    class_name: ETLBaseParquetPandas
    input_tables:
      orders: bronze_orders_bronze
    output_table_name: orders_silver

How to choose

Pick ETLBase when you run Spark with Iceberg catalogs and want table based reads and writes.
Pick ETLBaseDelta when your Spark environment writes Delta tables.
Pick ETLBaseParquet when you want Spark transformations and Parquet files on disk or object storage.
Pick ETLBaseParquetPandas for lightweight local pipelines without Spark.
Pick ETLBaseParquetPolars for a lightweight local path with Polars DataFrames.

Register a custom base class

In 0.2.6, runner._resolve_base_class supports two patterns:

Built in names from the table above.
A dotted path such as mypkg.bases.CustomBase.

If the value is not built in and does not contain a dot, the run fails with a KeyError.

Create a base class module in your project:


# etl/custom_bases.py
from bolt_pipeliner.bases.pandas_parquet import ETLBaseParquetPandas
 
 
class ETLBaseParquetPandasWithAudit(ETLBaseParquetPandas):
    def unload_data(self, processed_df):
        if processed_df is not None and not processed_df.empty:
            processed_df = processed_df.assign(load_source="daily_batch")
        super().unload_data(processed_df)

Reference it in etl_config.yaml:


gold:
  - module: gold_orders
    class_name: etl.custom_bases.ETLBaseParquetPandasWithAudit
    input_tables:
      orders: silver_orders_silver
    output_table_name: orders_mart

Run it:


bolt run --config configs/etl_config.yaml --verbose