PySpark Data Cleaning on Databricks: Nulls, withColumn and Safe Delta Writes

Cleaning data in PySpark comes down to a few deliberate choices: pick the columns you need, decide what makes a row valid, handle missing labels on purpose, derive new columns, and save the result without overwriting anything. This post follows the video's lesson on Databricks Free Edition, using eight fictional orders that end up as five clean rows in a new Delta table.

<iframe width="560" height="315" src="https://www.youtube.com/embed/J7iuMxWW6s0" title="PySpark Data Cleaning Tutorial: Nulls, withColumn &amp; Delta Writes on Databricks" frameborder="0" allowfullscreen></iframe>

Watch on YouTube

The notebook, source file and cheat sheet are in the lesson folder on GitHub. Use a Python notebook on Serverless compute and run the cells from the top.

Start with an explicit schema

The sample data is built in memory, so there is nothing to download. An explicit schema names each column, sets its type and allows missing values. Unit price is DECIMAL(10, 2), which avoids binary floating-point numbers for prices.

schema = StructType([
    StructField("order_id", IntegerType(), True),
    StructField("customer", StringType(), True),
    StructField("region", StringType(), True),
    StructField("quantity", IntegerType(), True),
    StructField("unit_price", DecimalType(10, 2), True),
    StructField("internal_note", StringType(), True),
])

The eight raw rows have different problems on purpose: a name with spaces around it, a missing customer, a blank region, a missing quantity, a missing price and a negative quantity. These problems need different fixes. Filling everything with zero would be wrong.

select returns a new DataFrame

select chooses the five columns you need and drops internal_note:

selected = raw_orders.select(
    "order_id", "customer", "region", "quantity", "unit_price"
)
assert "internal_note" not in selected.columns
assert "internal_note" in raw_orders.columns

A transformation returns a new DataFrame. It does not change the original. raw_orders still has all its columns, and selected still has all eight rows. Choosing columns is not the same as filtering rows.

Decide what a valid order means, then filter

Write the business rule before you write the filter. Here a valid order needs an identifier, a quantity and a price. Quantity must be greater than zero, and price must not be negative. A zero price is allowed. A missing price is not treated as a free sale.

valid = (
    F.col("order_id").isNotNull()
    & F.col("quantity").isNotNull()
    & F.col("unit_price").isNotNull()
    & (F.col("quantity") > 0)
    & (F.col("unit_price") >= 0)
)
accepted = selected.filter(valid)
rejected = selected.filter(~valid)

Combine Spark conditions with & and wrap each comparison in parentheses. Use isNull() or isNotNull() to test for missing values. Comparing a column to Python None with == does not work, and comparing a missing value to a number does not return true.

The result is five accepted rows and three rejected rows: 1002 (missing quantity), 1004 (quantity of -1) and 1005 (missing price). Keeping the rejected rows in their own DataFrame lets you inspect them instead of losing the evidence.

Handle missing labels with a different policy

Customer and region are descriptive labels, so a different rule fits. Trim the spaces, and if the result is empty or null, use Unknown:

def label_or_unknown(column_name):
    value = F.trim(F.col(column_name))
    return F.when(value.isNull() | (value == ""), F.lit("Unknown")).otherwise(value)

labeled = ( accepted .withColumn("customer", labelorunknown("customer")) .withColumn("region", F.upper(labelorunknown("region"))) ) ```

The helper returns a Spark column expression built from built-in functions (trim, when, lit, upper). It is not a Python user-defined function that runs row by row. withColumn adds or replaces a column and returns a new DataFrame.

After this step, Ana's name has no extra spaces, the missing customer is Unknown, and the blank region is UNKNOWN after uppercasing.

Derive the order total

clean_orders = labeled.withColumn(
    "order_total", (F.col("quantity") * F.col("unit_price")).cast("decimal(12,2)")
)

The five totals are 20.00, 22.50, 12.00, 8.50 and 3.00, which add up to 66.00. The notebook's checks confirm 8 raw rows, 5 clean rows, 3 rejected rows, and that raworders still has no ordertotal column.

orderBy makes the displayed rows easy to read. It does not guarantee the physical order of a saved table.

Write to a new Delta table that refuses to overwrite

The notebook creates a new schema named btcleanorders plus a random suffix in your current catalog, and a managed table called cleanorders inside it. The write uses error mode:

clean_orders.write.format("delta").mode("error").saveAsTable(table_fqn)

mode("error") fails if the table already exists. It does not overwrite and does not append. The lesson uses the exact spelling error because the longer alias errorifexists was rejected by the Free Edition Spark Connect client during testing. If the write already succeeded, skip that cell and run the verification cell.

Verify what was saved

An in-memory result is not proof. Read the stored table back with SQL:

spark.sql(f"""
SELECT COUNT(*) AS row_count, SUM(order_total) AS total_sales
FROM {table_fqn}
""").show()

The actual Free Edition run returned 5 rows and total sales of 66.00, with order IDs 1001, 1003, 1006, 1007 and 1008.

Practice challenge

Without writing another table: from cleanorders, keep NORTH rows with ordertotal >= 15, select orderid, customer and ordertotal, and order by order_id. Predict the answer first. You should get one row: 1001, Ana, 20.00.

Key takeaways

• select chooses columns, filter chooses rows, and both return new DataFrames.
• Treat null handling as a business decision. Reject rows with missing numbers instead of filling them with zero.
• Prefer built-in functions such as trim, when and upper over Python UDFs.
• Write with mode("error") so a re-run never replaces your table, then verify the stored result with a separate read.