Skip to content
FR

Cleaning and Combining Data with PySpark and SQL

Step 11 of 30

Cleaning and Combining Data with PySpark and SQL

Introduction

This lesson opens the Data Transformation and Modeling section of the official exam outline, the heaviest section of the exam at about 22% of the scored questions (roughly ten of 45). Lessons 3 to 5 got raw data into Delta tables. Now you turn that raw data into trustworthy, joined, analysis-ready tables. The exam guide expects you to read bronze tables, clean and standardize them into silver, combine datasets with the right join or union, and reshape columns, using SQL or PySpark. Databricks states that data manipulation code on the exam is shown in SQL when possible and otherwise in PySpark, so you must read both fluently.

After this lesson you will be able to explain the medallion architecture, write cleaning logic for nulls and types, choose between join types and set operations, and reshape tables by adding, dropping, renaming and splitting columns and exploding arrays.

Learning objectives

  • Describe the bronze, silver and gold layers and what belongs in each.
  • Clean a bronze table into silver: handle nulls, cast types, trim and standardize values, and filter invalid rows.
  • Choose between inner, left, broadcast, multi-key and cross joins, and predict the row count of each.
  • Distinguish UNION, UNION ALL and unionByName, and explain when each is safe.
  • Add, drop, rename and split columns, filter rows, and use explode and explode_outer on arrays.
  • Recognize the most common mistakes: join duplication, null handling in comparisons, and positional unions with reordered columns.

1. The medallion architecture

The medallion architecture organizes a lakehouse into three layers of increasing quality. It is a design pattern, not a product feature, but the exam uses its vocabulary constantly.

Layer Contents Typical operations Consumers
Bronze Raw data as ingested, often with added metadata (source file, load time) Append only, minimal changes, keep history Data engineers, replay and audit
Silver Cleaned, deduplicated, typed, conformed and joined data Cast, filter, join, standardize, validate Data engineers, data scientists, analysts
Gold Business-level aggregates and models shaped for a use case Aggregate, pivot, build dimensions and facts BI, dashboards, ML features

The reasoning is practical. Keeping bronze unchanged means you can always rebuild silver if a cleaning rule turns out to be wrong. Keeping silver as the single cleaned source means every gold table starts from the same definition of a "valid order". Each layer is usually a set of Delta tables in Unity Catalog, for example main.bronze.orders_raw, main.silver.orders and main.gold.daily_revenue.

Concrete example. A retailer, NorthPeak Outdoors, ingests 2 million order lines per day from its web shop with Auto Loader into bronze.orders_raw. The bronze table has every column as a string, a _rescued_data column and the source file path. Silver orders casts order_ts to a timestamp and amount to DECIMAL(12,2), removes test orders and rows without a customer, and standardizes country codes. Gold daily_revenue_by_country sums revenue per day and country for the finance dashboard.

2. Cleaning bronze data into silver

Cleaning is a sequence of small, explicit decisions. The same logic can be written in SQL or PySpark.

Reading the bronze table

SELECT * FROM main.bronze.orders_raw;

bronze_df = spark.read.table("main.bronze.orders_raw")

For incremental pipelines you would read the bronze table as a stream (spark.readStream.table(...)), but for the exam's cleaning questions a batch read is enough.

Standardizing data types

Raw files often arrive as strings. Cast them early so that comparisons, sorting and aggregations behave correctly. Sorting a string column "100", "20", "3" gives "100", "20", "3", not numeric order.

CREATE OR REPLACE TABLE main.silver.orders AS
SELECT
  CAST(order_id AS BIGINT)               AS order_id,
  CAST(customer_id AS BIGINT)            AS customer_id,
  to_timestamp(order_ts, 'yyyy-MM-dd HH:mm:ss') AS order_ts,
  CAST(amount AS DECIMAL(12,2))          AS amount,
  upper(trim(country))                   AS country
FROM main.bronze.orders_raw
WHERE customer_id IS NOT NULL
  AND amount IS NOT NULL;

from pyspark.sql import functions as F

silver_df = (
    spark.read.table("main.bronze.orders_raw")
    .select(
        F.col("order_id").cast("bigint").alias("order_id"),
        F.col("customer_id").cast("bigint").alias("customer_id"),
        F.to_timestamp("order_ts", "yyyy-MM-dd HH:mm:ss").alias("order_ts"),
        F.col("amount").cast("decimal(12,2)").alias("amount"),
        F.upper(F.trim("country")).alias("country"),
    )
    .filter(F.col("customer_id").isNotNull() & F.col("amount").isNotNull())
)
silver_df.write.mode("overwrite").saveAsTable("main.silver.orders")

