Learning: build a lakehouse on Azure Databricks with PySpark DataFrames (full loads, incremental MERGE and Jobs)

#About this session

By the end you will have one reusable notebook that can load any table, and a job that runs it for many tables in the right order.

#Module 1: the ideas you need first

#DataFrames in one paragraph

A Spark DataFrame is a distributed table with a schema. You describe what you want (read, filter, join, write) and Spark plans how to run it across the cluster. Nothing runs until an action (write, count, display) is called. In Scala there is also the typed Dataset API; in Python you always work with DataFrames.

#How the work is spread out

Example: 10 partitions on 8 cores run as 8 tasks, then 2 more. More partitions than cores means work comes in waves.

#The medallion layers

#Module 2: set up the lab

#Choose compute

Classic clusters take a few minutes to start; serverless starts in seconds.

#Create the catalog objects

Run in a SQL cell (replace main with a catalog you can use):

CREATE SCHEMA IF NOT EXISTS main.lab_bronze;
CREATE SCHEMA IF NOT EXISTS main.lab_silver;
CREATE VOLUME IF NOT EXISTS main.lab_bronze.landing;

The volume is where raw files land. Upload the sample files (Catalog Explorer → the volume → Upload to this volume) with this layout:

/Volumes/main/lab_bronze/landing/
  dim/product/product.csv
  dim/customer/customer.csv
  fact/internet_sales/20261020/internet_sales.csv

The files are pipe-delimited (|) with no header row. Any AdventureWorks export works; Microsoft publishes the AdventureWorksDW sample data.

Note

Older workspaces without Unity Catalog used dbutils.fs.mount with a service principal to expose ADLS as /mnt/.... Mounts are a legacy pattern; on Unity Catalog use volumes or external locations instead. The DataFrame code in this session is the same either way, only the paths change.

Check the upload:

display(dbutils.fs.ls("/Volumes/main/lab_bronze/landing/dim/product/"))

#Module 3: a reusable full-load notebook

Create a notebook called nb_full_load. Each cell below is one step.

#Step 1: parameters (widgets)

Widgets make the notebook reusable: the job passes different values for each table.

dbutils.widgets.text("source_dir", "dim/product")
dbutils.widgets.text("target_table", "main.lab_silver.dim_product")
dbutils.widgets.text("schema_ddl", "ProductKey INT, ProductAlternateKey STRING, ProductSubcategoryKey INT, "
                     "EnglishProductName STRING, StandardCost DECIMAL(19,4), Color STRING, "
                     "ListPrice DECIMAL(19,4), StartDate TIMESTAMP, EndDate TIMESTAMP, Status STRING")
dbutils.widgets.text("num_files", "2")

LANDING    = "/Volumes/main/lab_bronze/landing"
source_dir = dbutils.widgets.get("source_dir").strip("/")
target     = dbutils.widgets.get("target_table")
schema_ddl = dbutils.widgets.get("schema_ddl")
num_files  = int(dbutils.widgets.get("num_files"))
src_path   = f"{LANDING}/{source_dir}/"
print(src_path, "->", target)

#Step 2: read with an explicit schema

df = (spark.read
      .format("csv")
      .option("sep", "|")
      .option("header", "false")
      .option("quote", "")                 # the files don't quote text fields
      .option("recursiveFileLookup", "true")
      .option("mode", "FAILFAST")          # stop on malformed rows while learning
      .schema(schema_ddl)
      .load(src_path))

display(df.limit(5))

Why an explicit schema: with no header, Spark would name columns _c0, _c1, ..., and inferSchema reads the data twice and can guess types wrongly (for example, a product code that looks numeric).

#Step 3: add load metadata

from pyspark.sql import functions as F

df = (df
      .withColumn("_load_ts", F.current_timestamp())
      .withColumn("_source_file", F.col("_metadata.file_path")))

_metadata.file_path is the supported way to get the source file on Unity Catalog compute (input_file_name() is not available on all access modes). With these two columns you can always answer "which file did this row come from and when was it loaded?".

#Step 4: write a Delta table

(df.repartition(num_files)
   .write
   .format("delta")
   .mode("overwrite")
   .option("overwriteSchema", "true")
   .saveAsTable(target))

print(f"{spark.table(target).count()} rows in {target}")

saveAsTable creates a managed Unity Catalog table: no folder paths to manage, no separate CREATE TABLE ... LOCATION step. repartition(n) controls how many files are written; for small tables, fewer files are better.

#Step 5: return a result to the caller

dbutils.notebook.exit(f"loaded {target}")

Run the notebook by hand once with the defaults, then change the widgets to load dim/customer into main.lab_silver.dim_customer (with its own schema string).

#Module 4: look at what you created

DESCRIBE DETAIL  main.lab_silver.dim_product;   -- format, location, numFiles, sizeInBytes
DESCRIBE HISTORY main.lab_silver.dim_product;   -- every write as a version
SELECT _source_file, count(*) FROM main.lab_silver.dim_product GROUP BY _source_file;

Catalog Explorer shows the same information in its Overview, Sample data, Details and History tabs. Run the notebook again and look at DESCRIBE HISTORY: a new version appears, and older versions can still be read:

SELECT count(*) FROM main.lab_silver.dim_product VERSION AS OF 0;

#Module 5: orchestrate many tables

#The old way: a parent notebook

tables = [
  {"source_dir": "dim/product",  "target_table": "main.lab_silver.dim_product",  "schema_ddl": "..."},
  {"source_dir": "dim/customer", "target_table": "main.lab_silver.dim_customer", "schema_ddl": "..."},
]
for t in tables:
    print(dbutils.notebook.run("./nb_full_load", 600, t))

This works, but dependencies, retries, alerts and run history all have to be coded by hand, and everything runs one after another.

