Reading & Writing CSV, JSON & Parquet

By the end of this lesson you will be able to read CSV, JSON and Parquet files into Spark with the right options, give a schema instead of guessing, deal with broken records, and write results with the correct save mode and partition folders.

Reading and writing files is most of a data engineer's day. The good news: Spark uses one pattern for every format, spark.read.<format>(path) to read and df.write.<format>(path) to write.

Step 1: read a CSV file

Here is a small orders file. Notice row 4: its amount is the text NA.

order_id,customer,city,amount,order_date
1,Aarav,Pune,1299,2026-10-01
2,Priya,Kochi,499,2026-10-01
3,Kabir,Jaipur,2450,2026-10-02
4,Meera,Pune,NA,2026-10-02

In PySpark, the quick way is to let Spark guess the types:

orders = spark.read.csv("orders.csv", header=True, inferSchema=True)
orders.printSchema()

root
 |-- order_id: integer (nullable = true)
 |-- customer: string (nullable = true)
 |-- city: string (nullable = true)
 |-- amount: string (nullable = true)
 |-- order_date: date (nullable = true)

header=True uses the first line as column names. inferSchema=True makes Spark read the data once to guess each column's type. Look at amount: because of the text NA in one row, Spark decided the whole column is a string. That is not what we want.

Better: give a schema

On real data, always pass a schema. It is faster (no extra pass to guess) and safer (bad values are caught instead of silently changing the column type).

schema = "order_id INT, customer STRING, city STRING, amount INT, order_date DATE"

orders = (
    spark.read
    .option("header", True)
    .option("nullValue", "NA")
    .schema(schema)
    .csv("orders.csv")
)
orders.show()

+--------+--------+------+------+----------+
|order_id|customer|  city|amount|order_date|
+--------+--------+------+------+----------+
|       1|   Aarav|  Pune|  1299|2026-10-01|
|       2|   Priya| Kochi|   499|2026-10-01|
|       3|   Kabir|Jaipur|  2450|2026-10-02|
|       4|   Meera|  Pune|  NULL|2026-10-02|
+--------+--------+------+------+----------+

nullValue tells Spark which text means "missing". Spark 4 prints missing values as NULL in show().

The pandas equivalent, which you can run here, uses na_values and dtype:

import io
import pandas as pd

csv_text = """order_id,customer,city,amount,order_date
1,Aarav,Pune,1299,2026-10-01
2,Priya,Kochi,499,2026-10-01
3,Kabir,Jaipur,2450,2026-10-02
4,Meera,Pune,NA,2026-10-02
"""
orders = pd.read_csv(
    io.StringIO(csv_text),
    na_values=["NA"],
    dtype={"amount": "Int64"},
    parse_dates=["order_date"],
)
print(orders)
print(orders.dtypes)

   order_id customer    city  amount order_date
0         1    Aarav    Pune    1299 2026-10-01
1         2    Priya   Kochi     499 2026-10-01
2         3    Kabir  Jaipur    2450 2026-10-02
3         4    Meera    Pune    <NA> 2026-10-02
order_id               int64
customer                 str
city                     str
amount                 Int64
order_date    datetime64[us]
dtype: object

Step 2: decide what happens to broken rows

Real files have broken lines: a missing comma, text in a number column. The CSV and JSON readers have a mode option:

In permissive mode you can also keep the raw text of bad rows in an extra column, so you can inspect them later. Add a string column such as _corrupt_record to your schema and set .option("columnNameOfCorruptRecord", "_corrupt_record").

Here is the idea in plain Python, so you can watch each row being checked:

import csv, io

raw = """order_id,city,amount
1,Pune,1299
2,Kochi,four hundred
3,Jaipur
4,Lucknow,350
"""
good, bad = [], []
for row in csv.DictReader(io.StringIO(raw)):
    try:
        good.append({"order_id": int(row["order_id"]), "city": row["city"], "amount": int(row["amount"])})
    except (TypeError, ValueError):
        bad.append(row)

print("good rows:", good)
print("quarantined:", bad)

