Ingest from a Delta table¶

Not every source is a pile of raw files. Often the data you need already lives in a Delta table another team publishes in Unity Catalog — a bronze orders table — and your job is to read it and land it downstream in your own silver layer.

You could hand-write spark.readStream.table("catalog.schema.orders"), then a create_streaming_table, then an append_flow to wire them together. Or you declare a type: delta load and let Lakehouse Plumber write that Lakeflow for you. That is the idea on every page: declare your ETL, don’t hand-write it.

Let’s ingest an existing bronze orders Delta table into a silver streaming table, streaming, without touching PySpark.

Before you start¶

You need Lakehouse Plumber installed and a project to work in. If you followed the Get Started course you already have one; otherwise scaffold an empty project first. This guide assumes the bronze orders Delta table already exists in Unity Catalog — a Delta load reads an existing table, it does not create the source.

Declare the Delta load¶

A pipeline in Lakehouse Plumber is a flowgroup: a short YAML file describing a sequence of actions. This one has two — a load that reads the Delta table into a view, and a write that appends the view into a streaming table.

Create pipelines/silver_orders.yaml:

pipelines/silver_orders.yaml¶

# Ingest an existing bronze Delta table and land it in the silver layer.
# Action chain: load (delta, streaming read) -> write (streaming_table).
pipeline: silver_orders
flowgroup: orders_from_delta

actions:
  # 1. Load: stream an existing Delta table from Unity Catalog into a view.
  #    readMode: stream -> spark.readStream.table(...); readMode: batch -> spark.read.table(...).
  - name: load_orders_delta
    type: load
    readMode: stream
    source:
      type: delta
      catalog: "${catalog}"
      schema: "${bronze_schema}"
      table: orders
    target: v_orders_source

  # 2. Write: append the streamed rows into a silver streaming table.
  - name: write_orders_silver
    type: write
    source: v_orders_source
    write_target:
      type: streaming_table
      catalog: "${catalog}"
      schema: "${silver_schema}"
      table: orders

The load action names the source with three parts — catalog, schema, and table — which resolve to the fully qualified ${catalog}.${bronze_schema}.orders. readMode: stream is the field that decides how the table is read: stream generates spark.readStream.table(...) for continuous ingestion, and batch generates spark.read.table(...) for a one-time load. The write action appends the resulting view into a silver streaming table named orders.

The ${...} tokens are placeholders you resolve per environment, so the same flowgroup runs unchanged in dev, staging, and prod.

Point the tokens at your environment¶

Those tokens resolve from a per-environment substitutions file. Create substitutions/dev.yaml:

substitutions/dev.yaml¶

# Development environment tokens.
# Referenced from flowgroups with ${token} syntax.
dev:
  catalog: dev_catalog
  bronze_schema: bronze
  silver_schema: silver

This is the only place environment-specific values live. To read the same bronze table from a production catalog later, you write a prod.yaml with production values — the flowgroup itself never changes.

Generate the pipeline¶

Now compile the YAML into Lakeflow code. Validate first, then generate:

$ 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.40s)
✓ format (0.06s)
✓ monitoring (0.00s)
1 pipeline generated · 1 file · 0.5s

validate resolves every token and checks the actions before you commit to generating; generate writes the Python.

Read what Lakehouse Plumber wrote¶

Open generated/dev/silver_orders/orders_from_delta.py. This is the entire output — nothing is hidden behind a runtime:

generated/dev/silver_orders/orders_from_delta.py¶

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

from pyspark import pipelines as dp

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


# ============================================================================
# SOURCE VIEWS
# ============================================================================


@dp.temporary_view()
def v_orders_source():
    """Delta source: dev_catalog.bronze.orders"""
    df = spark.readStream.table("dev_catalog.bronze.orders")

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

    return df

It’s ordinary Lakeflow. Because you set readMode: stream, the source view is a @dp.temporary_view wrapping spark.readStream.table("dev_catalog.bronze.orders"). The target is a dp.create_streaming_table, and a @dp.append_flow streams the view into it. You described a source table and a target table; LHP wrote the three pieces of plumbing that connect them.

Switch to a batch read¶

Streaming is the default path for continuous ingestion, but the same load reads a one-time snapshot when you change one field. Set readMode: batch (or drop the field — batch is the default) and LHP generates spark.read.table(...) in place of spark.readStream.table(...). Everything else in the flowgroup stays the same. That is the declarative payoff: you change intent in one line, not the generated Python.

Note

A Delta load reads a table that already exists in Unity Catalog. It does not create the source table — that is the job of whatever pipeline writes it. If the table is missing at run time, the pipeline fails when it runs, not when LHP generates the code.

What you just did¶

Twenty lines of YAML. Forty-six lines of Lakeflow Python. Zero lines of PySpark from you — you never wrote readStream.table, create_streaming_table, or append_flow. And the output is code you own: version it, diff it, and open it in the Databricks editor like anything else.

The payoff compounds. Reading a second Delta table, switching to a batch snapshot, or adding a transform is a few more lines of YAML — not another forty-six lines of Python each time.

What’s next¶

Read the Change Data Feed. The same Delta source reads row-level changes (inserts, updates, deletes) instead of the current table state — set options: {readChangeFeed: "true"} on the source and pair it with a starting bound. See the CDC patterns guide.
See every Delta option. options also covers time travel (versionAsOf / timestampAsOf), ignoreDeletes, skipChangeCommits, and more. The load action reference lists them all with their constraints.
Add a transform and a quality gate. Reshape the rows and drop or quarantine bad records before they reach the silver table.