Lakehouse Medallion Bronze-to-Silver ETL

What you will learn

• Construct medallion pipeline validation boundaries between Bronze and Silver layers
• Formulate multi-condition PySpark boolean validation rules with F.col() expressions
• Route invalid records into a dead-letter quarantine table with descriptive error reasons
• Aggregate certified silver metrics without risk of null or invalid data corruption

What you need first

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
spark = SparkSession.builder.master('local[1]').getOrCreate()
df = spark.createDataFrame([(1, None), (2, 50)], ['id', 'val'])
print(df.filter(F.col('val').isNotNull()).count())

1

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
spark = SparkSession.builder.master('local[1]').getOrCreate()
df = spark.createDataFrame([(1, -5), (2, 80)], ['id', 'pct'])
quarantine = df.filter(F.col('pct') < 0)
print(quarantine.first()['id'])

1

Architecture and data flow

The data

Raw IoT sensor telemetry containing valid pings alongside null device IDs and negative battery values.

Step by step

Read each step, run its code, then check your output against the one shown.

1. 01

   Ingest raw bronze IoT sensor telematics DataFrame

   We initialize the raw bronze telemetry DataFrame representing incoming IoT sensor pings. In smart building installations, thousands of edge sensors stream temperature and battery readings every few seconds. Notice that ping p3 contains a null device_id, while ping p4 contains an uncalibrated negative battery percentage of -15. We inspect the raw DataFrame to establish a baseline before implementing validation filters. In lakehouse architectures, Bronze data is preserved immutably in its raw incoming state to maintain a complete historical audit trail.

   PythonStep 1Copy

   from pyspark.sql import SparkSession
   from pyspark.sql import functions as F
   
   spark = SparkSession.builder.master("local[1]").appName("medallion_etl").getOrCreate()
   raw_data = [
       ("p1", "dev_101", 22.5, 85, "2024-03-01 10:00:00"),
       ("p2", "dev_102", 24.1, 90, "2024-03-01 10:00:05"),
       ("p3", None, 21.0, 75, "2024-03-01 10:00:10"),
       ("p4", "dev_103", 19.8, -15, "2024-03-01 10:00:15"),
       ("p5", "dev_101", 23.0, 84, "2024-03-01 10:05:00"),
       ("p6", "dev_102", 24.5, 89, "2024-03-01 10:05:05"),
   ]
   df_bronze = spark.createDataFrame(raw_data, ["ping_id", "device_id", "temp_c", "battery_pct", "timestamp"])
   df_bronze.show(truncate=False)

   Show outputGuess what it prints first

   +-------+---------+------+-----------+-------------------+
   |ping_id|device_id|temp_c|battery_pct|timestamp          |
   +-------+---------+------+-----------+-------------------+
   |p1     |dev_101  |22.5  |85         |2024-03-01 10:00:00|
   |p2     |dev_102  |24.1  |90         |2024-03-01 10:00:05|
   |p3     |NULL     |21.0  |75         |2024-03-01 10:00:10|
   |p4     |dev_103  |19.8  |-15        |2024-03-01 10:00:15|
   |p5     |dev_101  |23.0  |84         |2024-03-01 10:05:00|
   |p6     |dev_102  |24.5  |89         |2024-03-01 10:05:05|
   +-------+---------+------+-----------+-------------------+

   Common mistake

   Modifying raw bronze data in place rather than treating Bronze tables as append-only immutable logs.

   How you will know

   You will see 6 bronze records with null device_id on p3 and -15 battery on p4.

