Why is inferSchema expensive?

Spark has to read the data once to work out the types, then read it again to load it. On a large CSV that doubles your I/O before any real work starts. An explicit schema skips the first pass entirely and also stops types from changing when the data changes.

Why is Parquet so much faster than CSV?

Parquet is columnar and typed, so reading three columns out of fifty reads only those three. It also stores min/max statistics per row group, letting Spark skip entire chunks that cannot match a filter. CSV has to be parsed row by row, in full, every time.

What does partitionBy actually do?

It writes one directory per distinct value, named like country=UK/. A later read with a filter on that column reads only the matching directories — partition pruning. It only pays off on columns you filter by often and that have modest cardinality.

Why did my write produce 200 small files?

Spark writes one file per partition, and operations involving a shuffle default to 200 partitions. Call coalesce or repartition before writing, or set spark.sql.shuffle.partitions, to control the file count.

Reading and Writing Data in PySpark

Real data comes from files, and how you read them decides how fast everything downstream runs. This lesson covers the three formats you will actually meet.

Some data to work with

mkdir -p data && cat > data/sales.csv <<'EOF'
order_id,customer,country,amount,order_date
1001,alice,UK,250.00,2026-01-15
1002,bob,US,180.50,2026-01-16
1003,carol,UK,320.75,2026-01-16
1004,dave,DE,95.00,2026-01-17
1005,erin,US,410.20,2026-01-18
EOF
wc -l data/sales.csv

6 data/sales.csv

Reading CSV — the slow way and the right way

Schema inference works but costs an extra pass over the data:

from pyspark.sql import SparkSession
import time

spark = SparkSession.builder.appName("io").master("local[*]").getOrCreate()

start = time.perf_counter()
df = spark.read.csv("data/sales.csv", header=True, inferSchema=True)
print(f"inferSchema read: {time.perf_counter() - start:.3f}s")
df.printSchema()

inferSchema read: 1.842s
root
 |-- order_id: integer (nullable = true)
 |-- customer: string (nullable = true)
 |-- country: string (nullable = true)
 |-- amount: double (nullable = true)
 |-- order_date: date (nullable = true)

Declare the schema instead:

from pyspark.sql.types import (
    StructType, StructField, IntegerType, StringType, DecimalType, DateType
)

schema = StructType([
    StructField("order_id",   IntegerType(),      False),
    StructField("customer",   StringType(),       False),
    StructField("country",    StringType(),       False),
    StructField("amount",     DecimalType(10, 2), False),
    StructField("order_date", DateType(),         False),
])

start = time.perf_counter()
df = spark.read.csv("data/sales.csv", header=True, schema=schema)
print(f"explicit schema read: {time.perf_counter() - start:.3f}s")
df.show()

explicit schema read: 0.213s
+--------+--------+-------+------+----------+
|order_id|customer|country|amount|order_date|
+--------+--------+-------+------+----------+
|    1001|   alice|     UK|250.00|2026-01-15|
|    1002|     bob|     US|180.50|2026-01-16|
|    1003|   carol|     UK|320.75|2026-01-16|
|    1004|    dave|     DE| 95.00|2026-01-17|
|    1005|    erin|     US|410.20|2026-01-18|
+--------+--------+-------+------+----------+

Nine times faster on five rows, and the gap widens with file size. There is a correctness argument too: inferSchema gave double for money, which cannot represent decimal amounts exactly. The explicit schema uses DecimalType(10, 2), which can.

Malformed rows

Add a bad row:

echo '1006,frank,FR,not-a-number,2026-01-19' >> data/sales.csv

By default Spark quietly nulls the bad field:

spark.read.csv("data/sales.csv", header=True, schema=schema).show()

+--------+--------+-------+------+----------+
|order_id|customer|country|amount|order_date|
+--------+--------+-------+------+----------+
...
|    1006|   frank|     FR|  NULL|2026-01-19|
+--------+--------+-------+------+----------+

A null where money should be, with no warning. That silence is the danger — the row flows into your aggregations and quietly understates a total.

Make the choice explicit. To fail loudly:

spark.read.csv("data/sales.csv", header=True, schema=schema, mode="FAILFAST").show()

pyspark.errors.exceptions.captured.SparkException:
[MALFORMED_RECORD_IN_PARSING] Malformed records are detected in record parsing: [1006,frank,FR,null,2026-01-19].

To keep the good rows and quarantine the bad ones:

quarantine_schema = schema.add(StructField("_corrupt_record", StringType(), True))

df = spark.read.csv("data/sales.csv", header=True, schema=quarantine_schema,
                    mode="PERMISSIVE", columnNameOfCorruptRecord="_corrupt_record")

clean = df.filter(df._corrupt_record.isNull()).drop("_corrupt_record")
bad   = df.filter(df._corrupt_record.isNotNull()).select("_corrupt_record")

print(f"clean rows: {clean.count()}   bad rows: {bad.count()}")
bad.show(truncate=False)

clean rows: 5   bad rows: 1
+-----------------------------------------+
|_corrupt_record                          |
+-----------------------------------------+
|1006,frank,FR,not-a-number,2026-01-19    |
+-----------------------------------------+

Now the bad row is data you can inspect and report, rather than a null nobody notices.

JSON

cat > data/events.json <<'EOF'
{"user": "alice", "action": "login",  "meta": {"ip": "10.0.0.1", "device": "mobile"}}
{"user": "bob",   "action": "search", "meta": {"ip": "10.0.0.2", "device": "desktop"}}
{"user": "alice", "action": "logout", "meta": {"ip": "10.0.0.1", "device": "mobile"}}
EOF

Spark expects one JSON object per line (JSON Lines), not a JSON array:

