Medallion Architecture with Partitioning

This page walks through a PySpark job that moves data through the Bronze, Silver and Gold layers of a Delta Lake medallion architecture and controls partitioning at each step. The code below is the original sketch; the sections after it explain what each partitioning choice does, fix the bugs in the sketch, and give a runnable version.

spark = SparkSession.builder \
  .appName("Medallion Architecture") \
  .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
  .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
  .config("spark.executor.instances", "256") \
  .config("spark.sql.shuffle.partitions", "1024") \
  .getOrCreate()

# Paths for the Bronze, Silver, and Gold layers
bronze_path = "/mnt/delta/bronze"
silver_path = "/mnt/delta/silver"
gold_path   = "/mnt/delta/gold"

# ----------- Bronze Layer (Raw Data Ingestion) ----------------------------- #
raw_data = spark.read().format("json").load("/mnt/raw_data/")
raw_data = raw_data.repartition(1024)
raw_data.write().format("delta").mode("overwrite").save(bronze_path)


# ----------- Silver Layer (Data Cleaning and Transformation) --------------- #
bronze_df = spark.read().format("delta").load(bronze_path)

# Filter data for today and the last 365 days
silver_df = bronze_df.filter(
    (col("date") >= date_sub(current_date(), 365)) & (col("date") <= current_date())
)

silver_df = silver_df.dropDuplicates().filter(col("status").isNotNull())
silver_df = silver_df.repartition(365, col("date"))
silver_df.write().format("delta").mode("overwrite").save(silver_path)


# ----------- Gold Layer (Aggregated and Optimized Data) ------------------ #
silver_df = spark.read().format("delta").load(silver_path)

# Perform aggregations for reporting.
gold_df   = silver_df.groupBy("category") 
                     .agg( 
                          count("order_id").alias("total_orders"), 
                          sum("sales_amount").alias("total_sales") 
                         ) 
gold_df = gold_df.repartition(100, col("category"), col("date"))
gold_df.write().format("delta").mode("overwrite").save(gold_path)

spark.stop()
    

PySpark Aggregation Tables

Data Description

Dataset containing three columns: category, order_id, and sales_amount. The dataset is grouped by the category column, and we perform aggregation to count the total number of orders and sum up the sales amounts for each category.

Original Dataset

Aggregated Result

After performing the aggregation, the dataset is grouped by category, and we compute two new columns: total_orders and total_sales.

How the Layers and Partitioning Fit

Bronze stores raw records as they arrived, with minimal transformation, so the pipeline can be replayed. Silver applies data quality: deduplication, null filtering, type casting and conforming keys. Gold holds business-level aggregates sized for dashboards and models. Partitioning decisions differ per layer. Bronze is usually append-only and partitioned by ingestion date. Silver tables that are filtered by event date benefit from date partitioning or clustering. Gold tables are often small enough that partitioning hurts more than it helps, because many partitions produce many tiny files.

A useful rule of thumb from Databricks guidance is to partition only tables that are at least around a terabyte, and to choose a column whose partitions will each hold at least about a gigabyte of data. Below that, the planning overhead of many directories and the cost of small files outweigh the benefit of pruning, and file-level statistics (min and max values per column, which Delta records automatically) already let the engine skip most irrelevant files. The aggregation tables at the end of this page show the logical result of the Gold step.

repartition vs partitionBy

The original code uses repartition() everywhere, which only redistributes rows across Spark tasks before the write. repartition(365, col("date")) hash-partitions by date into 365 tasks, so each output file tends to hold one day, but the table on disk is still a flat directory and readers cannot prune by date. To get partition pruning, write with .partitionBy("date"), which creates one directory per value. On current Databricks runtimes, liquid clustering (CLUSTER BY) is the recommended alternative for most new tables: it avoids over-partitioning and can be changed later without rewriting the table.

Issues in the Original Sketch

• spark.read() and df.write() are properties in PySpark, not methods: use spark.read.format(...) and df.write.format(...).
• The multi-line groupBy(...).agg(...) chain needs parentheses or backslashes to be valid Python.
• The Gold step repartitions by date, but the aggregate only groups by category, so date no longer exists; either group by both or drop it.
• dropDuplicates() with no columns compares whole rows; usually you deduplicate on a business key such as order_id.
• Imports for SparkSession and the functions (col, count, sum, date_sub) are missing, and importing sum shadows Python's built-in.

Corrected, Runnable Version

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.appName("medallion").getOrCreate()
bronze, silver, gold = "/mnt/delta/bronze", "/mnt/delta/silver", "/mnt/delta/gold"

# Bronze: land raw JSON, partitioned by ingest date
(spark.read.format("json").load("/mnt/raw_data/")
      .withColumn("ingest_date", F.current_date())
      .write.format("delta").mode("append")
      .partitionBy("ingest_date").save(bronze))

# Silver: last 365 days, typed, deduplicated on the business key
silver_df = (spark.read.format("delta").load(bronze)
    .withColumn("date", F.to_date("date"))
    .filter(F.col("date").between(F.date_sub(F.current_date(), 365), F.current_date()))
    .filter(F.col("status").isNotNull())
    .dropDuplicates(["order_id"]))
(silver_df.repartition("date")
    .write.format("delta").mode("overwrite")
    .partitionBy("date").save(silver))

# Gold: daily totals per category, small table, no partitioning
gold_df = (spark.read.format("delta").load(silver)
    .groupBy("category", "date")
    .agg(F.count("order_id").alias("total_orders"),
         F.sum("sales_amount").alias("total_sales")))
gold_df.coalesce(8).write.format("delta").mode("overwrite").save(gold)

repartition("date") before partitionBy("date") sends each day's rows to one task, so each date directory gets one file rather than one file per task. coalesce(8) reduces the Gold output to a handful of files without a full shuffle.

Choosing a Layout

Pitfalls

• Small files. 1024 shuffle partitions on a modest dataset produce thousands of tiny files; enable adaptive query execution and run OPTIMIZE.
• Overwrite semantics. mode("overwrite") replaces the whole table; use replaceWhere or MERGE for incremental loads.
• Hard-coded executors. Setting 256 executor instances in code fights cluster autoscaling; configure it on the cluster.

Related Topics

• Bronze Layer, Silver Layer, Gold Layer
• Liquid Clustering
• Column shuffle and repartition

Featured Videos