Ingest with a SQL query¶

A Delta load reads one table. But the data you want is often not one table — it is a join or an aggregate across several tables that already live in Unity Catalog. A type: sql load lets you express that as the SELECT you would write anyway, and lands the result as a view your pipeline can write.

You could hand-write it: a @dp.temporary_view function that wraps spark.sql(...), then the wiring that lands the view downstream. Or you declare a type: sql load, hand Lakehouse Plumber the query, and let it write the wrapper and the wiring. That is the idea on every page: declare your ETL, don’t hand-write it.

Let’s ingest a silver orders_enriched table by reading two existing bronze tables — orders and customers — and joining them in one query. No upstream load, no PySpark: the query itself is the source.

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 reads two existing bronze tables — orders and customers — that already live in Unity Catalog. A SQL load reads existing tables; it does not create them.

Declare the SQL load¶

A flowgroup is a short YAML file describing a sequence of actions. This one has two — a load whose source is a SQL query, and a write that lands the query’s result.

Create pipelines/orders_enriched.yaml:

pipelines/orders_enriched.yaml¶

# Ingest guide: use a SQL query as the source. The query reads two existing
# bronze tables directly and joins them into one enriched view, which a write
# lands in the silver layer.
# Action chain: load (sql query) -> write (materialized_view).
pipeline: silver_orders
flowgroup: orders_enriched

actions:
  # 1. Load: read existing tables through a SQL query. source.type: sql runs the
  #    query with spark.sql(...) and exposes the result as the v_orders_enriched
  #    view. The query is the source — there is no upstream load to reshape.
  - name: load_orders_enriched
    type: load
    source:
      type: sql
      sql: |
        SELECT
          o.order_id,
          o.amount,
          o.order_ts,
          c.customer_name,
          c.region
        FROM ${catalog}.${bronze_schema}.orders o
        LEFT JOIN ${catalog}.${bronze_schema}.customers c
          ON o.customer_id = c.customer_id
        WHERE o.order_id IS NOT NULL
    target: v_orders_enriched

  # 2. Write: land the enriched query result in a silver materialized view.
  #    The engine recomputes it from the query view on each refresh.
  - name: write_orders_enriched
    type: write
    source: v_orders_enriched
    write_target:
      type: materialized_view
      catalog: "${catalog}"
      schema: "${silver_schema}"
      table: orders_enriched

The load action’s source is typed sql and carries the query inline under sql. Lakehouse Plumber runs that query with spark.sql(...) and exposes the result as the v_orders_enriched view. The query reads two fully-qualified tables — ${catalog}.${bronze_schema}.orders and ${catalog}.${bronze_schema}.customers — joins them on customer_id, and keeps the rows that have an order_id.

The write lands that view in a silver orders_enriched materialized view. The result is a recomputed join, not an incremental append, so a materialized view is the right target: the engine keeps it current by re-running the query.

The query is inline here. To keep a long query in its own file, drop the sql key and point sql_path at a .sql file — the same ${...} tokens resolve inside it. The load action reference documents sql versus sql_path and the rule that exactly one of them is set.

A load, not a transform¶

A SQL load and a SQL transform both run a query through spark.sql(...), so it is worth being clear on which is which. A SQL transform sits in the middle of a flowgroup: it reads a view an upstream load produced and reshapes it. A SQL load is the entry point: its query reads tables directly, and there is no upstream action to read from.

Here load_orders_enriched names ${catalog}.${bronze_schema}.orders and .customers in its FROM clause, so it belongs at the front of the flowgroup as the source. Reach for a SQL load when your source is a query over existing tables; reach for a SQL transform when you are reshaping a view a load already produced.

Point the tokens at your environment¶

The ${...} 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 tables from a production catalog later, you write a prod.yaml with production values — the query 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.32s)
1 validated · 0.3s

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

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_enriched.py. This is the entire output — nothing is hidden behind a runtime:

generated/dev/silver_orders/orders_enriched.py¶

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

from pyspark.sql import DataFrame
from pyspark import pipelines as dp

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


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


@dp.temporary_view()
def v_orders_enriched():
    """SQL source: load_orders_enriched"""
    df = spark.sql("""SELECT
  o.order_id,
  o.amount,
  o.order_ts,
  c.customer_name,
  c.region
FROM dev_catalog.bronze.orders o
LEFT JOIN dev_catalog.bronze.customers c
  ON o.customer_id = c.customer_id
WHERE o.order_id IS NOT NULL
""")

    return df


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


@dp.materialized_view(
    name="dev_catalog.silver.orders_enriched",
    comment="Materialized view: orders_enriched",
    table_properties={},
)
def orders_enriched():
    """Write to dev_catalog.silver.orders_enriched from multiple sources"""
    # Materialized views use batch processing
    df = spark.read.table("v_orders_enriched")

    return df

Your query became a @dp.temporary_view named v_orders_enriched that wraps spark.sql(...) — with ${catalog} and ${bronze_schema} resolved to dev_catalog.bronze. The write became a @dp.materialized_view that reads that view with a batch spark.read.table(...). You wrote a query and named a target; LHP wrote the view function and the declaration that connect them.

Note

This query is a batch read of the tables it names. To read a source as a stream instead, wrap it in stream(...) inside the query — the same Databricks SQL construct the SQL transform guide covers in depth. A plain table name is read as a batch snapshot.

What you just did¶

Twenty-seven lines of YAML — ten of them the SELECT you would write anyway — compiled to fifty-one lines of Lakeflow Python. Zero lines of PySpark from you: you never wrote @dp.temporary_view, spark.sql, or the @dp.materialized_view declaration. And the output is code you own — version it, diff it, and open it in the Databricks editor like anything else.

Add a second query, join a third table, or point the same query at a production catalog, and each is a few more lines of YAML — not another view function written by hand every time.

What’s next¶

See every SQL-load option. The load action reference documents inline sql versus an external sql_path file, substitution inside .sql files, and the constraint that exactly one of the two is set.
Reshape a view a load produced. When the query reads a view already in the flowgroup rather than tables, that is a SQL transform — covered in the SQL transform guide.
Pull from an external database. When the query has to run against an RDBMS rather than Unity Catalog tables, a JDBC load carries the connection and the query. See the load action reference.