PySpark DataFrames: read CSV files with a schema and keep the bad rows

Most data engineering code is a handful of moves repeated: read a file, fix its types, throw out what is broken, keep the newest version of each row, join it to something, add it up and write the result. This module makes those moves in PySpark, the Python interface to Apache Spark, on a small set of orders from an online shop. The module has 5 parts, and by the end of the last you have the first step of the course's pipeline: a script that turns the shop's raw order files into clean tables, with the rows it rejected kept for someone to fix. This part sets up, reads the files and shows how Spark plans the work.

The data is small on purpose. You can see every row, so you can check every answer by eye. The same code runs on a billion rows across a cluster, which is the point of Spark: you describe what you want once, and Spark decides how to spread the work.

Set up

You can run this module on your own computer or on Databricks Free Edition. Both are free.

On your computer you need Python 3.10 or later and Java 17, 21 or 25, because Spark runs on the Java virtual machine and PySpark talks to it. Install PySpark with its SQL extras, which bring pandas and PyArrow:

pip install "pyspark[sql]==4.2.0"

On Databricks Free Edition, create a notebook. It runs on serverless compute, and a Spark session called spark is already there, so skip the session step below.

Now make the data. This script writes 2 small files, the shop's orders and its customers. Read the orders closely, because each problem in them is there for a reason: order 1001 and 1002 change status, so they appear more than once; order 1003 has an amount of abc; order 1004 is in the file twice, exactly the same; and order 1005 belongs to customer C09, who is not in the customers file.

from pathlib import Path

RAW = Path("raw")  # on Databricks: Path(f"/Volumes/{catalog}/{schema}/raw")
RAW.mkdir(parents=True, exist_ok=True)

(RAW / "orders.csv").write_text("""order_id,customer_id,status,amount,currency,ordered_at,updated_at
1001,C01,placed,25.50,EUR,2026-10-01T09:15:00,2026-10-01T09:15:00
1002,C02,placed,12.00,EUR,2026-10-01T10:02:00,2026-10-01T10:02:00
1001,C01,shipped,25.50,EUR,2026-10-01T09:15:00,2026-10-02T08:00:00
1003,C03,placed,abc,EUR,2026-10-02T11:30:00,2026-10-02T11:30:00
1004,C01,placed,40.00,EUR,2026-10-02T12:45:00,2026-10-02T12:45:00
1002,C02,cancelled,12.00,EUR,2026-10-01T10:02:00,2026-10-03T09:00:00
1005,C09,placed,8.75,EUR,2026-10-03T14:10:00,2026-10-03T14:10:00
1004,C01,placed,40.00,EUR,2026-10-02T12:45:00,2026-10-02T12:45:00
""")
(RAW / "customers.csv").write_text("""customer_id,name,country
C01,Ana,PT
C02,Ben,IE
C03,Chen,DE
""")

On Databricks, the files go in a Unity Catalog volume, a governed folder for files. Create one in the catalog and schema your notebook is using, then run the script above with RAW set to the volume's path:

catalog = spark.catalog.currentCatalog()
schema = spark.catalog.currentDatabase()
spark.sql(f"CREATE VOLUME IF NOT EXISTS {catalog}.{schema}.raw")

A volume has an ordinary file path, /Volumes/<catalog>/<schema>/<volume>, so plain Python can write into it. If you prefer, upload the 2 files instead, from New, then Add or upload data, then Upload files to a volume. Asking Spark for the current catalog, rather than typing a name, means the code works in any workspace.

Last, start Spark. On your computer this starts a local Spark that uses every core:

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.appName("orders").getOrCreate()
RAW, LAKE = "raw", "lake"  # where the files are, and where this module writes

Importing the functions module as F is the usual convention. Almost everything you do to a column, from F.col to F.sum, is in it. On Databricks, run only the import line and set RAW to your volume's path.

A DataFrame is a plan, not the data

The first thing to understand about Spark changes how you read every line after it. When you write a transformation, Spark does not do it. It writes it down.

A transformation is anything that returns a new DataFrame: select, filter, withColumn, join, groupBy. An action is anything that needs an answer: show, count, collect, write. Spark is lazy. Transformations only add steps to a plan, and the first action makes Spark read the data and run the whole plan at once.

That is not a quirk; it is why Spark is fast. Because it sees the whole plan before it starts, Spark can read only the columns the plan uses, apply filters while it reads the file instead of after, and choose how to run each join. You will see the first 2 in a plan below, and the third in part 3. It also means a mistake in a line can show up later, at the action, which takes some getting used to.

Read files with a schema

The quick way to read a CSV asks Spark to guess the types:

guess = spark.read.csv(f"{RAW}/orders.csv", header=True, inferSchema=True)
guess.printSchema()

root
 |-- order_id: integer (nullable = true)
 |-- customer_id: string (nullable = true)
 |-- status: string (nullable = true)
 |-- amount: string (nullable = true)
 |-- currency: string (nullable = true)
 |-- ordered_at: timestamp (nullable = true)
 |-- updated_at: timestamp (nullable = true)

Look at amount. It is a string, because a single value of abc does not fit a number, and inference picks the type every value fits. Any sum you write on it now is wrong or fails. Inference also reads the file an extra time just to guess, which on a large file is a whole extra pass.

In a pipeline, give the schema yourself. A DDL string, the same column list a CREATE TABLE takes, is the shortest way:

