On this page
Parameterize a template for many tables¶
A template shares an action sequence across flowgroups. It declares parameters
under parameters and uses {{ parameter }} placeholders in its actions.
A flowgroup names it with use_template and supplies template_parameters.
This guide builds a bronze-ingest template for two tables: orders as clustered
CSV and customers as JSON. Start with a working pipeline
and read Apply shared defaults with a preset if you want to share reader and table settings.
Create templates/ and presets/ directories if they do not exist.
See the worked reuse diagram for the resulting orders and customers flowgroups and their preset values.
Declare the parameters¶
A template’s parameters block declares what varies per table. Create
templates/bronze_ingest.yaml with three of them:
# The ingestion SHAPE captured once: stream a folder of files from the landing
# volume with Auto Loader into a view, then append them to a bronze streaming
# table. Three parameters drive what varies per table.
#
# The body is plain YAML; Jinja2 {{ param }} expressions are rendered inside
# string values. `entity` is required; `file_format` and `cluster_columns`
# have defaults, so a flowgroup only supplies them when it wants to override.
name: bronze_ingest
version: "1.0"
description: "Stream a folder of files into a bronze streaming table; format and clustering set per table."
presets:
- bronze_defaults
parameters:
- name: entity
required: true
description: "Entity name; drives the landing folder, the view name and the target table."
- name: file_format
required: false
default: csv
description: "File format under the entity's landing folder. Defaults to csv."
- name: cluster_columns
required: false
default: []
description: "Liquid-clustering columns for the bronze table. Empty means unclustered."
actions:
# Load: stream files for this entity from the landing volume.
- name: load_{{ entity }}_raw
type: load
source:
type: cloudfiles
path: "${landing_path}/{{ entity }}/*.{{ file_format }}"
format: "{{ file_format }}"
target: v_{{ entity }}_raw
# Write: append the raw rows into the entity's bronze streaming table.
- name: write_{{ entity }}_bronze
type: write
source: v_{{ entity }}_raw
write_target:
type: streaming_table
catalog: "${catalog}"
schema: "${bronze_schema}"
table: "{{ entity }}"
cluster_columns: "{{ cluster_columns }}"
Each parameter is a name plus a few rules. entity is marked required:
true — every flowgroup must supply it, and omitting it stops generation with a
clear error before any Python is written. file_format and cluster_columns
each carry a default, which makes them optional: a flowgroup that says
nothing about them gets csv and no clustering. A parameter is optional unless required: true is set. Required parameters
must be supplied explicitly even if they also declare a default.
You do not need to declare a parameter’s type; the engine does not enforce
type metadata. Lists and objects render as native collections, integers can
become integers, and true/false becomes a boolean. Decimal results remain
strings, and numeric-looking strings can become integers. That is why cluster_columns can default
to an empty list [] and render into a real cluster_by list downstream,
not the string "[]".
The template also carries presets: [bronze_defaults], so every table it
stamps inherits the same Auto Loader reader options and streaming-table
properties. You set the bronze standard once, in the template, for all of them.
Note
Templating fills {{ }} first; the ${...} tokens are left untouched
until later. So {{ entity }} and {{ file_format }} resolve during
template expansion, and ${catalog} / ${bronze_schema} /
${landing_path} resolve afterwards, per environment — the same template
ships unchanged to dev, staging, and prod.
Stamp tables that differ¶
Each table is a flowgroup that names the template and supplies its parameters.
orders overrides cluster_columns — a native YAML list — and relies on the
file_format default:
# orders: a high-volume CSV table.
# Overrides cluster_columns (a list) and layers the high_throughput preset on
# top of the template's bronze_defaults. Relies on the file_format default (csv).
pipeline: bronze
flowgroup: orders_ingest
use_template: bronze_ingest
presets:
- high_throughput
template_parameters:
entity: orders
cluster_columns:
- order_id
- order_date
customers overrides a different parameter — file_format — and relies on
the cluster_columns default, so it needs no list at all:
# customers: a JSON table, left unclustered.
# Overrides file_format (a different parameter than orders touches) and relies
# on the cluster_columns default (empty), so no preset is layered on.
pipeline: bronze
flowgroup: customers_ingest
use_template: bronze_ingest
template_parameters:
entity: customers
file_format: json
Note the two different keys. The template declares parameters under
parameters; a flowgroup supplies them under template_parameters. Each
flowgroup names only what makes its table different — one word for the entity,
plus the one or two parameters it wants to change from the defaults.
Layer a preset on one table¶
Look again at orders_ingest.yaml: alongside use_template it lists its own
presets: [high_throughput]. Template-level presets apply first; flowgroup-level
presets apply second and win on conflicts. So high_throughput can raise a
value the template’s bronze_defaults already set — for orders only:
# Flowgroup-level preset: raise the Auto Loader trigger batch for a high-volume
# table. Applied AFTER the template's bronze_defaults, so it overrides the
# maxFilesPerTrigger that preset set.
name: high_throughput
version: "1.0"
description: "Raise the Auto Loader trigger batch size for a high-volume table."
defaults:
load_actions:
cloudfiles:
options:
cloudFiles.maxFilesPerTrigger: 1000
bronze_defaults sets cloudFiles.maxFilesPerTrigger to 200 for every table
the template stamps. orders layers high_throughput on top, which sets it
to 1000. customers layers nothing, so it keeps 200. The template holds the
shared standard; a flowgroup tunes one table without editing the template or
touching the other tables.
Generate and compare the output¶
Validate first, then generate the Lakeflow code:
$ lhp validate --env dev
✓ discover (0.01s)
✓ preflight (0.00s)
✓ bronze 0 files
✓ validate (0.36s)
1 validated · 0.4s
$ lhp generate --env dev
✓ discover (0.01s)
✓ preflight (0.00s)
✓ bronze 2 files
✓ generate (0.40s)
✓ format (0.05s)
✓ monitoring (0.00s)
1 pipeline generated · 2 files · 0.5s
Two flowgroups in, two complete Lakeflow modules out. Open orders — it is the
full pipeline, with the layered preset and the clustering baked in, and not a
{{ }} in sight:
# Generated by LakehousePlumber
# Pipeline: bronze
# FlowGroup: orders_ingest
from pyspark import pipelines as dp
# Pipeline Configuration
PIPELINE_ID = "bronze"
FLOWGROUP_ID = "orders_ingest"
# ============================================================================
# SOURCE VIEWS
# ============================================================================
@dp.temporary_view()
def v_orders_raw():
"""Load data from csv files at /Volumes/dev_catalog/landing/files/orders/*.csv"""
df = (
spark.readStream.format("cloudFiles")
.option("cloudFiles.rescuedDataColumn", "_rescued_data")
.option("cloudFiles.inferColumnTypes", "true")
.option("cloudFiles.maxFilesPerTrigger", 1000)
.option("cloudFiles.format", "csv")
.load("/Volumes/dev_catalog/landing/files/orders/*.csv")
)
return df
# ============================================================================
# TARGET TABLES
# ============================================================================
# Create the streaming table
dp.create_streaming_table(
name="dev_catalog.bronze.orders",
comment="Streaming table: orders",
table_properties={"delta.enableChangeDataFeed": "true", "quality": "bronze"},
cluster_by=["order_id", "order_date"],
)
# Define append flow(s)
@dp.append_flow(
target="dev_catalog.bronze.orders",
name="f_orders_bronze",
comment="Append flow to dev_catalog.bronze.orders",
)
def f_orders_bronze():
"""Append flow to dev_catalog.bronze.orders"""
# Streaming flow
df = spark.readStream.table("v_orders_raw")
return df
customers came from the same template, yet the module differs everywhere its
parameters did:
# Generated by LakehousePlumber
# Pipeline: bronze
# FlowGroup: customers_ingest
from pyspark import pipelines as dp
# Pipeline Configuration
PIPELINE_ID = "bronze"
FLOWGROUP_ID = "customers_ingest"
# ============================================================================
# SOURCE VIEWS
# ============================================================================
@dp.temporary_view()
def v_customers_raw():
"""Load data from json files at /Volumes/dev_catalog/landing/files/customers/*.json"""
df = (
spark.readStream.format("cloudFiles")
.option("cloudFiles.rescuedDataColumn", "_rescued_data")
.option("cloudFiles.inferColumnTypes", "true")
.option("cloudFiles.maxFilesPerTrigger", 200)
.option("cloudFiles.format", "json")
.load("/Volumes/dev_catalog/landing/files/customers/*.json")
)
return df
# ============================================================================
# TARGET TABLES
# ============================================================================
# Create the streaming table
dp.create_streaming_table(
name="dev_catalog.bronze.customers",
comment="Streaming table: customers",
table_properties={"delta.enableChangeDataFeed": "true", "quality": "bronze"},
)
# Define append flow(s)
@dp.append_flow(
target="dev_catalog.bronze.customers",
name="f_customers_bronze",
comment="Append flow to dev_catalog.bronze.customers",
)
def f_customers_bronze():
"""Append flow to dev_catalog.bronze.customers"""
# Streaming flow
df = spark.readStream.table("v_customers_raw")
return df
Read the three lines that moved. orders has
cloudFiles.maxFilesPerTrigger at 1000 — the layered high_throughput
preset overriding the template’s 200 — while customers keeps 200. orders
reads cloudFiles.format "csv" (the default) and customers reads
"json" (its override). And orders carries
cluster_by=["order_id", "order_date"] while customers, on the empty-list
default, has no cluster_by at all. One template produced two genuinely
different modules; the only lines that changed are the ones each flowgroup chose
to change.
What you just did¶
Two flowgroups — sixteen lines of parameters between them — generated 111 lines
of Lakeflow Python across two modules from one template. The load-and-write
shape is written once. Defaults keep the common case terse — customers is six
lines of parameters — while per-flowgroup overrides and a layered preset handle
the exceptions without forking the template.
A third table is another short flowgroup: override the one or two parameters that make it different, not another fifty-odd lines of Python you would keep in sync by hand. That is how one template scales across a schema’s worth of tables.
What’s next¶
Make a scalar expression vary. An eligible action string containing
{{ ... }}may also use Jinja2 conditions, loops and filters. The engine evaluates individual string values; it does not expand whole-document loops or conditionally add/remove action entries. See the templates reference for the rendering and conversion rules.See every field. The templates reference lists the full parameter and template file schema, and how to organize templates in subdirectories (
use_template: ingestion/bronze_ingest).Scale past one pipeline. A template stamps tables into one pipeline; a blueprint stamps whole pipelines — across regions, tenants, or environments — from one definition.