Reshape data with a SQL transform

Let’s take a stream of raw order rows — string columns, cryptic names, no context — and turn them into typed, enriched, table-ready records. One transform action, written as the SQL SELECT you’d write anyway.

You could hand-write this: a @dp.temporary_view function that wraps spark.sql(...), remembering to wrap the streaming source in stream(...), then wire that view into the append flow that feeds the table. Or you declare transform_type: sql, hand Lakehouse Plumber the query, and let it write the wrapper and the wiring. That is the whole idea: declare your ETL, don’t hand-write it.

Before you start

A SQL 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 extends the orders ingest from Your first pipeline — a load action that streams raw CSV into v_orders_raw. If you have any flowgroup with a load and a write, the same shape applies.

Declare the transform

Insert one transform action between the load and the write. Give it a transform_type: sql, point source at the view to read, name the target view it produces, and put the query in sql:

pipelines/orders_enrich.yaml
# Guide: reshape data with a SQL transform.
# Action chain: load (cloudFiles CSV, streaming) -> transform:sql (stream-static
# enrichment join) -> write (streaming table). One flowgroup, no data quality.
pipeline: silver_orders
flowgroup: orders_enrich

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. Transform (SQL): cast the raw string columns, rename them, filter out
  #    rows with no order id, and enrich each order by joining the STREAMING
  #    orders view against the STATIC customer dimension table.
  - name: enrich_orders
    type: transform
    transform_type: sql
    source: v_orders_raw
    sql: |
      SELECT
        CAST(o.order_id AS BIGINT)    AS order_id,
        CAST(o.cust_id  AS BIGINT)    AS customer_id,
        c.customer_name,
        c.region,
        CAST(o.amt      AS DOUBLE)    AS amount,
        CAST(o.order_ts AS TIMESTAMP) AS order_ts
      FROM stream(v_orders_raw) o
      LEFT JOIN ${catalog}.${silver_schema}.dim_customers c
        ON CAST(o.cust_id AS BIGINT) = c.customer_id
      WHERE o.order_id IS NOT NULL
    target: v_orders_enriched

  # 3. Write: append the enriched rows into the silver streaming table.
  - name: write_orders_silver
    type: write
    source: v_orders_enriched
    write_target:
      type: streaming_table
      catalog: "${catalog}"
      schema: "${silver_schema}"
      table: orders_enriched

Actions chain by view name: load_orders_csv produces v_orders_raw, enrich_orders reads it and produces v_orders_enriched, and the write reads that. The query does four things at once — it casts the raw string columns to BIGINT / DOUBLE / TIMESTAMP, renames cust_id to customer_id, filters out rows with no order_id, and joins each order against a customer dimension to add customer_name and region. The ${catalog}, ${silver_schema}, and ${landing_path} tokens resolve per environment from substitutions/dev.yaml, so the same query ships unchanged to prod.

The query is inline here. To keep a long query in its own file instead, drop the sql key and point sql_path at a .sql file — the same ${...} tokens work inside it. The Transform action reference covers sql_path, multiple source views, and operational-metadata columns; this guide stays on the inline single-source case.

Choose streaming or static per source

Look at the FROM clause. The orders view is wrapped in stream(...); the customer dimension is referenced as a plain table name. That difference is deliberate, and it is the one rule worth internalising for SQL transforms.

Lakehouse Plumber passes your query straight into spark.sql(...) — it does not rewrite your FROM clauses. So each source’s read semantics are decided inside the SQL:

  • Wrap a source in stream(<view>) for an incremental, streaming read. Here stream(v_orders_raw) keeps the pipeline processing only new order files as they land.

  • Reference a source plainly — a view name, or a fully-qualified catalog.schema.table — for a static snapshot read. Here the dim_customers table is read as it stands at each update.

That is a stream-static join: an unbounded stream of orders enriched by a slowly-changing dimension read as a snapshot. Forget the stream(...) wrapper on a streaming source and the query silently runs in batch, reprocessing everything each update.

Note

stream(...) is a Databricks SQL construct, not Lakehouse Plumber syntax. You are writing ordinary Lakeflow SQL — Lakehouse Plumber only decides where it lands and how the view is wired in.

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.33s)
1 validated · 0.3s

$ lhp generate --env dev
✓ discover (0.01s)
✓ preflight (0.00s)
✓ silver_orders  1 file
✓ generate (0.34s)
✓ format (0.05s)
✓ monitoring (0.00s)
1 pipeline generated · 1 file · 0.4s

Open the generated file. The transform became a @dp.temporary_view that wraps your query in spark.sql(...) — with the tokens resolved and the stream(...) / static split preserved exactly as you wrote it:

generated/dev/silver_orders/orders_enrich.py
# Generated by LakehousePlumber
# Pipeline: silver_orders
# FlowGroup: orders_enrich

from pyspark import pipelines as dp

# Pipeline Configuration
PIPELINE_ID = "silver_orders"
FLOWGROUP_ID = "orders_enrich"


# ============================================================================
# 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(comment="SQL transform: enrich_orders")
def v_orders_enriched():
    """SQL transform: enrich_orders"""
    df = spark.sql("""SELECT
  CAST(o.order_id AS BIGINT)    AS order_id,
  CAST(o.cust_id  AS BIGINT)    AS customer_id,
  c.customer_name,
  c.region,
  CAST(o.amt      AS DOUBLE)    AS amount,
  CAST(o.order_ts AS TIMESTAMP) AS order_ts
FROM stream(v_orders_raw) o
LEFT JOIN dev_catalog.silver.dim_customers c
  ON CAST(o.cust_id AS BIGINT) = c.customer_id
WHERE o.order_id IS NOT NULL""")

    return df


# ============================================================================
# TARGET TABLES
# ============================================================================

# Create the streaming table
dp.create_streaming_table(
    name="dev_catalog.silver.orders_enriched",
    comment="Streaming table: orders_enriched",
)


# Define append flow(s)
@dp.append_flow(
    target="dev_catalog.silver.orders_enriched",
    name="f_orders_silver",
    comment="Append flow to dev_catalog.silver.orders_enriched",
)
def f_orders_silver():
    """Append flow to dev_catalog.silver.orders_enriched"""
    # Streaming flow
    df = spark.readStream.table("v_orders_enriched")

    return df

The ${catalog}.${silver_schema}.dim_customers reference resolved to dev_catalog.silver.dim_customers — a static read, no stream() — while stream(v_orders_raw) came through untouched. The view is wired into the same append flow that feeds dev_catalog.silver.orders_enriched.

What you just did

You wrote one action — an 11-line SQL SELECT wrapped in six lines of YAML — and Lakehouse Plumber generated the 16-line @dp.temporary_view function, wrapped your query in spark.sql(...), resolved the environment tokens, and wired v_orders_enriched into the append flow. Across the whole flowgroup: 35 lines of YAML became 74 lines of Lakeflow Python, with zero lines of PySpark from you. You never wrote @dp.temporary_view, spark.sql, or a readStream.table by hand.

The SQL is yours to own — the casts, the join, the filter are exactly what you typed. Lakehouse Plumber wrote only the boilerplate that turns a query into a wired-up Lakeflow view.

What’s next

  • 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.

  • Reach for a Python transform when the logic outgrows a SELECT — UDFs, external calls, or multi-source DataFrame code, covered in the Python transform guide.

  • For the full option list — sql_path, multiple source views, and operational metadata — see the Transform action reference.