Two behaviors matter. First, a failed cast does not always raise an error: with ANSI mode off, CAST('abc' AS INT) returns NULL, and try_cast is the explicit way to say "return null instead of failing". With ANSI mode on (the default in recent Databricks Runtime versions and on serverless compute), a plain CAST of an invalid string raises an error, so try_cast becomes the safe choice for dirty data. Second, a value that is all whitespace is not null, so trim first and test for the empty string explicitly if needed.

Handling nulls

A NULL means "unknown", and almost every comparison with it is unknown too. amount = NULL is never true, even for rows where amount is null. Use IS NULL and IS NOT NULL.

Goal SQL PySpark
Keep rows with a value WHERE col IS NOT NULL df.filter(F.col("col").isNotNull()) or df.dropna(subset=["col"])
Replace nulls with a default coalesce(col, 0) df.fillna({"col": 0}) or F.coalesce("col", F.lit(0))
Drop rows with any null n/a (write the conditions) df.dropna() (any column)
Null-safe equality a <=> b a.eqNullSafe(b)

dropna() with no arguments drops a row if any column is null, which can silently delete most of a wide table. Always pass subset and decide per column whether a null is acceptable, replaceable or a reason to reject the row.

Example. In NorthPeak's feed, discount is null when no discount applied. Dropping those rows would lose most orders, so silver uses coalesce(discount, 0). customer_id, in contrast, is required, so rows without it are filtered out (and ideally written to a quarantine table for investigation).

Filtering invalid rows

Use WHERE (SQL) or filter/where (PySpark) to remove records that break business rules: negative amounts, orders dated in the future, test accounts. Filter after casting so that the conditions run on correct types.

3. Combining data with joins

A join combines rows from two tables based on a condition. Choosing the wrong type is a classic source of wrong answers.

Join type Rows returned Unmatched left rows Unmatched right rows
Inner (default) Only matching rows Dropped Dropped
Left (left outer) All left rows, matched right columns Kept, right columns NULL Dropped
Right (right outer) All right rows Dropped Kept, left columns NULL
Full outer All rows from both Kept Kept
Left semi Left rows that have a match (left columns only) Dropped n/a
Left anti Left rows that have no match Kept n/a
Cross Every left row paired with every right row n/a n/a

Inner and left joins

SELECT o.order_id, o.amount, c.segment
FROM main.silver.orders o
INNER JOIN main.silver.customers c
  ON o.customer_id = c.customer_id;

joined = orders.join(customers, on="customer_id", how="inner")

When the join column has the same name on both sides, on="customer_id" keeps a single copy of the column. Writing orders.customer_id == customers.customer_id keeps both columns, and referring to customer_id afterwards raises an ambiguity error unless you drop one.

Which one to use? If the business question is "revenue per customer segment" and some orders have an unknown customer, an inner join silently removes that revenue. A left join keeps the orders and gives them a null segment, which you can label 'Unknown' with coalesce. The totals then reconcile with the source.

Multi-key joins

When one column does not uniquely identify a row, join on several. Daily exchange rates are unique per (currency, rate_date), so joining orders to rates on currency alone would multiply each order by the number of days.

SELECT o.order_id, o.amount * r.rate AS amount_eur
FROM orders o
JOIN fx_rates r
  ON o.currency = r.currency
 AND to_date(o.order_ts) = r.rate_date;

cond = [orders.currency == fx.currency,
        F.to_date(orders.order_ts) == fx.rate_date]
converted = orders.join(fx, cond, "inner")

Broadcast joins

When one side is small (a dimension of a few thousand rows), Spark can send a full copy to every executor and join without shuffling the large table. This is a broadcast join.

from pyspark.sql.functions import broadcast
result = orders.join(broadcast(countries), "country")

SELECT /*+ BROADCAST(c) */ o.order_id, c.country_name
FROM orders o JOIN countries c ON o.country = c.country_code;

Spark broadcasts automatically when a table is below spark.sql.autoBroadcastJoinThreshold (10 MB by default in open-source Spark; Databricks adaptive execution also converts joins at runtime using its own threshold). The hint is useful when you know a table is small but statistics are missing. The next lesson covers these settings in more detail. A broadcast join does not change the result, only the execution strategy: it never changes which rows match.

Cross joins