events = spark.read.json("data/events.json")
events.printSchema()
events.show(truncate=False)

root
 |-- action: string (nullable = true)
 |-- meta: struct (nullable = true)
 |    |-- device: string (nullable = true)
 |    |-- ip: string (nullable = true)
 |-- user: string (nullable = true)

+------+---------------------+-----+
|action|meta                 |user |
+------+---------------------+-----+
|login |{mobile, 10.0.0.1}   |alice|
|search|{desktop, 10.0.0.2}  |bob  |
|logout|{mobile, 10.0.0.1}   |alice|
+------+---------------------+-----+

Nested fields are read with dot notation:

events.select("user", "action", "meta.device").show()

+-----+------+-------+
| user|action| device|
+-----+------+-------+
|alice| login| mobile|
|  bob|search|desktop|
|alice|logout| mobile|
+-----+------+-------+

For a pretty-printed multi-line JSON file, pass multiLine=True — but note that a multi-line file cannot be split across executors, so it is read by one core no matter how large it is.

Parquet, and why it wins

clean.write.mode("overwrite").parquet("data/sales_parquet")

ls -la data/sales_parquet/ && du -sh data/sales.csv data/sales_parquet

_SUCCESS
part-00000-8f2a1c94-3e7b-4d21-9a5c-6b1e2f9d3a70-c000.snappy.parquet

4.0K	data/sales.csv
8.0K	data/sales_parquet

On six rows Parquet is bigger — the format header and footer dominate. Scale up to see the real behaviour:

from pyspark.sql import functions as F

big = clean.crossJoin(spark.range(200_000).select(F.col("id").alias("n")))
big.write.mode("overwrite").parquet("data/big_parquet")
big.write.mode("overwrite").option("header", True).csv("data/big_csv")

$ du -sh data/big_csv data/big_parquet
 42M	data/big_csv
4.1M	data/big_parquet

Ten times smaller. And reading a subset of columns:

start = time.perf_counter()
spark.read.parquet("data/big_parquet").select("country").count()
print(f"parquet, one column: {time.perf_counter() - start:.3f}s")

start = time.perf_counter()
spark.read.csv("data/big_csv", header=True, schema=schema).select("country").count()
print(f"csv,     one column: {time.perf_counter() - start:.3f}s")

parquet, one column: 0.284s
csv,     one column: 2.117s

Parquet reads only the country column off disk. CSV has to parse every row in full to find where the third field ends. Use Parquet for anything you will read more than once.

Partitioned writes

clean.write.mode("overwrite").partitionBy("country").parquet("data/sales_by_country")

find data/sales_by_country -type d

data/sales_by_country
data/sales_by_country/country=DE
data/sales_by_country/country=UK
data/sales_by_country/country=US

One directory per value. A filtered read now touches only the matching directory:

uk = spark.read.parquet("data/sales_by_country").filter("country = 'UK'")
uk.explain()

== Physical Plan ==
*(1) ColumnarToRow
+- FileScan parquet [order_id#12,customer#13,amount#14,order_date#15,country#16]
   Batched: true, DataFilters: [],
   PartitionFilters: [isnotnull(country#16), (country#16 = UK)],
   PushedFilters: [], ReadSchema: struct<...>

PartitionFilters is the line that matters — Spark resolved the filter against directory names and never opened the US or DE files at all.

Partition on columns you filter by often and that have low-to-moderate cardinality. Partition by something like order_id and you get one tiny directory per row, which is far slower than no partitioning at all.

Controlling output file count

big.write.mode("overwrite").parquet("data/many_files")

ls data/many_files/*.parquet | wc -l

One file per partition. After a shuffle you would get 200 by default, most of them tiny. Coalesce first:

big.coalesce(1).write.mode("overwrite").parquet("data/one_file")

$ ls data/one_file/*.parquet | wc -l
1

Use coalesce(n) to reduce partitions without a shuffle, and repartition(n) when you need an even redistribution and can afford one. Aim for files of roughly 128 MB to 1 GB.

Write modes

Mode	Behaviour
`error` (default)	fails if the path exists
`overwrite`	replaces the path
`append`	adds new files alongside existing ones
`ignore`	silently does nothing if the path exists

overwrite with partitionBy replaces every partition by default, not just the ones in your DataFrame. To replace only the partitions you are writing:

spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

This one has destroyed real datasets. Set it before any partitioned overwrite of a table you are only updating in part.

Practice

1. Read the CSV with DecimalType(10,2) and with DoubleType. Sum the amounts and compare.

print("decimal:", clean.agg(F.sum("amount")).first()[0])
print("double: ", double_df.agg(F.sum("amount")).first()[0])

decimal: 1256.45
double:  1256.4499999999998

Binary floating point cannot represent 0.10 or 0.20 exactly, so the errors accumulate. Use DecimalType for money — always.

2. Write partitioned by order_date on a year of daily data. How many directories?

365, each holding one small file. Reads filtered by an exact date are fast, but a full scan now opens 365 files instead of a handful, and the metadata listing alone can dominate the job. Partition by month for daily data, or by date only when queries always filter on it.

3. Write with mode("append") twice. What happens to row count?

after first write:  5
after second write: 10

Append adds files without checking for duplicates — Spark has no primary key. Deduplication is your job, either with dropDuplicates before writing or by using a table format like Delta or Iceberg that supports merge.

4. Read a Parquet file written with a five-column schema, selecting one column. Confirm column pruning in the plan.

FileScan parquet [country#16] ... ReadSchema: struct<country:string>

ReadSchema lists only country. Spark pushed the projection into the file reader, so the other four columns are never read off disk. This is the single biggest reason to prefer Parquet over CSV.

Next: transforming columns — expressions, conditionals, and the functions you will use daily.