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

A PySpark DataFrame is a table of named, typed columns that describes a computation instead of holding the data. Each transformation, such as filter, withColumn or join, adds a step to a plan, and nothing runs until an action, such as show, count or write, asks for a result. Spark then optimises the whole plan and runs it, on your laptop or across a cluster, with the same code.

- Page: https://tecyfy.com/data-engineering/course/pyspark-dataframes
- Course: Data engineering course (https://tecyfy.com/data-engineering/course), Module 2: PySpark DataFrames, Read files with a schema
- Parts of this module: Read files with a schema (https://tecyfy.com/data-engineering/course/pyspark-dataframes); Keep the newest row (https://tecyfy.com/data-engineering/course/pyspark-dataframes/keep-the-newest-row); Join, aggregate and write (https://tecyfy.com/data-engineering/course/pyspark-dataframes/join-aggregate-and-write); Serverless, DuckDB and Polars (https://tecyfy.com/data-engineering/course/pyspark-dataframes/serverless-duckdb-polars); Build it: clean the raw orders (https://tecyfy.com/data-engineering/course/pyspark-dataframes/build-it)

At the end of this part you can:

- Start Spark on your laptop or on Databricks Free Edition, with the module's data
- Read CSV files with a schema, and keep the rows that do not parse instead of losing them
- Save what you read before you split it, in a way that also works on serverless
- Read a query plan with explain() and say what Spark will read

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:

```bash
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.

```python
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:

```python
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:

```python
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:

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

```text
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:

```python
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)
```

```text
+--------+------+---------------------------------------------------------------+
|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.

(An interactive demo goes here on the page: https://tecyfy.com/data-engineering/course/pyspark-dataframes)

## Save what you read, then split it

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

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

```text
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](/data-engineering/course/pyspark-dataframes/serverless-duckdb-polars) 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:

```python
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())
```

```text
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 Architecture](https://tecyfy.com/concepts/medallion-architecture) 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:

```python
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.

```python
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:

```text
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.

(An interactive demo goes here on the page: https://tecyfy.com/data-engineering/course/pyspark-dataframes)

Now ask for the result:

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

```text
+--------+-----------+------+----------+
|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](/data-engineering/course/pyspark-dataframes/keep-the-newest-row) fixes both, by keeping only the newest version of each order.

## Check yourself

1. You read orders.csv with inferSchema=True. Its amount column has 2 million numbers and 1 value of "n/a", and Spark makes the column a string. What should you change?

   A. Turn on DROPMALFORMED so Spark drops the row with "n/a" while it infers the schema
   B. Give the reader a schema with amount as a decimal, so "n/a" becomes null and the row is kept in a corrupt record column
   C. Cast the column to a decimal after reading, because inference always reads numbers as strings
   D. Raise the number of rows Spark samples so it sees more numbers than text

2. Your notebook on Databricks serverless reads raw JSON with a _bad column for corrupt records, then calls df.cache() so it can count the bad rows, as Spark's own error message suggested. The cell fails. What is the fix that works on serverless?

   A. Use df.persist() instead, since serverless refuses only cache()
   B. Call spark.sparkContext.setCheckpointDir() and checkpoint the DataFrame
   C. Read the file with an RDD so the corrupt record column is not needed
   D. Save the parsed rows as a table first, then count the bad rows from the table

### Answers

1. B. Inference picks the type that fits every value it reads, so 1 value of text makes the whole column a string, and it costs an extra pass over the data. A schema fixes the type you need; in PERMISSIVE mode the bad value becomes null and the whole line is kept in the corrupt record column, so you can count and fix it. DROPMALFORMED would lose the row without telling you, and a cast after reading turns "n/a" into null with no record of what it was.
2. D. Serverless runs only Spark Connect, which does not support cache, persist or checkpoint, and has no sparkContext or RDDs. Spark refuses a query on a raw file that uses only the corrupt record column, so you save what you read first, which is the bronze table a pipeline keeps anyway, and query that.

## Questions

### What is a DataFrame in PySpark?

A PySpark DataFrame is a table with named columns, each with a type, that Spark can spread across many machines. It does not hold the data the way a pandas DataFrame does. It describes how to compute the data, and Spark runs that description when an action such as show, count or write asks for a result.

### What is the difference between a transformation and an action in Spark?

A transformation, such as select, filter, withColumn, join or groupBy, returns a new DataFrame and only adds a step to the plan. An action, such as show, count, collect or write, needs a result, so it makes Spark optimise the whole plan and run it. Nothing is read from disk until the first action.

### How do I read a CSV file in PySpark with a schema?

Pass the schema to the reader before the file, as a DDL string such as "order_id BIGINT, amount DECIMAL(10,2)" or as a StructType, with spark.read.schema(schema).option("header", True).csv(path). A schema saves the extra pass over the data that inferSchema makes, and the columns get the types you chose instead of whatever fits every value.

### Why does Spark refuse to count the rows of my corrupt record column?

Spark does not allow a query on a raw CSV, JSON or XML file whose columns are only the corrupt record column, and fails with UNSUPPORTED_FEATURE.QUERY_ONLY_CORRUPT_RECORD_COLUMN. Save what you read, as Parquet files or a table, and count the bad rows from the saved copy. Caching also works on a laptop, but not on Databricks serverless.

## Sources

- [Spark overview](https://spark.apache.org/docs/latest/)
- [Installation](https://spark.apache.org/docs/latest/api/python/getting_started/install.html)
- [RDD programming guide](https://spark.apache.org/docs/latest/rdd-programming-guide.html)
- [CSV files](https://spark.apache.org/docs/latest/sql-data-sources-csv.html)
- [DataFrame.explain](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.explain.html)
- [Serverless compute limitations](https://docs.databricks.com/aws/en/compute/serverless/limitations)
- [Databricks Free Edition](https://docs.databricks.com/aws/en/getting-started/free-edition)
- [Work with files in Unity Catalog volumes](https://docs.databricks.com/aws/en/volumes/volume-files)
- [Work with files on Databricks](https://docs.databricks.com/aws/en/files/)
- [CREATE VOLUME](https://docs.databricks.com/aws/en/sql/language-manual/sql-ref-syntax-ddl-create-volume)
- [What is Delta Lake?](https://docs.databricks.com/aws/en/delta/)