A cross join produces the Cartesian product: 1,000 rows times 500 rows gives 500,000 rows. It is legitimate for generating combinations (every store with every day of a calendar table) and dangerous otherwise. Spark requires the explicit CROSS JOIN keyword or crossJoin(), which protects against accidental products caused by a missing condition.

SELECT s.store_id, d.cal_date
FROM stores s CROSS JOIN calendar d;

Semi and anti joins

To answer "which customers have never ordered?", an anti join is clearer and safer than NOT IN (which behaves badly with nulls):

SELECT c.customer_id
FROM customers c
LEFT ANTI JOIN orders o ON c.customer_id = o.customer_id;

4. Combining data with set operations

Joins add columns side by side. Set operations stack rows from tables with the same shape.

Operation Behavior Duplicates Column matching
UNION ALL (PySpark union) Append all rows Kept By position
UNION / UNION DISTINCT Append and remove duplicate rows Removed (costs a shuffle) By position
unionByName Append, matching columns by name Kept By name; allowMissingColumns=True fills missing columns with null

In PySpark, DataFrame.union behaves like SQL UNION ALL: it does not deduplicate. This surprises many people. Use .distinct() afterward for UNION semantics.

eu_orders = spark.read.table("main.silver.orders_eu")
us_orders = spark.read.table("main.silver.orders_us")

# Same columns, different order in the US table -> positional union would mix them up
all_orders = eu_orders.unionByName(us_orders, allowMissingColumns=True)

Why unionByName matters. Suppose orders_eu has (order_id, amount, currency) and orders_us was built with (order_id, currency, amount). A positional union puts currency codes into the amount column. Spark may raise a type error, or, if the types are compatible, produce silently wrong data. unionByName is the safe default when sources are built by different teams. SQL offers the same with UNION ALL BY NAME in recent Databricks Runtime versions.

5. Reshaping columns and rows

Adding, renaming and dropping columns

df2 = (
    silver_df
    .withColumn("order_date", F.to_date("order_ts"))
    .withColumn("net_amount", F.col("amount") - F.coalesce(F.col("discount"), F.lit(0)))
    .withColumnRenamed("amount", "gross_amount")
    .drop("_rescued_data")
)

SELECT
  *,
  to_date(order_ts) AS order_date,
  amount - coalesce(discount, 0) AS net_amount
FROM main.silver.orders;

ALTER TABLE main.silver.orders RENAME COLUMN amount TO gross_amount;
ALTER TABLE main.silver.orders DROP COLUMN _rescued_data;

Note the difference between transforming a query result (withColumn, SELECT) and changing a table (ALTER TABLE). RENAME COLUMN and DROP COLUMN on a Delta table require column mapping to be enabled on the table; Databricks enables it when you set the table property, and the operations are metadata-only, so they are fast.

Splitting columns

split turns a delimited string into an array, and getItem (or bracket syntax) picks an element.

people = df.withColumn("parts", F.split("full_name", " "))
people = (people
          .withColumn("first_name", F.col("parts").getItem(0))
          .withColumn("last_name", F.col("parts").getItem(1))
          .drop("parts"))

SELECT split(full_name, ' ')[0] AS first_name,
       split(full_name, ' ')[1] AS last_name
FROM customers;

Exploding arrays

explode produces one row per element of an array; the other columns repeat. Rows whose array is null or empty disappear with explode; explode_outer keeps them with a null element.

SELECT order_id, explode(items) AS item FROM orders_nested;
SELECT order_id, explode_outer(items) AS item FROM orders_nested;

After exploding, aggregates such as SUM(order_total) are inflated because the order total repeats once per item. Always aggregate at the right grain: sum item amounts, or deduplicate to one row per order before summing the order total.

Worked example: building a silver table without losing revenue

Scenario. NorthPeak needs silver.order_lines with the customer segment. Source tables:

  • silver.orders: 1,000,000 rows, one per order, 8,000 rows have customer_id not present in silver.customers.
  • silver.customers: 250,000 rows, unique on customer_id.
  • bronze.order_items_raw: nested items per order in an array column items.

The finance team requires that the silver total revenue equals the source total.

Step 1: choose the join type. Finance requires reconciliation, so no order can disappear. An inner join would drop the 8,000 orders. Use a left join from orders to customers.

Step 2: check the join key's uniqueness. customers is unique on customer_id, so each order matches at most one customer and the row count stays 1,000,000. If customers had duplicates, the left join would multiply orders. Verify with SELECT customer_id, COUNT(*) FROM silver.customers GROUP BY 1 HAVING COUNT(*) > 1.

