Reshape data with a Python transform¶
Let’s take that same stream of raw order rows and reshape it with logic a SQL
SELECT can’t reach: a Python UDF that collapses a dozen free-text channel
spellings — web, Website, WWW, online — into one controlled
value. One transform action, written as the DataFrame function you’d write
anyway.
A SQL transform hands Lakehouse Plumber a SELECT; a Python transform hands
it a function — reach for it when the logic outgrows a SELECT: a UDF, a
multi-source DataFrame join, or a call to an external library.
You could hand-write this: a @dp.temporary_view that reads the source view
with spark.readStream.table(...), calls your function, and returns the
DataFrame — then remember to copy that module into the pipeline package and fix
up its import so it resolves at run time. Or you declare transform_type:
python, point at the module and the function, and let Lakehouse Plumber write
the wrapper, copy the module, and wire the import. That is the whole idea:
declare your ETL, don’t hand-write it.
Before you start¶
A Python transform sits in the middle of a flowgroup: it reads a view a
load produced and hands a new view to a write. This guide continues the
orders ingest from Your first pipeline — a load action that streams raw
CSV into v_orders_raw. Any flowgroup with a load and a write has the same
shape.
Declare the transform¶
Insert one transform action between the load and the write. A Python
transform carries no query. Instead module_path points at a .py file
(relative to the project root), function_name names the function to call,
and parameters is a dict passed straight through to it:
# One flowgroup: stream raw order CSVs, reshape them with a Python transform
# (cast the raw columns and canonicalize the free-text channel with a UDF —
# logic that outgrows a SQL SELECT), then append the result into a silver
# streaming table.
# Action chain: load (cloudFiles CSV) -> transform:python -> write.
pipeline: silver_orders
flowgroup: orders_normalize
actions:
# 1. Load: stream raw order CSVs from the landing volume with Auto Loader.
- name: load_orders_csv
type: load
source:
type: cloudfiles
path: "${landing_path}/orders/*.csv"
format: csv
target: v_orders_raw
# 2. Transform (Python): call normalize_orders() to cast the raw string
# columns and canonicalize the free-text channel with a UDF. The action
# names the module and function; parameters flow through to the function.
- name: normalize_orders
type: transform
transform_type: python
source: v_orders_raw
module_path: transformations/order_transforms.py
function_name: normalize_orders
parameters:
default_channel: unknown
readMode: stream
target: v_orders_normalized
# 3. Write: append the normalized rows into the silver streaming table.
- name: write_orders_silver
type: write
source: v_orders_normalized
write_target:
type: streaming_table
catalog: "${catalog}"
schema: "${silver_schema}"
table: orders
Actions chain by view name, exactly as a SQL transform does:
load_orders_csv produces v_orders_raw, normalize_orders reads it and
produces v_orders_normalized, and the write reads that. readMode: stream
reads the source as a stream, so only new order files are processed each update.
The ${catalog}, ${silver_schema}, and ${landing_path} tokens resolve
per environment from substitutions/dev.yaml, so the same flowgroup ships
unchanged to prod.
Write the transform function¶
The action pointed at transformations/order_transforms.py and
normalize_orders. Here is that module — the code you own:
"""Order-normalization transform for the silver_orders pipeline.
Casts raw CSV order columns to their real types and canonicalizes the
free-text ``channel`` value with a Python UDF — a many-to-one mapping that a
SQL ``SELECT`` cannot express cleanly.
"""
from pyspark.sql import DataFrame, SparkSession
from pyspark.sql.functions import col, udf
from pyspark.sql.types import StringType
# Every spelling the upstream order systems have emitted, mapped to the one
# canonical channel the silver layer stores. Maintaining this as a Python dict
# is the reason this transform is Python and not SQL.
_CHANNEL_ALIASES = {
"web": "web",
"website": "web",
"www": "web",
"online": "web",
"app": "mobile",
"mobile": "mobile",
"ios": "mobile",
"android": "mobile",
"store": "in_store",
"in-store": "in_store",
"instore": "in_store",
"pos": "in_store",
}
def normalize_orders(df: DataFrame, spark: SparkSession, parameters: dict) -> DataFrame:
"""Cast the raw columns and canonicalize the order channel.
Args:
df: source-view rows (raw string columns from the CSV load).
spark: the active SparkSession.
parameters: values from the action's ``parameters:`` block.
"""
default_channel = parameters.get("default_channel", "unknown")
def canonical_channel(raw: str) -> str:
if raw is None:
return default_channel
return _CHANNEL_ALIASES.get(raw.strip().lower(), default_channel)
canonical_channel_udf = udf(canonical_channel, StringType())
return (
df.withColumn("order_id", col("order_id").cast("bigint"))
.withColumn("customer_id", col("cust_id").cast("bigint"))
.withColumn("amount", col("amt").cast("double"))
.withColumn("channel", canonical_channel_udf(col("channel")))
.select("order_id", "customer_id", "amount", "channel")
)
The signature is fixed by the source shape. With a single source view the
function takes (df, spark, parameters); with a list of sources it takes
(dataframes, spark, parameters); with no source — a data generator — it
takes (spark, parameters). The parameters argument arrives as the dict
from the action’s parameters: block, so default_channel is read with
parameters.get(...).
This is the logic a SELECT can’t carry: the channel canonicalization is a
Python dict lookup wrapped in a UDF, with a nested helper. Lakehouse Plumber
copies the whole file, not just normalize_orders — the module-level
_CHANNEL_ALIASES table and the inner canonical_channel helper come
along — so factor logic into helpers and constants freely.
Note
Lakehouse Plumber calls your function; it does not run it at generate time.
It copies the module into the pipeline and wires the import. Edit the
original in transformations/ — never the copy under
generated/.../custom_python_functions/, which carries a DO-NOT-EDIT
header and is overwritten on every generate.
Generate and read the output¶
Validate the flowgroup, then generate the Lakeflow code:
$ lhp validate --env dev
✓ discover (0.01s)
✓ preflight (0.00s)
✓ silver_orders 0 files
✓ validate (0.36s)
1 validated · 0.4s
$ lhp generate --env dev
✓ discover (0.00s)
✓ preflight (0.00s)
✓ silver_orders 1 file
✓ generate (0.34s)
✓ format (0.06s)
✓ monitoring (0.00s)
1 pipeline generated · 1 file · 0.4s
Open the generated file. The transform became a @dp.temporary_view that
reads the source with spark.readStream.table — because you set readMode:
stream — rebuilds your parameters dict, and calls normalize_orders.
The import at the top is wired to the copied module:
# Generated by LakehousePlumber
# Pipeline: silver_orders
# FlowGroup: orders_normalize
from pyspark import pipelines as dp
from custom_python_functions.order_transforms import normalize_orders
# Pipeline Configuration
PIPELINE_ID = "silver_orders"
FLOWGROUP_ID = "orders_normalize"
# ============================================================================
# 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.format", "csv")
.load("/Volumes/dev_catalog/landing/files/orders/*.csv")
)
return df
# ============================================================================
# TRANSFORMATION VIEWS
# ============================================================================
@dp.temporary_view()
def v_orders_normalized():
"""Python transform: order_transforms.normalize_orders"""
# Load source view(s)
v_orders_raw_df = spark.readStream.table("v_orders_raw")
# Apply Python transformation
parameters = {"default_channel": "unknown"}
df = normalize_orders(v_orders_raw_df, spark, parameters)
return df
# ============================================================================
# TARGET TABLES
# ============================================================================
# Create the streaming table
dp.create_streaming_table(
name="dev_catalog.silver.orders", comment="Streaming table: orders"
)
# Define append flow(s)
@dp.append_flow(
target="dev_catalog.silver.orders",
name="f_orders_silver",
comment="Append flow to dev_catalog.silver.orders",
)
def f_orders_silver():
"""Append flow to dev_catalog.silver.orders"""
# Streaming flow
df = spark.readStream.table("v_orders_normalized")
return df
Lakehouse Plumber copied transformations/order_transforms.py to
generated/dev/silver_orders/custom_python_functions/order_transforms.py and
emitted from custom_python_functions.order_transforms import
normalize_orders so the wrapper resolves at run time. The
v_orders_normalized view is wired into the same append flow that feeds
dev_catalog.silver.orders.
What you just did¶
You wrote one transform action — ten lines of YAML naming the module, the
function, and one parameter — and a 54-line Python module that is yours: the
alias table, the UDF, and the casts are exactly what you typed. Lakehouse
Plumber generated the 11-line @dp.temporary_view that reads
v_orders_raw, rebuilds your parameters dict, and calls
normalize_orders; copied your module into the pipeline package with a
DO-NOT-EDIT header; emitted the from custom_python_functions... import; and
wired v_orders_normalized into the append flow. Across the flowgroup: 41
lines of YAML became 69 lines of wired Lakeflow Python, with zero lines of
``@dp`` glue from you.
The function is yours to own — Lakehouse Plumber wrote only the boilerplate that
turns it into a wired-up view, plus the file copying and import rewriting a
hand-written @dp.temporary_view would have left you to do by hand.
What’s next¶
Stay in SQL when the logic fits a SELECT — casts, renames, filters, and stream-static joins don’t need Python, covered in the SQL transform guide.
Gate the rows with a data-quality check — attach expectations to drop or quarantine bad records after the transform, covered in the data-quality guide.
For the full option list — multiple source views,
operational_metadata, and the rules for importing local helper modules — see the Transform action reference.