Reading & Writing Data: Sources & File Formats

Every Spark job reads data first and writes data last. How you read and write decides the speed, the cost and the safety of the job.

You know the storage ideas from "Storage Layers & File Formats": row and columnar files, Parquet, compression and folders named by value. This chapter shows the Spark code and what Spark does with each choice.

A DataFrameReader loads data into a DataFrame. You get one from spark.read.

A DataFrameWriter saves a DataFrame. You get one from df.write.

The data is the shop orders 101 to 106. Some files in this chapter are made up, and the text says so. New made-up orders have ids from 210.

The chapter has two halves.

1. Reading: the chain of calls, formats, read tasks, schemas, options, bad rows, databases and changing schemas.
2. Writing: partition folders, the number of files, and save modes.

Every read is one chain of calls. You pick a format, add options, give a schema, and load a path.

An option is a named setting for one read or write, such as header. A path is a file or a folder. A schema is the list of columns and their types.

Here is the chain. It reads the orders file and writes it back as Parquet.

orders = (spark.read
          .format("csv")
          .option("header", "true")
          .schema("order_id INT, customer_id INT, status STRING, amount DECIMAL(10,2), order_date DATE")
          .load("orders.csv"))

orders.write.mode("overwrite").parquet("orders_parquet")

The schema line gives each column a type. A later section explains it.

The mode line says what to do if the folder already exists. The save modes section explains that.

Each format also has a shortcut. spark.read.parquet(path) means format("parquet").load(path). The writer has the same shortcuts: csv, json, parquet and orc.

Spark reads many formats with the same chain. This table lists the ones you meet most.

Parquet and ORC store each column together, with the smallest and largest value of each block of rows. "Storage Layers & File Formats" explains why that helps. Now watch Spark use it.

Read the Parquet folder from the last section. Ask for orders above 103, and only two columns.

orders = spark.read.parquet("orders_parquet")
orders.filter("order_id > 103").select("order_id", "amount").explain()

# Plan (simplified):
# *(1) Project [order_id, amount]
# +- *(1) Filter (isnotnull(order_id) AND (order_id > 103))
#    +- *(1) FileScan parquet [order_id, amount]
#         PartitionFilters: []
#         PushedFilters: [IsNotNull(order_id), GreaterThan(order_id,103)]
#         ReadSchema: struct<order_id:int,amount:decimal(10,2)>

ReadSchema lists the two columns Spark reads. The columns customer_id, status and order_date stay on disk. This is column pruning.

PushedFilters is the filter that Spark hands to the Parquet reader. The reader can skip a block when its largest order_id is 103 or less. This is predicate pushdown.

With 6 rows nothing is skipped, but on big files whole blocks are.

Spark still shows its own Filter step and checks the rows it gets. A CSV file has no blocks and no statistics, so Spark must read and parse every line.

A task is one unit of work. One read task reads one piece of the input. "Partitioning & Parallelism" covers tasks in depth.

Spark cuts a file into pieces of at most spark.sql.files.maxPartitionBytes. The default is 128 MB.

Say you have one Parquet file of 1,024 MB. This is a made-up size.

history = spark.read.parquet("orders_history")   # made up: one 1,024 MB file
print(history.rdd.getNumPartitions())

Not every file can be cut. This table uses the same 1,024 MB size.

A gzip file cannot be cut, because you cannot start reading in the middle. One task reads all of it while the other cores wait. A multiLine JSON file also needs one task for the whole file.

Spark can also pack many small files into one piece. So the number of tasks is not always the number of files.

CSV and JSON files do not store types. Spark must learn them. You can let Spark guess, or you can give a schema.

To guess for CSV, you set inferSchema to true. Spark reads the data once to guess the types, then reads it again to load it. A big file costs two passes, and you see an extra job in the Spark UI.

Take a small made-up file with two columns of the orders, amounts included. Read it both ways.

guessed = (spark.read.option("header", "true")
          .option("inferSchema", "true")
          .csv("order_amounts.csv"))