2. 02

   Demonstrate silent data corruption with naive dropna filtering

   Wrong turn first: this step shows the mistake before the fix

   Here is the naive approach developers often attempt first: calling df_bronze.dropna(subset=['device_id']). While dropna successfully removes the null device ping p3, it completely misses the negative battery reading of -15 percent on ping p4. The corrupted battery reading is passed directly into silver tables, polluting average battery metrics and triggering false hardware alarm thresholds. Furthermore, discarded records are lost completely without. We demonstrate this failed attempt to highlight why domain-level validation rules are required.

   PythonStep 2Copy

   from pyspark.sql import SparkSession
   
   spark = SparkSession.builder.master("local[1]").appName("medallion_etl").getOrCreate()
   raw_data = [
       ("p1", "dev_101", 22.5, 85, "2024-03-01 10:00:00"),
       ("p2", "dev_102", 24.1, 90, "2024-03-01 10:00:05"),
       ("p3", None, 21.0, 75, "2024-03-01 10:00:10"),
       ("p4", "dev_103", 19.8, -15, "2024-03-01 10:00:15"),
       ("p5", "dev_101", 23.0, 84, "2024-03-01 10:05:00"),
       ("p6", "dev_102", 24.5, 89, "2024-03-01 10:05:05"),
   ]
   df_bronze = spark.createDataFrame(raw_data, ["ping_id", "device_id", "temp_c", "battery_pct", "timestamp"])
   
   # Naive approach: dropna allows negative battery values through
   naive_silver = df_bronze.dropna(subset=["device_id"])
   print("Naive silver rows (negative battery passed through):")
   naive_silver.select("ping_id", "device_id", "battery_pct").show(truncate=False)

   Show outputGuess what it prints first

   Naive silver rows (negative battery passed through):
   +-------+---------+-----------+
   |ping_id|device_id|battery_pct|
   +-------+---------+-----------+
   |p1     |dev_101  |85         |
   |p2     |dev_102  |90         |
   |p4     |dev_103  |-15        |
   |p5     |dev_101  |84         |
   |p6     |dev_102  |89         |
   +-------+---------+-----------+

   Common mistake

   Relying strictly on null checks while ignoring numerical domain validity constraints like range limits.

   How you will know

   Observe that naive filtering retains p4 with battery_pct -15.

3. 03

   Define comprehensive boolean validation rule expressions

   We formulate comprehensive boolean validation expressions that verify structural integrity and domain validity. We require device_id to be non-null, battery_pct to be bounded between 0 and 100 inclusive, and temp_c to fall within reasonable operating limits (-20 to 60 degrees Celsius). We evaluate this rule by adding an 'is_valid' boolean column to the DataFrame. Combining multiple column expressions using bitwise logical operators (& and |) allows Spark to execute vectorized validation checks across all worker partitions simultaneously without Python runtime overhead.

   PythonStep 3Copy

   from pyspark.sql import SparkSession
   from pyspark.sql import functions as F
   
   spark = SparkSession.builder.master("local[1]").appName("medallion_etl").getOrCreate()
   raw_data = [
       ("p1", "dev_101", 22.5, 85, "2024-03-01 10:00:00"),
       ("p2", "dev_102", 24.1, 90, "2024-03-01 10:00:05"),
       ("p3", None, 21.0, 75, "2024-03-01 10:00:10"),
       ("p4", "dev_103", 19.8, -15, "2024-03-01 10:00:15"),
       ("p5", "dev_101", 23.0, 84, "2024-03-01 10:05:00"),
       ("p6", "dev_102", 24.5, 89, "2024-03-01 10:05:05"),
   ]
   df_bronze = spark.createDataFrame(raw_data, ["ping_id", "device_id", "temp_c", "battery_pct", "timestamp"])
   
   valid_rule = (
       F.col("device_id").isNotNull() &
       (F.col("battery_pct") >= 0) &
       (F.col("battery_pct") <= 100) &
       (F.col("temp_c") >= -20) &
       (F.col("temp_c") <= 60)
   )
   flagged_df = df_bronze.withColumn("is_valid", valid_rule)
   flagged_df.select("ping_id", "device_id", "battery_pct", "is_valid").show(truncate=False)

   Show outputGuess what it prints first

   +-------+---------+-----------+--------+
   |ping_id|device_id|battery_pct|is_valid|
   +-------+---------+-----------+--------+
   |p1     |dev_101  |85         |true    |
   |p2     |dev_102  |90         |true    |
   |p3     |NULL     |75         |false   |
   |p4     |dev_103  |-15        |false   |
   |p5     |dev_101  |84         |true    |
   |p6     |dev_102  |89         |true    |
   +-------+---------+-----------+--------+

   Common mistake

   Using Python 'and' keyword instead of the PySpark bitwise operator '&' when chaining column expressions.

   How you will know

   Output shows an is_valid column where p3 and p4 are marked False while others are True.

