How to create a Delta Live Table in Lakehouse: step by step

This tutorial shows how to create a Delta Live Table in Lakehouse for continuous ingestion and data transformation, useful when you need reliable pipelines with history and monitoring. The task focuses on a concrete example: reading JSON files from a directory, applying simple transformations and publishing a Delta table ready for consumption.

Prerequisites

• Account with access to Microsoft Fabric and permissions to create a Lakehouse and pipelines.
• A Lakehouse with a directory for ingestion files (e.g.: /lakehouse/raw/events/).
• Basic knowledge of Python/PySpark and the Delta format.
• Example JSON files with a consistent schema.

Step 1: Understand what a Delta Live Table in Lakehouse is

Delta Live Table is an approach to build declarative pipelines that produce managed Delta tables. It enables continuous ingestion, schema handling and integrated monitoring. This step is conceptual: visualize the flow — source (JSON files) → transformation (cleaning, types) → destination (Delta table).

Step 2: Create a pipeline workspace

Create a pipeline in the Fabric environment that runs PySpark code on a continuous or scheduled trigger. Define the name, runtime type and the Lakehouse where the tables will be written.

# Exemplo conceptual (interface do Fabric tem gui); se usares CLI/SDK configura o pipeline com:
# runtime: pyspark
# target: Lakehouse//tables/

Step 3: Minimal PySpark code for a Delta Live Table

Write a PySpark script that reads JSON in streaming mode, applies simple transformations and writes to a Delta table. Keep the code minimal to test the flow.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, to_timestamp

spark = SparkSession.builder.getOrCreate()

# Fonte: ficheiros JSON em modo 'cloudFiles' (auto-infer schema quando disponível)
raw_df = (
    spark.readStream
         .format("cloudFiles")
         .option("cloudFiles.format", "json")
         .option("cloudFiles.inferColumnTypes", "true")
         .load("/lakehouse/raw/events/")
)

# Transformações: limpar campos e converter timestamps
clean_df = (
    raw_df
    .withColumn("event_time", to_timestamp(col("event_time"), "yyyy-MM-dd'T'HH:mm:ss"))
    .withColumn("user_id", col("user_id").cast("string"))
    .na.drop(subset=["event_time", "user_id"])  # descartar linhas inválidas
)

# Escrita em Delta para o Lakehouse (modo append, checkpoint obrigatório)
(output = clean_df.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/lakehouse/checkpoints/events_checkpoint/")
    .start("/lakehouse/tables/events_delta/"))

Step 4: Configure schema policies and error handling

Define how the pipeline handles schema changes and corrupt data. In Fabric/Lakehouse environments, configure Auto Loader or cloudFiles options to skip corrupt files and log errors.

# Opções adicionais para robustez (exemplo cloudFiles)
.option("cloudFiles.schemaEvolutionMode", "addNewColumns")
.option("cloudFiles.maxFilesPerTrigger", "100")
.option("badRecordsPath", "/lakehouse/errors/bad_records/")

Step 5: Validate and create the managed Delta table

After the stream writes to the Delta path, register the folder as a Delta table in the Lakehouse catalog for SQL querying and consumption by Power BI.

-- SQL executado no SQL endpoint ou notebook para criar tabela a partir do caminho Delta
CREATE TABLE IF NOT EXISTS lakehouse.events_delta
USING DELTA
LOCATION '/lakehouse/tables/events_delta/';

Step 6: Common operations and errors to avoid

Monitor checkpointLocation to avoid state loss; do not share the same checkpoint between different pipelines. If you encounter schema mismatch errors, enable schemaEvolution or perform pre-validation. Avoid writing directly to the same directory with concurrent jobs without ACID (Delta partially solves this, but has limits).

# Erros comuns (consulta de diagnóstico)
-- Verifica o estado da tabela
DESCRIBE HISTORY lakehouse.events_delta;
-- Conferir ficheiros corruptos
ls /lakehouse/errors/bad_records/

Verify the result

Confirm that the Delta table received data: query the table with SQL and verify there are recent records. Check for the existence of the checkpoint and the _delta_log files in the table directory.

SELECT COUNT(*) FROM lakehouse.events_delta;
SELECT event_time, user_id FROM lakehouse.events_delta ORDER BY event_time DESC LIMIT 10;
# No storage, confirma:
# /lakehouse/tables/events_delta/_delta_log/
# /lakehouse/checkpoints/events_checkpoint/

Conclusion

You now have a basic Delta Live Table pipeline in Lakehouse for continuous ingestion of JSON, minimal transformation and publishing as a Delta table. Next steps: add quality tests, enrichments and monitoring in Fabric. Tip: start with a small set of files to validate the schema before enabling streaming in production.