guessed.printSchema()

given = (spark.read.option("header", "true")
         .schema("order_id INT, amount DECIMAL(10,2)")
         .csv("order_amounts.csv"))
given.printSchema()

You can also build a StructType object. The DDL string is shorter, so most code uses it.

For JSON, Spark guesses by default, and it reads the data to do so.

The option samplingRatio, for example 0.1, makes Spark guess from that share of the objects. It is faster, but a type that only appears in the rest is missed.

These options change how Spark reads text files. You set each one with .option("name", "value").

By default, Spark reads JSON as one object per line. This format is called JSON Lines. A file that spreads one object over many lines needs multiLine.

Here is a made-up file in that second style. It holds orders 101 and 102.

# orders_array.json (made up)
# [
#   {
#     "order_id": 101,
#     "status": "paid"
#   },
#   {
#     "order_id": 102,
#     "status": "paid"
#   }
# ]

orders = spark.read.option("multiLine", "true").json("orders_array.json")
orders.show()

A multiLine file cannot be cut into pieces. One task reads the whole file, as the read tasks section showed.

A bad row is a line that Spark cannot parse with your schema. An example is text in a number column. The read mode says what Spark does with such a line.

The mode is an option named mode. There are three values.

• PERMISSIVE is the default. Spark keeps the row and can store the raw line in a column.
• DROPMALFORMED drops the bad row without a message.
• FAILFAST stops the job at the first bad row.

For PERMISSIVE, you add a string column named _corrupt_record to your schema. Spark puts the raw bad line there. Without it, the row stays but the raw line is lost.

Here is a made-up file. The id of the second line is the text abc.

schema = "order_id INT, status STRING, amount DECIMAL(10,2), _corrupt_record STRING"

orders = (spark.read.option("header", "true")
          .option("mode", "PERMISSIVE")
          .schema(schema)
          .csv("orders_bad.csv"))

orders.select("order_id", "_corrupt_record").show(truncate=False)

Now read the same file with the other two modes.

JDBC is the standard way for Java programs to talk to a database. Spark uses it to read a database table.

By default Spark opens one connection and one task reads the whole table. To read in parallel, you give four options together. They are partitionColumn, lowerBound, upperBound and numPartitions.

Spark cuts the range of ids into steps of equal size. The size of one step is the stride.

Say the shop database holds the orders table, with order ids 101 to 106. Read it in 2 parts.

import os

orders = (spark.read.format("jdbc")
          .option("url", "jdbc:postgresql://db-host:5432/shop")
          .option("dbtable", "orders")
          .option("user", "reader")
          .option("password", os.environ["DB_PASSWORD"])
          .option("partitionColumn", "order_id")
          .option("lowerBound", "101")
          .option("upperBound", "107")
          .option("numPartitions", "2")
          .option("fetchsize", "1000")
          .load())

The column must be a number, a date or a timestamp. Pick one with evenly spread values.

The bounds do not filter rows. They only set the stride. Change them to 101 and 103 and look at what happens.

All 6 orders still come back. But the parts are uneven: 1 row against 5. The first part and the last part also take every row outside the bounds.

The option fetchsize is the number of rows each task gets from the database in one trip. A JDBC driver is the small library for one kind of database, and some fetch few rows by default. The Oracle JDBC driver fetches 10.

A larger fetchsize can speed up the read.

partitionBy writes one folder for each value of a column. You know this layout from "Storage Layers & File Formats". Here you write it with Spark.

Write the 6 orders from orders.csv with partitionBy on status. One task holds them all. Then read one status back.

# orders: the 6 orders read from orders.csv (one task)
orders.write.mode("overwrite").partitionBy("status").parquet("orders_by_status")

paid = spark.read.parquet("orders_by_status").filter("status = 'paid'")
paid.select("order_id").explain()

# Plan (simplified):
# *(1) Project [order_id]
# +- *(1) FileScan parquet [order_id,status]
#      PartitionFilters: [isnotnull(status), (status = paid)]
#      PushedFilters: []
#      ReadSchema: struct<order_id:int>

