On this page

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. Quotes and backslashes reach Spark unchanged, so a pattern such as RLIKE '^\\d+$' needs no extra escaping. Only whitespace changes: the query is stripped of leading and trailing whitespace (including blank lines before and after it), and whitespace at the end of each line is removed when the file is written. 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.

Syntax and options

See SQL transform (sql) for all supported settings and defaults.