ORDERS = """
    order_id BIGINT, customer_id STRING, status STRING, amount DECIMAL(10,2),
    currency STRING, ordered_at TIMESTAMP, updated_at TIMESTAMP, _bad STRING
"""
orders = (
    spark.read.schema(ORDERS)
    .option("header", True)
    .option("mode", "PERMISSIVE")
    .option("columnNameOfCorruptRecord", "_bad")
    .csv(f"{RAW}/orders.csv")
)
orders.select("order_id", "amount", "_bad").show(truncate=False)

+--------+------+---------------------------------------------------------------+
|order_id|amount|_bad                                                           |
+--------+------+---------------------------------------------------------------+
|1001    |25.50 |NULL                                                           |
|1002    |12.00 |NULL                                                           |
|1001    |25.50 |NULL                                                           |
|1003    |NULL  |1003,C03,placed,abc,EUR,2026-10-02T11:30:00,2026-10-02T11:30:00|
|1004    |40.00 |NULL                                                           |
|1002    |12.00 |NULL                                                           |
|1005    |8.75  |NULL                                                           |
|1004    |40.00 |NULL                                                           |
+--------+------+---------------------------------------------------------------+

Money is a DECIMAL, never a DOUBLE, because a double cannot hold 0.10 exactly and the cents drift when you add up millions of rows.

The 2 options are how you keep bad rows instead of losing them. PERMISSIVE is the default mode: when a line does not parse, Spark sets the fields it could not read to null, and if the schema has a string column with the name you give columnNameOfCorruptRecord, it puts the whole original line there. Order 1003 is now visible, with the exact text that was wrong. The other modes are worse for a pipeline: DROPMALFORMED drops the line without telling you, and FAILFAST stops the whole read at the first bad line.

spark.read.csv(path, header=True, inferSchema=True)

Save what you read, then split it

You might expect to count the bad rows straight away. Try it, and Spark refuses:

orders.filter(F.col("_bad").isNotNull()).count()

AnalysisException: [UNSUPPORTED_FEATURE.QUERY_ONLY_CORRUPT_RECORD_COLUMN] The feature is not
supported: Queries from raw JSON/CSV/XML files are disallowed when the referenced columns only
include the internal corrupt record column

Spark's message suggests caching the DataFrame first. Do not follow it: caching is not available on Databricks serverless, as part 4 explains. The fix that works everywhere is the one a pipeline wants anyway. Save the rows exactly as you read them, then work from the saved copy:

orders.write.mode("overwrite").parquet(f"{LAKE}/bronze_orders")
bronze = spark.read.parquet(f"{LAKE}/bronze_orders")

bad = bronze.filter(F.col("_bad").isNotNull())
good = bronze.filter(F.col("_bad").isNull()).drop("_bad")
print(bad.count(), good.count())

1 7

Parquet is the file format almost every data tool reads: it stores each column together, compressed, with its type, so a query reads only the columns it needs. The saved copy is what the Medallion ArchitectureMedallion ArchitectureOrganising a lakehouse in 3 layers of rising quality: bronze for raw data, silver for cleaned and joined data, and gold for tables shaped for reports and AI. calls bronze, the raw data kept as it arrived. Module 7 builds the full bronze, silver and gold pipeline; for now, keeping the raw copy means you can always rebuild everything after it.

On Databricks, save a table instead of a folder of files. Tables are Delta Lake tables by default, which module 4 covers:

orders.write.mode("overwrite").saveAsTable(f"{catalog}.{schema}.bronze_orders")
bronze = spark.table(f"{catalog}.{schema}.bronze_orders")

Select, filter and see the plan

Here is a short plan: the orders that are still placed, with the day each was made.

placed = (
    good.filter(F.col("status") == "placed")
    .withColumn("order_date", F.to_date("ordered_at"))
    .select("order_id", "customer_id", "amount", "order_date")
)
placed.explain()

Nothing has run yet. explain() prints the plan Spark made, read from the bottom up. Trimmed to its last line, the file scan, it says:

FileScan parquet [order_id#66L,customer_id#67,status#68,amount#69,ordered_at#71,_bad#73]
  PushedFilters: [IsNotNull(status), IsNull(_bad), EqualTo(status,placed)]
  ReadSchema: struct<order_id:bigint,customer_id:string,status:string,amount:decimal(10,2),...

That is the laziness paying off. ReadSchema lists only the columns the plan uses, so currency and updated_at are never read from disk. PushedFilters shows both filters, yours on status and the earlier one on _bad, handed to the Parquet reader, which can skip whole blocks of a file that cannot match. The numbers after each column, such as #66, are Spark's internal ids and change from run to run.

placed = (
    good
    .filter(F.col("status") == "placed")
    .withColumn("order_date", F.to_date("ordered_at"))
    .select("order_id", "customer_id", "amount", "order_date")
)

Now ask for the result:

placed.orderBy("order_id").show()

+--------+-----------+------+----------+
|order_id|customer_id|amount|order_date|
+--------+-----------+------+----------+
|    1001|        C01| 25.50|2026-10-01|
|    1002|        C02| 12.00|2026-10-01|
|    1004|        C01| 40.00|2026-10-02|
|    1004|        C01| 40.00|2026-10-02|
|    1005|        C09|  8.75|2026-10-03|
+--------+-----------+------+----------+

This answer is wrong, and the data says why. Orders 1001 and 1002 have moved on, to shipped and cancelled, but their old placed rows still match. Order 1004 is counted twice. Part 2 fixes both, by keeping only the newest version of each order.