Cleaning, Joining and Reshaping Data with PySpark and SQL (Databricks Data Engineer Associate)

A bronze-to-silver step reads the raw table by its Unity Catalog name, cleans it in PySpark or SQL, and writes the result to a new silver Delta table with a write that is safe to rerun. Read a governed table with spark.read.table("catalog.schema.table") (or SELECT ... FROM catalog.schema.table): it returns a batch DataFrame of the table's latest version, and you never need the managed storage path. Path-based readers such as spark.read.load(path) or spark.read.parquet(path) expect a file location, not a table name, and reading a Delta table's Parquet files directly bypasses its transaction log, so the read can include files that were removed by later commits or never committed, and does not reliably match the table's current version. spark.readStream.table(...) turns the read into an incremental streaming query, which suits incremental pipelines rather than a full nightly rebuild.

raw = spark.read.table("main.bronze.shipments")
clean = (raw
    .dropna(subset=["shipment_id"])
    .withColumn("freight_cost", col("freight_cost").cast("decimal(18,2)")))
clean.write.mode("overwrite").saveAsTable("main.silver.shipments")

The save mode decides what a rerun does, which matters because jobs are retried:

In SQL, CREATE OR REPLACE TABLE main.silver.products AS SELECT ... (CTAS) materializes the cleaned query into a Delta table and replaces its contents on each run. CREATE TABLE IF NOT EXISTS ... AS SELECT only creates it once, INSERT INTO appends, and CREATE OR REPLACE VIEW stores no data at all (views belong to the gold-layer lesson). Both mode("overwrite") and CREATE OR REPLACE TABLE write a new version of the same table, so its history stays available for DESCRIBE HISTORY and time travel; dropping and recreating the table loses that history.

When the cleaned schema changes

By default an overwrite replaces the data but keeps the table's schema, and Delta schema enforcement checks the new DataFrame against it. A column whose type changed (for example STRING to DECIMAL) or a column the table does not have makes the write fail. A column the cleaned DataFrame no longer outputs does not fail the write: it stays in the table and is filled with nulls. To change a column's type or actually remove a column, add .option("overwriteSchema", "true"), which replaces schema and data in one commit and keeps the history. mergeSchema is a different tool: it allows compatible evolution such as adding new columns during a write, not replacing a column's type.

Clean nulls column by column: dropna removes rows that are missing values you cannot do without, and fillna replaces nulls with a default where a default is correct. Both act on every column unless you scope them, so the subset argument (or a dict) is what keeps a rule from touching columns it should not.

Two behaviours catch people out. First, fillna only touches columns whose type matches the value: a numeric value fills numeric columns and a string value fills string columns, and a non-matching column is silently skipped. A digits-only quantity stored as a string keeps its nulls after fillna(0); cast it to INT first, then fill. Second, order matters: filling every string column with a placeholder before dropna(subset=["device_id"]) hides the null identifiers, so the rows you meant to remove survive. Drop first, then fill only what should be filled.

df.na.replace(...) replaces existing values (for example the string '--'), not nulls. In SQL the same cleaning uses WHERE device_id IS NOT NULL and coalesce(channel, 'unknown') (or nvl).

Standardizing types means giving each silver column the type its meaning needs: dates as DATE, timestamps as TIMESTAMP, counts as INT or BIGINT, and money as an exact DECIMAL(precision, scale). Bronze data loaded from CSV or JSON often holds all of these as strings, so the silver step converts them explicitly with cast (PySpark col("x").cast("int"), SQL CAST(x AS INT)) and the parsing functions.

• Dates and timestamps in a non-ISO format need a pattern: to_date(col("ship_date"), "dd.MM.yyyy") for '15.03.2026', to_timestamp(col, pattern) for timestamps. In patterns, MM is the month and mm is minutes; dd is the day. A plain cast to DATE only understands ISO-style strings such as '2026-03-15'. date_format goes the other way: it turns a date into a formatted string.
• Money belongs in DECIMAL. DOUBLE and FLOAT are approximate binary types, so cent values drift when summed. Choose a precision large enough for the biggest value: DECIMAL(5,2) tops out at 999.99, while DECIMAL(18,2) holds large totals exactly.
• Text is standardized with trim, lower/upper and regexp_replace before comparisons or joins depend on it.

Malformed values: CAST versus try_cast

What happens to a value that cannot be converted, such as 'unknown' cast to DECIMAL, depends on ANSI mode. With ANSI mode on (the default on Databricks SQL warehouses; on clusters it follows the spark.sql.ansi.enabled setting, whose default depends on the Databricks Runtime version), CAST raises an error and the whole statement fails. With it off, CAST returns NULL. try_cast(unit_price AS DECIMAL(10,2)) returns NULL for malformed values in either mode, so the statement succeeds, valid values convert, and the bad rows stay in the table with a null you can review. Wrapping a failing CAST in coalesce does not help, because the error is raised before coalesce sees a value.

A join pairs each row on one side with every matching row on the other, and the join type decides what happens to rows that have no match. An inner join keeps only matched pairs; a left (left outer) join keeps every row of the left side and fills the right side's columns with nulls where nothing matches.