Step 3: label the unmatched rows.

SELECT o.order_id, o.items,
       coalesce(c.segment, 'Unknown') AS segment
FROM silver.orders o
LEFT JOIN silver.customers c ON o.customer_id = c.customer_id;

Step 4: explode the items. Use explode_outer, since some orders (for example fully refunded ones) may have an empty array, and dropping them would again remove them from counts.

Step 5: aggregate at the right grain. Revenue must be summed from the item amount, not from the repeated order total.

Why the alternatives are worse. An inner join loses 8,000 orders and breaks the reconciliation. A full outer join also brings in customers without orders (null order columns) which are irrelevant for an order table. A plain explode drops empty-array orders. A cross join is never appropriate here, since it would create 250 billion rows (1,000,000 times 250,000).

Common pitfalls and exam tips

  • Equality with null. col = NULL is never true. Use IS NULL, or the null-safe operator <=>.
  • dropna() without subset. It removes a row with a null in any column. Expect an exam scenario where a table "loses most of its rows".
  • Join key duplicates. A one-to-many join multiplies rows. Check the grain of both tables before choosing a join and before summing.
  • Same-named columns. Joining with a condition expression keeps both columns; joining on="name" keeps one.
  • union is not UNION. In PySpark, union keeps duplicates. Use distinct() for set semantics and unionByName when column order differs.
  • explode drops empty arrays. Use explode_outer to keep them.
  • Cast failures. A bad cast returns null (or fails under ANSI mode). try_cast is the explicit non-failing form.
  • Layers. Anything that is a business aggregate belongs in gold; cleaned row-level data belongs in silver; unchanged raw history belongs in bronze.

Exam tip: When the question says "the totals must match the source" or "no records may be lost", think left join and explode_outer. When it says "columns arrive in different orders", think unionByName. When it says "one side is small", think broadcast, which affects performance and never the result.

Mini case study

Situation. A logistics company loads two bronze tables of shipment events: bronze.events_eu (columns shipment_id, event_type, event_ts) and bronze.events_asia (columns event_ts, shipment_id, event_type, with event_ts as a string). It also has silver.carriers (unique on carrier_id, 60 rows). Analysts want one silver table silver.shipment_events with the carrier name. Some events have a null carrier_id, and the operations team insists those events stay visible.

Task. Describe how to build the table and justify each choice.

Solution.

  1. Cast event_ts in the Asia table to a timestamp before combining so both inputs have the same types.
  2. Combine with unionByName, not union: the column orders differ, and a positional union would put timestamps in the shipment column or fail. Use UNION ALL semantics because events are legitimate repeated facts, and deduplicate separately (next lesson) with a key if duplicates are a concern.
  3. Join to silver.carriers with a left join on carrier_id so events with a null or unknown carrier remain. The carriers table is unique and tiny (60 rows), so it can be broadcast with broadcast(carriers); this speeds the join without changing the output.
  4. Replace the missing carrier name with coalesce(name, 'Unknown').
  5. Drop helper and raw columns (such as _rescued_data) and write the result with saveAsTable.

A join type of inner would hide the events with unknown carriers, the opposite of the requirement.

Key terms

  • Medallion architecture: bronze (raw), silver (cleaned and conformed), gold (business-level) layering.
  • Silver table: cleaned, typed, deduplicated, validated data.
  • Inner / left / full outer join: joins that keep only matches, all left rows, or all rows of both sides.
  • Semi / anti join: filter the left table by the presence or absence of a match, returning no right columns.
  • Broadcast join: join strategy that copies a small table to every executor to avoid a shuffle.
  • Cross join: Cartesian product of two tables.
  • UNION ALL / union: append rows by position, keeping duplicates.
  • unionByName: append rows matching columns by name.
  • explode / explode_outer: one row per array element, dropping or keeping empty arrays.
  • coalesce: first non-null value among its arguments.

Summary checklist

  • I can explain what belongs in bronze, silver and gold.
  • I can cast types, handle nulls with IS NULL, coalesce, fillna and dropna(subset=...).
  • I can choose between inner, left, semi, anti, full and cross joins and predict the row count.
  • I know a broadcast join changes performance, not the result.
  • I can tell union, UNION, UNION ALL and unionByName apart.
  • I can add, drop, rename and split columns in SQL and PySpark.
  • I can use explode and explode_outer and aggregate at the right grain.

Official sources