PartitionFilters shows partition pruning. Spark reads only the files in the folder status=paid. It never opens the files in the other two folders.

The query returns orders 101, 102, 104 and 106.

Pick a column with few values that you often filter on, such as the date. A column such as order_id makes one folder for each order.

Each task that holds rows writes its own file. With partitionBy, a task writes one file for each value it holds.

So the file count is tasks times values, at most. Take a made-up orders table read in 200 tasks, where every task holds all 3 statuses.

# orders_big: made up, 200 tasks, every task has all 3 statuses
orders_big.write.partitionBy("status").parquet("out_a")

orders_big.repartition("status").write.partitionBy("status").parquet("out_b")

The repartition has a cost. One task now writes all the paid rows. If paid holds most of the data, that task is slow.

The option maxRecordsPerFile caps the rows in one file. Take the 6 orders in one task, with a cap of 2 rows.

# orders: the 6 orders read from orders.csv (one task)
(orders.write.mode("overwrite")
       .partitionBy("status")
       .option("maxRecordsPerFile", 2)
       .parquet("orders_by_status_2"))

coalesce(n) lowers the number of tasks without a shuffle. repartition(n) shuffles the data into n even tasks. Both change the file count, because tasks write the files.

A save mode says what Spark does when the target already exists. You set it with .mode("..."). There are four.

The setting for overwrite is spark.sql.sources.partitionOverwriteMode. The default is static. Static overwrite deletes everything under the path, then writes.

The other value, dynamic, replaces only the partition folders that get new rows.

Take a table orders_by_month, partitioned by month. It holds January (orders 101 to 106) and a made-up February (orders 210 and 211). A new February batch arrives with orders 210, 211 and 212.

batch = spark.read.parquet("february_batch")   # made up: orders 210, 211, 212, month 2026-02

# Run one of these, each from the same starting table.

# A. static overwrite (the default)
batch.write.mode("overwrite").partitionBy("month").parquet("orders_by_month")

# B. dynamic overwrite
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")
batch.write.mode("overwrite").partitionBy("month").parquet("orders_by_month")

# C. append
batch.write.mode("append").partitionBy("month").parquet("orders_by_month")

Run B replaced only the month that had new rows. The batch had no January rows, so the January folder stayed as it was.

Over time you add columns. Old Parquet files do not have the new column. A folder can then hold files with different schemas.

Take a made-up folder orders_2026 with two files. The old file has order_id and amount. The new file also has a column channel.

By default, Spark takes the schema from the footer of one file. The footer is the end of a Parquet file, where it keeps its schema.

The new column can be missing. The option mergeSchema is off by default. Turn it on for one read.

orders = (spark.read.option("mergeSchema", "true")
          .parquet("orders_2026")
          .orderBy("order_id"))
orders.show()

Merging reads the footer of every file. That costs time when a folder has many files. Merging fails if one column has two different types in two files.

A plain folder of Parquet files has no record of its schema history. "Delta Lake & the Lakehouse" shows how a table format tracks it.

Five points to keep.

• Read with the chain: format, options, schema, path. Parquet lets Spark skip columns and blocks.
• Give a schema. Guessing reads the data twice, and it can pick a wrong type.
• PERMISSIVE keeps a bad row. DROPMALFORMED hides it. FAILFAST stops the job.
• Files written equal tasks times partition values, at most. Control them with repartition and maxRecordsPerFile.
• Overwrite on a partitioned table needs partitionOverwriteMode dynamic, or it deletes the other partition folders.

Answer these without looking back.

1. A JDBC read has lowerBound 101, upperBound 103 and numPartitions 2. Which part reads orders 102 to 106?
2. A gzip CSV file is 1,024 MB. How many read tasks does it get, and why?
3. Why can Spark skip blocks of a Parquet file, but not of a CSV file?
4. A job writes with partitionBy and 50 tasks. Each task holds 4 values. How many files can it write at most?