Writing a job

After this page, you will be able to write a valid job module for bolt-pipeliner==0.2.6. You will understand the process_data(self, input_tables) contract and how it changes for Spark, Pandas, and Polars.

All behavior here is verified from bolt_pipeliner/runner.py and the base classes under bolt_pipeliner/bases/.

The required function signature

Each job module must expose this top-level function:

def process_data(self, input_tables):
    ...

runner.py imports your module and binds this function to the active base class instance before calling etl.run().

What self gives you

self is the runtime instance of the base class defined by class_name in your YAML.

Common attributes you can use in all engines:

• self.input_tables: the same mapping passed as input_tables
• self.output_table_name: output table name from job config
• self.partition_by: partition columns from YAML
• self.incremental: job level incremental flag
• self.incremental_column: resolved incremental column name
• self.incremental_policy: resolved incremental mode and options
• self.unload_data(df): write helper used by the base class

Spark base classes (ETLBase, ETLBaseDelta, ETLBaseParquet) also provide:

• self.spark: active SparkSession
• self._create_table(df) and self._replace_table_partitions(df) on ETLBase (Iceberg base)

What input_tables contains

input_tables is a dictionary of already loaded DataFrames keyed by the aliases in your job config.

Example YAML:

silver:
  - module: silver_orders
    class_name: ETLBaseParquetPandas
    input_tables:
      orders: bronze_orders_clean
      customers: bronze_customers_clean
    output_table_name: customer_daily_sales

At runtime, your function receives:

input_tables["orders"]
input_tables["customers"]

Spark example

Use this pattern with ETLBase, ETLBaseDelta, or ETLBaseParquet.

from pyspark.sql import functions as F
 
 
def process_data(self, input_tables):
    orders = input_tables["orders"]
 
    return (
        orders
        .withColumn("order_date", F.to_date("order_ts"))
        .groupBy("customer_id", "order_date")
        .agg(F.sum("amount").alias("gross_amount"))
        .withColumn("year_month", F.date_format("order_date", "yyyyMM").cast("int"))
    )

Pandas example

Use this pattern with ETLBaseParquetPandas.

import pandas as pd
 
 
def process_data(self, input_tables):
    orders = input_tables["orders"].copy()
 
    orders["order_date"] = pd.to_datetime(orders["order_ts"]).dt.date
    out = (
        orders.groupby(["customer_id", "order_date"], as_index=False)["amount"]
        .sum()
        .rename(columns={"amount": "gross_amount"})
    )
    out["year_month"] = pd.to_datetime(out["order_date"]).dt.strftime("%Y%m").astype(int)
    return out

Polars example

Use this pattern with ETLBaseParquetPolars.

import polars as pl
 
 
def process_data(self, input_tables):
    orders = input_tables["orders"]
 
    return (
        orders
        .with_columns(pl.col("order_ts").str.strptime(pl.Datetime, strict=False).alias("order_ts_parsed"))
        .with_columns(pl.col("order_ts_parsed").dt.date().alias("order_date"))
        .group_by(["customer_id", "order_date"])
        .agg(pl.col("amount").sum().alias("gross_amount"))
        .with_columns((pl.col("order_date").dt.year() * 100 + pl.col("order_date").dt.month()).alias("year_month"))
    )

Manual write pattern

If you set unload: false, the base class skips automatic persistence after process_data returns.

For Spark Iceberg jobs with ETLBase, you can write manually in your function:

def process_data(self, input_tables):
    result = input_tables["orders"]
    self._create_table(result)
    return result.limit(0)

For Pandas or Polars jobs, call self.unload_data(result) when you need manual control.