Generate and inspect¶

Now compile the YAML into Lakeflow code and read what comes out. This is where Lakehouse Plumber earns its keep: you wrote declarative YAML; it writes the PySpark.

Validate first¶

validate resolves every ${...} token and checks the actions without writing anything. On a bundle project you pass the pipeline config with -pc:

$ lhp validate --env dev -pc config/pipeline_config.yaml
✓ discover (0.01s)
✓ preflight (0.00s)
✓ sample_gold  0 files
✓ sample_ingest  0 files
✓ sample_lakehouse_event_log_monitoring  0 files
✓ sample_silver  0 files
✓ validate (0.38s)
4 validated · 0.7s

Four pipelines: the three you explored, plus a monitoring pipeline Lakehouse Plumber adds automatically because lhp.yaml enables monitoring. 0 files means nothing is written — validate only checks.

Note

-pc is required on a bundle project. Without it, generate stops at preflight with LHP-CFG-023 rather than emit half-configured resources. Non-bundle projects don’t need it.

Generate¶

$ lhp generate --env dev -pc config/pipeline_config.yaml
✓ discover (0.01s)
✓ preflight (0.00s)
✓ sample_gold  1 file
✓ sample_ingest  3 files
✓ sample_lakehouse_event_log_monitoring  1 file
✓ sample_silver  3 files
✓ generate (0.41s)
✓ format (0.06s)
✓ monitoring (0.00s)
✓ bundle_sync (0.02s)
4 pipelines generated · 8 files · 0.8s

Eight Python files under generated/dev/, one per flowgroup, plus the bundle resources under resources/lhp/. The format step runs the output through a formatter; bundle_sync writes the bundle resources.

Read what Lakehouse Plumber wrote¶

Open generated/dev/sample_ingest/orders_ingest.py — the whole output of that eight-line flowgroup:

generated/dev/sample_ingest/orders_ingest.py¶

# Generated by LakehousePlumber
# Pipeline: sample_ingest
# FlowGroup: orders_ingest

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

# Pipeline Configuration
PIPELINE_ID = "sample_ingest"
FLOWGROUP_ID = "orders_ingest"


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

# Schema hints for orders table
orders_schema_hints = """
    o_orderkey BIGINT,
    o_custkey BIGINT,
    o_totalprice DECIMAL(15,2),
    o_orderdate DATE,
    received_at TIMESTAMP
""".strip().replace("\n", " ")


@dp.temporary_view()
def v_orders_raw():
    """Load data from json files at /Volumes/main/sample_bronze/landing/orders/"""
    df = (
        spark.readStream.format("cloudFiles")
        .option("cloudFiles.inferColumnTypes", "true")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.schemaHints", orders_schema_hints)
        .load("/Volumes/main/sample_bronze/landing/orders/")
    )

    # Add operational metadata columns
    df = df.withColumn("_processing_timestamp", F.current_timestamp())
    df = df.withColumn("_source_file_path", F.col("_metadata.file_path"))
    df = df.withColumn("_source_file_size", F.col("_metadata.file_size"))

    return df


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

# Create the streaming table
dp.create_streaming_table(
    name="main.sample_bronze.orders_raw",
    comment="Streaming table: orders_raw",
    table_properties={"delta.enableChangeDataFeed": "true", "quality": "bronze"},
)


# Define append flow(s)
@dp.append_flow(
    target="main.sample_bronze.orders_raw",
    name="f_orders_bronze",
    comment="Append flow to main.sample_bronze.orders_raw",
)
def f_orders_bronze():
    """Append flow to main.sample_bronze.orders_raw"""
    # Streaming flow
    df = spark.readStream.table("v_orders_raw")

    return df

It’s ordinary Lakeflow: a @dp.temporary_view for the Auto Loader source, a dp.create_streaming_table for the target, and a @dp.append_flow wiring them together. Your schema hints became a typed DDL string; your operational_metadata list became three withColumn calls; your bronze_layer preset became the table_properties. Nothing is hidden behind a runtime — you could have written this by hand.

The CDC dimension is just as plain. Open generated/dev/sample_silver/dim_customer.py:

generated/dev/sample_silver/dim_customer.py¶

# ============================================================================

# Create the streaming table for CDC
dp.create_streaming_table(
    name="main.sample_silver.dim_customer",
    comment="Streaming table: dim_customer",
    table_properties={"quality": "silver"},
)

# CDC flow: f_dim_customer
dp.create_auto_cdc_flow(
    target="main.sample_silver.dim_customer",
    source="v_customer_changes_clean",
    name="f_dim_customer",
    keys=["c_custkey"],
    sequence_by="_commit_version",
    stored_as_scd_type=2,
    except_column_list=["_change_type", "_commit_version", "_commit_timestamp"],
    apply_as_deletes="_change_type = 'delete'",
)

Your three-key CDC config became a dp.create_auto_cdc_flow call with the keys, the sequencing column, and SCD type 2 — the exact API you’d otherwise have to remember and hand-wire.

What you just did¶

You turned eight flowgroups into eight readable Lakeflow files — zero lines of PySpark from you. And it’s code you own: version it, diff it, open it in the Databricks editor. Change a flowgroup and regenerate, and only the affected files change. That’s the compounding payoff — the pattern is written once, and Lakehouse Plumber repeats the plumbing for every table.

Next¶

The code is on disk. Ship it to a Databricks workspace — Deploy and run.