good rows: [{'order_id': 1, 'city': 'Pune', 'amount': 1299}, {'order_id': 4, 'city': 'Lucknow', 'amount': 350}]
quarantined: [{'order_id': '2', 'city': 'Kochi', 'amount': 'four hundred'}, {'order_id': '3', 'city': 'Jaipur', 'amount': None}]

Keeping bad rows aside in a quarantine table is a common production pattern. You will build it properly in Data Quality for Pipelines.

Step 3: read JSON

Spark expects JSON Lines by default: one complete JSON object per line. This is the usual format for logs and event streams.

{"user": "u1", "event": "view", "page": "/phones", "ts": "2026-11-01T10:00:05"}
{"user": "u2", "event": "add_to_cart", "page": "/shoes", "ts": "2026-11-01T10:00:09"}

events = spark.read.json("events/2026-11-01/")

You can pass a folder, and Spark reads every file in it. If a file is one big JSON array spread over many lines, add .option("multiLine", True). Nested JSON objects become struct columns, which you will meet in Nulls, Dates & Complex Types.

Step 4: read and write Parquet

Parquet is a column-based binary file format. It stores the schema inside the file, compresses well, and lets Spark read only the columns and row groups a query needs. For analytics, Parquet is the normal choice. You will compare formats in File Formats: Parquet, ORC & Avro.

orders.write.mode("overwrite").parquet("clean/orders")
back = spark.read.parquet("clean/orders")
back.printSchema()     # types come back exactly; no inferSchema needed

How Spark writes: a folder, not a file

This surprises everyone the first time. df.write.csv("out/orders") creates a folder, with one file per partition:

out/orders/
  _SUCCESS
  part-00000-....snappy.parquet
  part-00001-....snappy.parquet

Each executor task writes its own partition at the same time, which is why writing is fast. The empty _SUCCESS file is written last, so other jobs can check that the write finished.

Save modes

Partitioned folders

partitionBy writes one sub-folder per value of a column. Later queries that filter on that column read only the matching folders.

orders.write.mode("overwrite").partitionBy("order_date").parquet("clean/orders")

clean/orders/
  order_date=2026-10-01/part-....parquet
  order_date=2026-10-02/part-....parquet

The Python below copies this layout with pandas and plain files, so you can see the folder names it makes.

import os
import pandas as pd

orders = pd.DataFrame({
    "order_id": [1, 2, 3, 4],
    "city": ["Pune", "Kochi", "Jaipur", "Pune"],
    "amount": [1299, 499, 2450, 89],
    "order_date": ["2026-10-01", "2026-10-01", "2026-10-02", "2026-10-02"],
})
for day, part in orders.groupby("order_date"):
    folder = f"clean/orders/order_date={day}"
    os.makedirs(folder, exist_ok=True)
    part.drop(columns="order_date").to_csv(f"{folder}/part-00000.csv", index=False)

for root, dirs, files in sorted(os.walk("clean/orders")):
    for f in files:
        print(os.path.join(root, f))

clean/orders/order_date=2026-10-01/part-00000.csv
clean/orders/order_date=2026-10-02/part-00000.csv

The order_date column is not stored inside the files; its value lives in the folder name. Spark adds it back as a column when it reads the folder.

Common mistakes

• Using inferSchema in production. It reads data twice and can guess wrong. Give a schema.
• Expecting one output file. Spark writes a folder of part files. Use coalesce(1) only for small results.
• Forgetting the save mode. The default fails if the folder exists; overwrite replaces it, so be careful.
• Partitioning by a column with many values, such as user_id. You get millions of tiny folders.

Key points

• spark.read and df.write use one pattern for CSV, JSON, Parquet and other formats.
• Give a schema; use options such as header, nullValue and mode to control parsing.
• PERMISSIVE, DROPMALFORMED and FAILFAST decide what happens to broken rows.
• Spark writes a folder of part files plus _SUCCESS; save modes control existing data.
• partitionBy creates column=value folders that later queries can skip.

clicks = (
    spark.read
    .schema("user STRING, event STRING, page STRING, ts TIMESTAMP")
    .option("mode", "FAILFAST")
    .json("raw/clicks/")
)
(clicks.write
    .mode("overwrite")
    .partitionBy("event")
    .parquet("clean/clicks/"))