====================================== Reshape data with a SQL transform ====================================== .. meta:: :description: Reshape rows between a load and a write with a Lakehouse Plumber SQL transform — cast types, rename columns, filter, and enrich with a stream-static join, all from one declarative action. 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``: .. literalinclude:: ../../_fixtures/guide_transform_sql/pipelines/orders_enrich.yaml :language: yaml :caption: pipelines/orders_enrich.yaml :emphasize-lines: 20-36 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()`` 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: .. code-block:: console $ 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: .. literalinclude:: ../../_fixtures/guide_transform_sql/generated/dev/silver_orders/orders_enrich.py :language: python :caption: generated/dev/silver_orders/orders_enrich.py :emphasize-lines: 34-49 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 :doc:`/reference/actions/transform/sql` for all supported settings and defaults.