Stage an intermediate result in a temp table¶
Let’s take a cleaned stream of orders and stage it once, so two downstream branches — a silver append and a high-value flag — both read the same result instead of each re-running the cast-and-filter query.
Chain a few transform_type: sql views together and every consumer re-runs
the whole chain: a view is a query definition, not stored rows. You could hand
that off to a @dp.table(temporary=True) you write yourself, remembering the
temporary=True flag and the read. Or you declare transform_type:
temp_table, and Lakehouse Plumber writes the materialised table for you. That
is the idea on every page: declare your ETL, don’t hand-write it.
Before you start¶
A temp table sits in the middle of a flowgroup, between the actions that produce
its input and the actions that read its output. This guide assumes a load
that streams raw CSV orders into v_orders_raw — the same ingest you built in
Get Started. Any flowgroup with a view to stage and more than one action that
reads it fits the same shape.
Declare the temp table¶
Insert one transform action with transform_type: temp_table. Point
source at the view to materialise, name the target table it produces,
and — because the raw view is a stream — declare readMode: stream and wrap
the source in stream(...) inside the query. The sql does the heavy work
once: cast the raw string columns, and drop rows with no order id.
# Stage the cleaned orders once in a temp table, then fan out to two writes.
# Action chain: load (cloudFiles CSV) -> transform:temp_table -> two consumers:
# (A) write the staged rows to silver.orders
# (B) flag high-value orders off the SAME staged rows -> write silver.orders_flagged
pipeline: silver_stage
flowgroup: orders_stage
actions:
# 1. Load: stream raw CSV order files 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. Temp table: materialise the cleaned, typed orders ONCE. Both downstream
# branches read this result instead of re-running the cast-and-filter query.
- name: stage_orders
type: transform
transform_type: temp_table
source: v_orders_raw
readMode: stream
sql: |
SELECT
CAST(order_id AS BIGINT) AS order_id,
CAST(cust_id AS BIGINT) AS customer_id,
CAST(region AS STRING) AS region,
CAST(amt AS DOUBLE) AS amount,
CAST(order_ts AS TIMESTAMP) AS order_ts
FROM stream(v_orders_raw)
WHERE order_id IS NOT NULL
target: orders_staged
# 3a. Branch A: append the staged rows into the silver orders table.
- name: write_orders_silver
type: write
source: orders_staged
write_target:
type: streaming_table
catalog: "${catalog}"
schema: "${silver_schema}"
table: orders
# 3b. Branch B: flag high-value orders off the SAME staged rows, then write them.
- name: flag_high_value
type: transform
transform_type: sql
source: orders_staged
sql: |
SELECT *, amount >= 1000 AS is_high_value
FROM stream(orders_staged)
target: v_orders_flagged
- name: write_orders_flagged
type: write
source: v_orders_flagged
write_target:
type: streaming_table
catalog: "${catalog}"
schema: "${silver_schema}"
table: orders_flagged
Actions chain by name. stage_orders produces orders_staged, and two
actions read it: write_orders_silver appends it to the silver orders
table, and flag_high_value runs a second SELECT over it to mark
high-value orders before its own write. That reuse is the reason to reach for a
temp table — more on it below. The sql is optional: drop it and the action
becomes a straight passthrough materialisation of source.
Note
The stream(...) rule from the SQL transform applies here too. Lakehouse
Plumber hands your query straight to spark.sql(...), so wrap a streaming
source in stream(<view>) to keep the staged table streaming. Reference a
source plainly for a static snapshot read.
Generate the pipeline¶
Validate first, then generate the Lakeflow code:
$ lhp validate --env dev
✓ discover (0.01s)
✓ preflight (0.00s)
✓ silver_stage 0 files
✓ validate (0.31s)
1 validated · 0.3s
$ lhp generate --env dev
✓ discover (0.01s)
✓ preflight (0.00s)
✓ silver_stage 1 file
✓ generate (0.36s)
✓ format (0.05s)
✓ monitoring (0.00s)
1 pipeline generated · 1 file · 0.4s
Read what Lakehouse Plumber wrote¶
Open the generated file. This is the entire output — nothing is hidden behind a runtime:
# Generated by LakehousePlumber
# Pipeline: silver_stage
# FlowGroup: orders_stage
from pyspark import pipelines as dp
# Pipeline Configuration
PIPELINE_ID = "silver_stage"
FLOWGROUP_ID = "orders_stage"
# ============================================================================
# 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.table(
temporary=True,
)
def orders_staged():
"""Temporary table: orders_staged"""
df = spark.sql("""SELECT
CAST(order_id AS BIGINT) AS order_id,
CAST(cust_id AS BIGINT) AS customer_id,
CAST(region AS STRING) AS region,
CAST(amt AS DOUBLE) AS amount,
CAST(order_ts AS TIMESTAMP) AS order_ts
FROM stream(v_orders_raw)
WHERE order_id IS NOT NULL
""")
return df
@dp.temporary_view(comment="SQL transform: flag_high_value")
def v_orders_flagged():
"""SQL transform: flag_high_value"""
df = spark.sql("""SELECT *, amount >= 1000 AS is_high_value
FROM stream(orders_staged)""")
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("orders_staged")
return df
# Create the streaming table
dp.create_streaming_table(
name="dev_catalog.silver.orders_flagged", comment="Streaming table: orders_flagged"
)
# Define append flow(s)
@dp.append_flow(
target="dev_catalog.silver.orders_flagged",
name="f_orders_flagged",
comment="Append flow to dev_catalog.silver.orders_flagged",
)
def f_orders_flagged():
"""Append flow to dev_catalog.silver.orders_flagged"""
# Streaming flow
df = spark.readStream.table("v_orders_flagged")
return df
Look at what the temp table became versus its neighbours. orders_staged is a
@dp.table(temporary=True) — a table Lakeflow computes and stores. The
sibling SQL transform v_orders_flagged two functions down is a
@dp.temporary_view() — a query definition with no stored rows. That is the
whole distinction: a temporary view is re-evaluated by every reader; a temp
table materialises the rows once. Both downstream branches then read the stored
table — the silver append flow calls spark.readStream.table("orders_staged")
and the flag view reads stream(orders_staged) — so the cast-and-filter query
runs once, not once per branch.
When to reach for a temp table¶
A transform_type: sql view is the default middle-of-flowgroup transform, and
it is the right call when one action reads the result. Reach for temp_table
when the trade-off tips:
Several downstream actions read the same intermediate. Materialising it once keeps the upstream query from being recomputed for each consumer — the two-branch case in this guide.
A heavy DAG needs a checkpoint. Staging a costly join or aggregation as a table breaks a long view chain into a materialised step the rest of the flowgroup builds on.
The temp table is dropped when the pipeline completes — it is scratch space for
the run, not a published table. When a single action consumes the result, stay
with a sql view and skip the materialisation.
What you just did¶
You declared one action — a 15-line temp_table transform — and Lakehouse
Plumber generated the 16-line @dp.table(temporary=True) function, resolved
the environment tokens, and wired the staged table into both downstream
branches. Across the whole flowgroup: 49 lines of YAML became 102 lines of
Lakeflow Python, with zero lines of PySpark from you. You never wrote
@dp.table, temporary=True, or a readStream.table by hand.
The payoff is structural, not just line count. The cast-and-filter query is written once, materialised once, and read by two actions — change it in one place and both branches follow.
What’s next¶
Gate the staged rows with a data-quality check — attach expectations to drop or quarantine bad records before they reach a table, covered in the data-quality guide.
Keep the transform a plain view when only one action reads it — the SQL transform guide covers the default case and the
stream(...)rule in full.For every
temp_tablefield —sql,readMode, and the passthrough form — see the Transform action reference.