#The better way: a job with tasks

Create a job (Jobs & Pipelines → Create job) with one notebook task per table, each passing its parameters, and dependencies where order matters. As an Asset Bundle definition:

resources:
  jobs:
    lab_lakehouse_load:
      name: lab-lakehouse-load
      tasks:
        - task_key: dim_product
          notebook_task:
            notebook_path: ../src/nb_full_load.py
            base_parameters: { source_dir: dim/product, target_table: main.lab_silver.dim_product, schema_ddl: "ProductKey INT, ..." }
        - task_key: dim_customer
          notebook_task:
            notebook_path: ../src/nb_full_load.py
            base_parameters: { source_dir: dim/customer, target_table: main.lab_silver.dim_customer, schema_ddl: "CustomerKey INT, ..." }
        - task_key: fact_internet_sales
          depends_on: [ { task_key: dim_product }, { task_key: dim_customer } ]
          notebook_task:
            notebook_path: ../src/nb_incremental_load.py
            base_parameters: { source_dir: fact/internet_sales, target_table: main.lab_silver.fact_internet_sales, merge_keys: "SalesOrderNumber,SalesOrderLineNumber", schema_ddl: "..." }
      email_notifications:
        on_failure: [ data-team@example.com ]
      timeout_seconds: 3600

The two dimension tasks run in parallel, and the fact task waits for both. You also get a run timeline, retries per task, notifications and conditional tasks (If/else) without writing any code for them. Without a new_cluster or job cluster, tasks run on serverless where it is enabled.

#Module 6: the incremental pattern

Fact files arrive daily in date-named folders (fact/internet_sales/20261021/). Reloading everything every day gets slower as data grows; instead, merge only the newest folder into the table.

Create nb_incremental_load from a copy of nb_full_load and add a merge_keys widget. Then replace steps 2–4 with the following.

#Read only the newest folder

import re

folders = sorted(f.name.strip("/") for f in dbutils.fs.ls(src_path) if re.fullmatch(r"\d{8}/", f.name))
latest  = folders[-1]
print("Latest folder:", latest)

df = (spark.read.format("csv").option("sep", "|").option("header", "false").option("quote", "")
      .schema(schema_ddl).load(f"{src_path}{latest}/")
      .withColumn("_load_ts", F.current_timestamp())
      .withColumn("_source_file", F.col("_metadata.file_path"))
      .withColumn("_folder_date", F.to_date(F.lit(latest), "yyyyMMdd")))

#First run: create the table

if not spark.catalog.tableExists(target):
    df.write.format("delta").saveAsTable(target)
    dbutils.notebook.exit(f"created {target} from {latest}")

#Every later run: MERGE

from delta.tables import DeltaTable

keys = [k.strip() for k in dbutils.widgets.get("merge_keys").split(",")]
condition = " AND ".join(f"t.{k} = s.{k}" for k in keys)

before = spark.table(target).count()
(DeltaTable.forName(spark, target).alias("t")
   .merge(df.alias("s"), condition)
   .whenMatchedUpdateAll()
   .whenNotMatchedInsertAll()
   .execute())
after = spark.table(target).count()

print(f"rows in file: {df.count()}, table before: {before}, after: {after}")

Exact numbers for the run are also in the table history:

display(DeltaTable.forName(spark, target).history(5)
        .select("version", "operation",
                F.col("operationMetrics.numTargetRowsInserted").alias("inserted"),
                F.col("operationMetrics.numTargetRowsUpdated").alias("updated")))

#Test it

Create a folder 20261021 with a two-line file: one line with an existing order number and a changed amount (an update) and one with a new order number (an insert). Run the notebook. Expect: inserted 1, updated 1, and the table count up by 1.

Warning

MERGE fails if two source rows match the same target row ("multiple source rows matched"). If a file can contain the same key twice, keep the latest row first, for example with row_number() over the key ordered by a modified-date column, and filter to row 1.

#Limits of this simple pattern

#Exercises

1. Load dim/customer with the full-load notebook by changing only widgets.
2. Add a third dimension (dim/date) as a new task in the job, and make the fact task depend on it.
3. Write a gold table: total sales amount per product category per month from fact_internet_sales and dim_product, as a SQL task at the end of the job.
4. Make the incremental notebook process all unprocessed folders, using a control table main.lab_silver._load_control (table_name, last_folder).
5. Break a file on purpose (add an extra | to one line) and see what FAILFAST does. Then try mode = PERMISSIVE with columnNameOfCorruptRecord.

#Knowledge check

1. Why give the CSV reader an explicit schema instead of inferSchema?
2. What does _metadata.file_path give you, and why add it to every row?
3. What is the difference between saveAsTable on Unity Catalog and writing to a path plus CREATE TABLE ... LOCATION?
4. Name two things a job gives you that a parent notebook with dbutils.notebook.run does not.
5. Why must each source row match at most one target row in a MERGE?

#Summary

• Every load notebook is either full (overwrite) or incremental (MERGE into existing data); start with full, add incremental when size demands it.
• Explicit schemas, metadata columns and Delta tables make loads predictable and traceable.
• Parameters turn one notebook into a loader for every table; jobs handle order, retries and alerts.
• Next step in the learning path: Auto Loader for continuous file ingestion, then Lakeflow declarative pipelines.

#Further reading

• This session follows the lab flow of John Miner's MSSQLTips article Engineering a Lakehouse with Azure Databricks with Spark Dataframes, updated here for Unity Catalog volumes and managed tables. The original uses Hive metastore and ADLS mounts, and includes a downloadable notebook bundle.
• Databricks documentation: Upsert into a Delta Lake table using merge, File metadata column, Configure task dependencies in jobs.