Incremental processing

After this page, you will be able to choose the right incremental mode for each job and configure incremental_* keys correctly. You will also know when to use manual partition writes with unload: false.

All behavior here is verified against bolt_pipeliner/runner.py and incremental logic in bolt_pipeliner/bases/_incremental.py for bolt-pipeliner==0.2.6.

Turn incremental on

Incremental behavior is job level. Set incremental: true on a job to enable policy based writes.


silver:
  - module: silver_orders
    class_name: ETLBaseParquetPandas
    input_tables:
      orders: bronze_orders_clean
    output_table_name: orders_features
    incremental: true
    incremental_column: year_month
    incremental_type: int
    incremental_unit: 3

If incremental is omitted, the runner passes False to the base class.

Incremental modes

incremental_unit controls the write mode:

overwrite: replace the full target table
append: insert only new incremental values not already in the target
Positive integer N: window mode, rewrite the latest N values plus any newer incoming values

The package also accepts integer aliases:

-1 means overwrite
0 means append

Any value lower than -1 fails with a validation error.

Type and grain constraints

incremental_type must be one of:

int
date

When incremental_type: date, incremental_date_grain must be one of yearly, monthly, or daily.

Accepted aliases are normalized:

year becomes yearly
month becomes monthly
day becomes daily

Validation rules in 0.2.6:

int: values must be whole numbers
date with yearly: each value must be January 1
date with monthly: each value must be the first day of the month
date with daily: any parseable date is valid

Return value requirements

For incremental jobs in append or window mode, the returned DataFrame must include the incremental column. If the column is missing, the run fails.

For overwrite, the policy writes the returned DataFrame as is, so incremental column validation is not required.

Manual partition unloading with `unload: false`

Use this pattern when you need full control over write timing and partition replacement, for example in a memory heavy Spark Iceberg job.

Set unload: false in YAML so the base class skips automatic unload:


gold:
  - module: gold_orders
    class_name: ETLBase
    input_tables:
      orders: silver_orders_features
    output_table_name: orders_mart
    partition_by: [year_month]
    incremental: true
    incremental_column: year_month
    incremental_type: int
    incremental_unit: append
    unload: false

Then write partitions manually inside process_data and return an empty DataFrame:


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