Enforce a schema on a view¶

Let’s take a raw orders view — untyped string columns, cryptic names, and a stray src_system column no one downstream wants — and lock it to an explicit contract: exact column names, exact types, exact column list, in a fixed order.

You could hand-write that contract as a SQL transform: a SELECT with a CAST(... AS ...) AS ... for every column, re-typed inline and re-derived each time the raw shape drifts. Or you declare transform_type: schema, keep the contract in one schema file, and let Lakehouse Plumber write the renames, the casts, and the select. That is the idea on every page: declare your ETL, don’t hand-write it.

Before you start¶

A schema transform sits in the middle of a flowgroup: it reads a view a load produced and hands a typed 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. Any flowgroup with a load and a write fits the same shape.

Write the schema contract¶

Put the column contract in its own file. Each line is an arrow mapping — source_column -> target_name: TYPE — so the schema reads as a table of what goes in and what comes out. Create schemas/orders.yaml:

schemas/orders.yaml¶

# Explicit column contract for the orders view, in arrow format:
#   source_column -> target_name: TYPE
# Under strict enforcement every column NOT listed here is dropped, and the
# output columns are selected in exactly this order.
columns:
  - "o_id      -> order_id: BIGINT"
  - "c_id      -> customer_id: BIGINT"
  - "gross_amt -> amount: DOUBLE"
  - "o_ts      -> order_ts: TIMESTAMP"

Each entry renames a cryptic raw column to a clear name and casts it to a target type in one line. The arrow format has three other forms — old -> new renames without a cast, col: TYPE casts without a rename, and a bare col keeps a column as-is — but a clean source -> target: TYPE per column is all this contract needs. The file is a versioned artifact in its own right: diff it, review it, and reuse it across flowgroups that share the shape.

Declare the transform¶

Insert one transform action between the load and the write. Give it transform_type: schema, point source at the view to reshape, point schema_file at the contract, and set enforcement: strict:

pipelines/orders_typed.yaml¶

# Guide: enforce an explicit schema on a view.
# Action chain: load (cloudFiles CSV, streaming) -> transform:schema (strict
# enforcement from an external schema file) -> write (streaming table).
pipeline: silver_orders
flowgroup: orders_typed

actions:
  # 1. Load: stream raw CSV order files from the landing volume with Auto Loader.
  #    Raw columns are untyped strings with cryptic names, plus a src_system
  #    column the downstream table has no use for.
  - name: load_orders_csv
    type: load
    source:
      type: cloudfiles
      path: "${landing_path}/orders/*.csv"
      format: csv
    target: v_orders_raw

  # 2. Transform (schema): apply the explicit column contract in
  #    schemas/orders.yaml — rename each cryptic column to a clear name, cast it
  #    to its target type, and (strict) drop every column not listed.
  - name: enforce_orders_schema
    type: transform
    transform_type: schema
    source: v_orders_raw
    schema_file: "schemas/orders.yaml"
    enforcement: strict
    target: v_orders_typed

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

enforcement: strict is the field that makes this a contract rather than a set of edits: only the columns named in the schema survive, every unlisted column (the src_system cruft) is dropped, and the output columns come out in the file’s order. The one-sentence contrast with a sql transform: there you would retype the CAST(... AS ...) list and the surviving-column set inside the query every time; here the contract lives in one file the transform reads.

Note

Under strict enforcement the schema columns must exist in the source — a missing declared column fails the pipeline at run time, so a silent shape change surfaces as an error rather than a wrong table. Omit enforcement (or set permissive, the default) to cast and rename the listed columns while letting every other column pass through unchanged.

Generate the pipeline¶

Validate the flowgroup, then generate the Lakeflow code:

$ lhp validate --env dev
✓ discover (0.01s)
✓ preflight (0.00s)
✓ silver_orders  0 files
✓ validate (0.38s)
1 validated · 0.4s

$ lhp generate --env dev
✓ discover (0.01s)
✓ preflight (0.00s)
✓ silver_orders  1 file
✓ generate (0.38s)
✓ 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/dev/silver_orders/orders_typed.py¶

# Generated by LakehousePlumber
# Pipeline: silver_orders
# FlowGroup: orders_typed

from pyspark.sql import functions as F
from pyspark.sql.types import StructType
from pyspark import pipelines as dp

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


# ============================================================================
# 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_typed():
    """Schema application: enforce_orders_schema"""
    df = spark.readStream.table("v_orders_raw")

    # Apply column renaming
    df = df.withColumnRenamed("o_id", "order_id")
    df = df.withColumnRenamed("c_id", "customer_id")
    df = df.withColumnRenamed("gross_amt", "amount")
    df = df.withColumnRenamed("o_ts", "order_ts")

    # Apply type casting
    df = df.withColumn("order_id", F.col("order_id").cast("BIGINT"))
    df = df.withColumn("customer_id", F.col("customer_id").cast("BIGINT"))
    df = df.withColumn("amount", F.col("amount").cast("DOUBLE"))
    df = df.withColumn("order_ts", F.col("order_ts").cast("TIMESTAMP"))

    # Strict schema enforcement - select only specified columns
    # Schema-defined columns (will fail if missing)
    columns_to_select = ["order_id", "customer_id", "amount", "order_ts"]

    df = df.select(*columns_to_select)

    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_typed")

    return df

The schema transform became a @dp.temporary_view that turns each contract line into concrete PySpark: four withColumnRenamed calls map the cryptic names, four withColumn(...).cast(...) calls set the types, and — because enforcement is strict — a final df.select(...) keeps only the four declared columns, in the declared order. src_system is never selected, so it never reaches the silver table. The typed view is then wired into the append flow that feeds dev_catalog.silver.orders.

What you just did¶

You wrote one action — 7 lines of YAML pointing at a 5-line schema file — and Lakehouse Plumber generated the 24-line @dp.temporary_view: the four renames, the four casts, and the select that enforces the column list and order. Across the whole flowgroup: 30 lines of YAML became 83 lines of Lakeflow Python, with zero lines of PySpark from you. You never wrote withColumnRenamed, a .cast(), or a .select() by hand.

The contract is the payoff, not just the line count. Names, types, column set, and order all live in one reviewable file — change the shape there and the generated view follows, and a source that drifts out of contract fails loudly instead of leaking the wrong columns downstream.

What’s next¶

Reach for a SQL transform when the reshape needs real logic — a join, a filter, a computed column — beyond renaming and casting a fixed column list. The SQL transform guide covers that case and the stream(...) rule.
Gate the typed rows with a data-quality check — attach expectations to drop or quarantine bad records after the schema is enforced, covered in the data-quality guide.
For inline schemas (schema_inline), the permissive mode, the full arrow syntax, and readMode, see the Transform action reference.