4. 04

   Route records to Silver table and Quarantine dead-letter log

   We separate the flagged dataset into two distinct DataFrames: certified silver records where is_valid is True, and quarantine dead-letter records where is_valid is False. For quarantine records, we construct an explicit rejection_reason column using F.when expressions to document whether the failure was caused by a null device ID or an invalid battery reading. Dead-letter routing ensures that invalid records are preserved for firmware engineering triage while keeping the production silver table strictly clean, defect-free, and audit-ready across all tenant buildings.

   PythonStep 4Copy

   from pyspark.sql import SparkSession
   from pyspark.sql import functions as F
   
   spark = SparkSession.builder.master("local[1]").appName("medallion_etl").getOrCreate()
   raw_data = [
       ("p1", "dev_101", 22.5, 85, "2024-03-01 10:00:00"),
       ("p2", "dev_102", 24.1, 90, "2024-03-01 10:00:05"),
       ("p3", None, 21.0, 75, "2024-03-01 10:00:10"),
       ("p4", "dev_103", 19.8, -15, "2024-03-01 10:00:15"),
       ("p5", "dev_101", 23.0, 84, "2024-03-01 10:05:00"),
       ("p6", "dev_102", 24.5, 89, "2024-03-01 10:05:05"),
   ]
   df_bronze = spark.createDataFrame(raw_data, ["ping_id", "device_id", "temp_c", "battery_pct", "timestamp"])
   
   valid_rule = (
       F.col("device_id").isNotNull() &
       (F.col("battery_pct") >= 0) &
       (F.col("battery_pct") <= 100) &
       (F.col("temp_c") >= -20) &
       (F.col("temp_c") <= 60)
   )
   flagged_df = df_bronze.withColumn("is_valid", valid_rule)
   silver_df = flagged_df.filter(F.col("is_valid")).drop("is_valid")
   quarantine_df = (
       flagged_df
       .filter(~F.col("is_valid"))
       .withColumn(
           "rejection_reason",
           F.when(F.col("device_id").isNull(), "NULL_DEVICE_ID")
           .when((F.col("battery_pct") < 0) | (F.col("battery_pct") > 100), "INVALID_BATTERY_RANGE")
           .otherwise("CORRUPT_TEMP_READING")
       )
       .drop("is_valid")
   )
   print("Quarantine dead-letter log:")
   quarantine_df.select("ping_id", "device_id", "battery_pct", "rejection_reason").show(truncate=False)

   Show outputGuess what it prints first

   Quarantine dead-letter log:
   +-------+---------+-----------+---------------------+
   |ping_id|device_id|battery_pct|rejection_reason     |
   +-------+---------+-----------+---------------------+
   |p3     |NULL     |75         |NULL_DEVICE_ID       |
   |p4     |dev_103  |-15        |INVALID_BATTERY_RANGE|
   +-------+---------+-----------+---------------------+

   Common mistake

   Failing to attach explicit error reason strings to quarantine records, making automated alerts unactionable.

   How you will know

   Quarantine output shows p3 with NULL_DEVICE_ID and p4 with INVALID_BATTERY_RANGE.