The table you must not lose rows from is the preserved side: invoices.join(suppliers, "supplier_id", "left") keeps every invoice even when its supplier record has not arrived yet.

Row multiplication

Joins never deduplicate. If the right side has several rows for a key (a price history, or one row per sales channel), every left row with that key appears once per match, so a one-row-per-invoice table grows after the join. Reduce the right side to one row per key before joining, for example prices.filter("is_latest"). Deduplicating on the left key after the join is not equivalent: dropDuplicates keeps an arbitrary one of the matched rows, so the surviving right-side value is not necessarily the current one.

Filters on the right side: ON versus WHERE

Where you put a condition on the right side's columns changes an outer join's result. In the ON clause it limits which right rows can match, and the left join still keeps every left row. In WHERE it runs after the join: unmatched rows have null right columns, the comparison evaluates to null, and the row is dropped, so the left join silently behaves like an inner join. The same applies in PySpark: invoices.join(prices.filter("is_latest"), "product_id", "left") keeps every invoice, while filtering on is_latest after the left join removes invoices with no price row.

SELECT s.shipment_id, c.carrier_name
FROM silver.shipments s
LEFT JOIN silver.carriers c
  ON s.carrier_id = c.carrier_id AND c.active = true   -- every shipment kept

A join on a composite key needs every key part to match, combined with AND. How you write it depends on whether the key columns have the same names on both sides.

• Same names: pass a list, inventory.join(products, ["product_id", "site"], "inner") (SQL: USING (product_id, site)). The keys are combined with AND and the result keeps one copy of each key column, so selecting product_id afterwards is not ambiguous.
• Different names: write an expression, one equality per key pair, joined with &: (shipments.wh_code == warehouses.warehouse_code) & (shipments.country == warehouses.country). Both sides' columns stay in the result, so drop or alias the duplicates you do not need, or later references to a shared name such as country are ambiguous.

In PySpark, combine Column conditions with &, | and ~, and wrap each comparison in parentheses (& binds tighter than ==). Python's and/or keywords do not work on Columns: they raise a "Cannot convert column into bool" error, and "product_id" and "site" on plain strings simply evaluates to one string. Using | instead of & matches rows that share either key, and leaving out a key part lets IDs that repeat across sites match the wrong row.

Null key parts

In SQL logic, NULL = NULL is null, not true, so rows whose key contains a null never match in an equi-join. When null is a meaningful key value (for example a top-level region with no parent region, present in both tables), use null-safe equality on the column that can be null: r.parent_id.eqNullSafe(t.parent_id) in PySpark, <=> in SQL. It returns true for two nulls and false for a null versus a value. Adding OR right_key IS NULL instead is wrong: it matches the null row to every row on the other side.

A broadcast join copies a small table to every executor so that the large table can be joined where it sits, without shuffling it across the cluster. Without it, a join of two large inputs is usually a sort-merge join, which shuffles both sides by the key; for a multi-terabyte fact table joined to a small dimension, that shuffle is most of the cost.

• PySpark: wrap the small side, clicks.join(broadcast(campaigns), "campaign_id"), with broadcast imported from pyspark.sql.functions.
• SQL: add a hint naming the small relation or its alias: SELECT /*+ BROADCAST(d) */ ... FROM events e JOIN devices d ON e.device_id = d.device_id.

A hint is part of the code or query text, so it needs no session or cluster setting change. Spark also broadcasts automatically when a side is smaller than spark.sql.autoBroadcastJoinThreshold; tuning that setting is covered in the lesson on deduplication, aggregation and basic Spark tuning. Two rules decide whether a broadcast is right:

Other join hints exist, such as MERGE (sort-merge) and SHUFFLE_HASH, but both shuffle the data by the key, so they do not remove the shuffle of the large table. Repartitioning the large side by the key also shuffles it.

Cross joins

A cross join (df1.crossJoin(df2), SQL CROSS JOIN) returns the cartesian product: every row of one side paired with every row of the other, so the output has rows(left) x rows(right) rows. It is the right tool in two common cases:

• A scaffold of all combinations, such as every product for every week. Left joining the facts to that scaffold on the full key (["product_id", "week_start"]), with the scaffold as the preserved side, then fillna(0, subset=["units_sold"]), produces rows for weeks with no sales.
• Attaching a one-row result (an overall average or a run timestamp) to every row. Crossing with a single-row DataFrame multiplies the row count by one. You cannot do this with withColumn("x", other_df["x"]), because a Column can only be used inside its own DataFrame.

A cross join followed by an equality filter is not a performance trick: the optimizer plans it as the same equi-join. Unions are not a substitute either; they stack rows rather than pairing them.

A union stacks the rows of two inputs with the same columns. In SQL, UNION removes duplicate rows (it means UNION DISTINCT) and UNION ALL keeps every row, which is also cheaper because no deduplication step runs. When identical rows are legitimate separate records, or the inputs cannot overlap, use UNION ALL. INTERSECT and EXCEPT compare sets; they do not append.

The PySpark methods differ from SQL in two ways:

Position-based matching is the trap: if two DataFrames hold the same columns in a different order, union puts one side's values under the other side's column names, and when the types are compatible nothing fails. Appending a missing column with withColumn places it last, which does not line it up with a column that sits in the middle of the other DataFrame. unionByName resolves columns by name, and allowMissingColumns=True handles a feed that gained or lost a column. Adding distinct() never fixes alignment; it only removes duplicates.

Column changes in a DataFrame use four methods: withColumn(name, expr) adds a column or replaces one with the same name, drop(name) removes it, withColumnRenamed(existing, new) renames it, and select (with alias) chooses and renames columns in one step. In SQL the same thing happens in the SELECT list (prod_desc AS product_description).

silver = (df
    .withColumnRenamed("prod_desc", "product_description")  # existing name first
    .drop("src_file")
    .withColumn("load_date", current_date()))  # DATE; current_timestamp() is a TIMESTAMP

Missing names are silent no-ops. withColumnRenamed returns the DataFrame unchanged when the existing name is not in the schema (including when the two arguments are reversed), and drop ignores a name that does not exist. After an upstream rename, the cleaning job still succeeds and the problem surfaces only downstream. Selecting the expected columns explicitly, or checking df.columns, makes such drift fail early. Copying a column with withColumn("new", col("old")) is not a rename: the old column stays.

Renaming or dropping a column in a Delta table

A DataFrame rename changes only that DataFrame, not the stored table. To change the table itself use ALTER TABLE ... RENAME COLUMN or ALTER TABLE ... DROP COLUMN (add columns with ALTER TABLE ... ADD COLUMNS). Delta Lake allows RENAME and DROP COLUMN only when column mapping is enabled. On an existing table, enable it with ALTER TABLE t SET TBLPROPERTIES ('delta.columnMapping.mode' = 'name'); after that, rename and drop change metadata only, with no rewrite of the data files. The other mode, 'id', must be set when the table is created. Enabling column mapping upgrades the table's Delta protocol, so very old readers can no longer read it. Rebuilding the table with CTAS also renames a column, but it rewrites all of the data.

Splitting a column

split(col, pattern, limit) turns a string into an array. Read elements with getItem(i) (or arr[i]), which is zero-based, or with element_at(arr, n), which is one-based (index 0 is an error; negative indexes count from the end). Reading past the end returns null with ANSI mode off and raises an error with it on; try_element_at returns null in either mode.

• The pattern is a regular expression. Characters such as |, . and $ have special meaning: an unescaped | is an empty alternation that matches between every character, so the result is single characters. Use a character class ("[|]") or escape it with a backslash. A forward slash is not an escape.
• The limit caps the pieces. With a positive limit, the pattern is applied at most limit - 1 times and the last element holds the rest of the string, so split(col("file_name"), "_", 3) cuts at the first two underscores and leaves any further underscores inside the third element. A limit of zero or less (the default is -1) splits as many times as possible; a limit of 1 does not split.
• In SQL, split_part(str, delimiter, n) takes a literal delimiter and a one-based part number.

To turn array elements into separate columns, use getItem; to turn them into separate rows, use explode.

Filter rows with df.filter(condition) or its alias df.where(condition), using a Column expression or a SQL string; a row is kept only when the condition is true. Because comparisons with null evaluate to null (SQL three-valued logic), a row whose column is null is dropped by both tier != 'trial' and tier = 'trial', and by their negations.

Other filtering rules worth knowing:

• between(lower, upper) is inclusive at both ends, so the first and last day of a month bound a whole calendar month of DATE values. Strict > and < drop the boundary days.
• Combine conditions with & and |, each comparison in parentheses; Python's and raises an error on Columns.
• NOT IN with a null in the list returns no rows. x NOT IN (v1, v2, NULL) expands to x <> v1 AND x <> v2 AND x <> NULL; the last comparison is unknown, so the predicate is never true. Deduplicating the list does not remove the null. NOT EXISTS (SELECT 1 ... WHERE b.key = o.key) or a left_anti join asks only whether a match exists and gives the expected result.

explode turns an array column into rows: one output row per element, with the other selected columns repeated. Four generator functions cover the combinations of "do I need the element's position?" and "must rows with a null or empty array survive?":

The non-outer functions emit nothing for an empty array as well as for a null one, so replacing nulls with an empty array does not keep those rows; only the _outer variants do. On a map column, explode returns key and value columns.

Arrays of structs

An array of structs (for example lines with fields product_code, units and unit_price) is flattened in two steps: explode the array into one struct per row, then expand the struct's fields with .*, keeping the parent key so each line stays linked to its invoice.

(invoices
  .select("invoice_id", explode("lines").alias("ln"))
  .select("invoice_id", "ln.*"))  # invoice_id, product_code, units, unit_price

Star expansion works on a struct, not on an array, so select("lines.*") fails. In SQL, inline(lines) explodes an array of structs straight into columns.

Several arrays at once

Spark allows only one generator per select, and exploding two arrays one after the other produces every combination (a cartesian product per row). When elements at the same position belong together, combine the arrays first: explode(arrays_zip("labels", "scores")) yields one struct per position, whose fields are named after the input columns. concat joins arrays end to end, which loses the pairing.