5. 05

   Compute certified silver telematics metrics per device

   In the final step, we aggregate certified silver readings to generate gold-ready metrics. We group silver_df by device_id and calculate the total reading count, average temperature, and average battery percentage using F.round. Both dev_101 and dev_102 have 2 valid readings each: dev_101 averages 22.75 degrees Celsius with 84.5 percent battery, while dev_102 averages 24.3 degrees Celsius with 89.5 percent battery. The resulting metrics are mathematically certified against negative outliers or null keys, delivering clean operational telemetry for building management systems.

   PythonStep 5Copy

   from pyspark.sql import SparkSession
   from pyspark.sql import functions as F
   
   spark = SparkSession.builder.master("local[1]").appName("medallion_etl").getOrCreate()
   raw_data = [
       ("p1", "dev_101", 22.5, 85, "2024-03-01 10:00:00"),
       ("p2", "dev_102", 24.1, 90, "2024-03-01 10:00:05"),
       ("p3", None, 21.0, 75, "2024-03-01 10:00:10"),
       ("p4", "dev_103", 19.8, -15, "2024-03-01 10:00:15"),
       ("p5", "dev_101", 23.0, 84, "2024-03-01 10:05:00"),
       ("p6", "dev_102", 24.5, 89, "2024-03-01 10:05:05"),
   ]
   df_bronze = spark.createDataFrame(raw_data, ["ping_id", "device_id", "temp_c", "battery_pct", "timestamp"])
   
   valid_rule = (
       F.col("device_id").isNotNull() &
       (F.col("battery_pct") >= 0) &
       (F.col("battery_pct") <= 100) &
       (F.col("temp_c") >= -20) &
       (F.col("temp_c") <= 60)
   )
   silver_df = df_bronze.filter(valid_rule)
   
   device_summary = (
       silver_df
       .groupBy("device_id")
       .agg(
           F.count("*").alias("readings"),
           F.round(F.avg("temp_c"), 2).alias("avg_temp_c"),
           F.round(F.avg("battery_pct"), 1).alias("avg_battery_pct")
       )
       .orderBy("device_id")
   )
   device_summary.show(truncate=False)

   Show outputGuess what it prints first

   +---------+--------+----------+---------------+
   |device_id|readings|avg_temp_c|avg_battery_pct|
   +---------+--------+----------+---------------+
   |dev_101  |2       |22.75     |84.5           |
   |dev_102  |2       |24.3      |89.5           |
   +---------+--------+----------+---------------+

   Common mistake

   Applying rounding functions before aggregation rather than on the aggregated summary expression.

   How you will know

   Output shows 2 devices (dev_101 and dev_102) with 2 readings each and clean positive battery averages.

Build it yourself

Write your solution below, run it in your browser, and pass every check.

What the result means

Interpretation

Dual-routing Bronze-to-Silver ETL prevents data corruption while preserving operational observability. By routing invalid pings into dead-letter quarantine tables with explicit failure causes rather than silently dropping rows, data engineering teams provide hardware engineering teams with instant visibility into faulty firmware while delivering 100 percent certified metrics to downstream analytical consumers.

The message you would send

IoT Bronze-to-Silver medallion validation active: we implemented multi-condition data quality filters and automated dead-letter quarantine routing across 12,000 smart building installations. Negative battery anomalies and missing device identifiers dropped from production silver tables to zero, and firmware triage teams identified and patched 2 faulty gateway microcodes within 24 hours.

Put it on your CV

Resume or LinkedIn bullet

Designed PySpark lakehouse medallion ingestion pipelines with dual-routing dead-letter quarantine auditing for 12,000 IoT devices.

README for GitHub

# Lakehouse Medallion Bronze-to-Silver ETL in PySpark

## Problem
Raw IoT telemetry pings arrived with null device IDs and negative battery percentages (-15%), polluting executive energy dashboards.

## Solution Architecture
- Bronze: Append-only immutable landing zone.
- Validation: Vectorized PySpark expressions for device nullness, battery bounds (0..100), and temperature limits (-20..60C).
- Silver: Certified valid records.
- Quarantine: Dead-letter table capturing rejected pings with specific `rejection_reason` codes.

## Results
Reduced silver data defects to zero and provided hardware teams with real-time firmware diagnostic telemetry.

Keep going

• LessonLakehouse medallion pipeline
• Case studyMedallion Silver Quarantine: Isolating Bad Data at Ingestion
• Case studyWhich Events Are Broken: Identifying Ingestion Schema Failures