Spark File I/O: Formats, Save Modes and partitionBy

1.One interface for every file format

Spark exposes data sources through one pair of builders. spark.read returns a DataFrameReader, described in the reference as the interface "used to load a DataFrame from external storage systems". df.write returns its mirror, a DataFrameWriter. Both work the same way. You can name the format with format("csv") and finish with load(path) or save(path), or you can call a shortcut method that names the format for you, such as csv(path), json(path), parquet(path), orc(path) or text(path). Between those two ends you add option(key, value) calls, and on the read side you can also call schema(...).

# With options
df = spark.read.format("csv") \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .load("path/to/file.csv")

Inferring a schema has a cost. The Parquet guide recommends that you "Specify a schema when reading Parquet files to avoid the overhead of schema inference." You can pass the schema to schema() as a StructType or as a DDL string such as "name STRING, age INT".

The default format also matters. When you call save() or saveAsTable() without a format, what you get depends on the platform. The documentation states that Databricks uses Delta Lake as its default protocol for reading and writing data and tables, while Apache Spark uses Parquet. The formats group into three families: table formats (Delta Lake, Iceberg), binary columnar formats (Parquet, ORC), and human-readable formats (JSON, CSV, XML, Text).

Writing works the same way for every format, and so does overwriting. When the target location already holds data, you replace it by setting the save mode to overwrite. The writer reference defines it as "Overwrite existing data." You can chain .mode("overwrite") before save(), or pass mode="overwrite" to a format shortcut such as parquet(), orc() or jdbc(). The same call applies to a file path, to saveAsTable(), and to a database table written through JDBC or a bundled connector, where it replaces the table's contents. Without it, the default mode refuses to write over existing data (see the next section for all four modes). The Parquet guide shows the pattern: the first write creates the files, and a later write with mode("overwrite") to the same path replaces them.

df.write.format("parquet").mode("overwrite").save("/Volumes/<catalog>/<schema>/<volume>/reviews_parquet")

Sources12345

2.Save modes: what happens when the target already exists

mode(saveMode) "Specifies the behavior when data or table already exists." It accepts four values. Two of them (error and errorifexists) are spellings of the same mode. You can set the mode by chaining .mode("overwrite"), or by passing it as a keyword to a format method, as in df.write.parquet(d, mode="overwrite"). The Parquet writer's signature lists 'error' or 'errorifexists' (default), which confirms that without a mode, a write to an existing location fails. On an exam question, a missing mode() call usually signals this error behaviour.

ignore behaves differently from error. When data already exists, ignore does nothing and raises no exception, so the job succeeds even though nothing was written.

Sources64

3.partitionBy: one directory per column value

partitionBy(*cols) sets the physical layout of a file write. In the reference's words, "the output is laid out on the file system similar to Hive's partitioning scheme". Each distinct value of each partition column gets its own column=value subdirectory. The Parquet guide gives the reason: "Write partitioned Parquet files for optimized query performance on large datasets." A query that filters on the partition column can skip the directories it doesn't need.

The official example below writes two rows partitioned by name with mode("overwrite") and reads them back in two ways. Reading the root path returns both age and name, because Spark rebuilds name from the directory names. Reading the name=Alice directory directly returns only age. Inside that directory, the partition value is part of the path, not a column in the files.

with tempfile.TemporaryDirectory(prefix="partitionBy") as d:
    spark.createDataFrame(
        [{"age": 100, "name": "Alice"}, {"age": 120, "name": "Ruifeng Zheng"}]
    ).write.partitionBy("name").mode("overwrite").format("parquet").save(d)

    spark.read.parquet(d).sort("age").show()
    # +---+-------------+
    # |age|         name|
    # +---+-------------+
    # |100| Alice|
    # |120|Ruifeng Zheng|
    # +---+-------------+

    # Read one partition as a DataFrame.
    spark.read.parquet(f"{d}{os.path.sep}name=Alice").show()
    # +---+
    # |age|
    # +---+
    # |100|
    # +---+

partitionBy accepts several columns, and the order you give sets the nesting order. The Parquet guide derives year and month from a check_in date with withColumn, then calls .partitionBy("year", "month"). Partition columns must already exist in the DataFrame, so you derive them before the write. You can also pass partition columns as an argument instead of chaining the method, for example parquet(path, mode, partitionBy, compression).

# Partition by multiple columns
df.write. ? ("year", "month").parquet("path/to/output.parquet")

Sources72

4.Reading and partitioning files from SQL

You can also read files with Spark SQL instead of the DataFrame API. The Parquet guide says to "Use read_files to query Parquet files directly from cloud storage using SQL without creating a table." read_files is a table-valued function. It takes a path and named options such as format => 'parquet', and it supports JSON, CSV, XML, TEXT, BINARYFILE, PARQUET, AVRO and ORC.

SELECT * FROM read_files(
  '/Volumes/<catalog>/<schema>/<volume>/reviews_parquet',
  format => 'parquet'
)

For data you will query again, SQL can create a table from a query, and the same statement can partition it. PARTITIONED BY in SQL does the same job as partitionBy on the writer. Note that the derived year and month columns are computed in the SELECT, just as the Python version adds them with withColumn first.

The exam guide also mentions the shorthand `SELECT * FROM parquet.path `. The documentation used for this lesson doesn't describe that syntax, so this lesson doesn't make claims about it. Check the Spark SQL data sources guide for the details.

-- Write partitioned Parquet files by year and month
CREATE TABLE bookings_parquet_partitioned
USING PARQUET
PARTITIONED BY (year, month)
AS SELECT *, year(check_in) AS year, month(check_in) AS month
FROM samples.wanderbricks.bookings;

Sources2

Exam traps

Each one states something that sounds right. Open it to see what is actually true.

1. 1.A plain df.write.parquet(path) to an existing location silently overwrites it.Why is that wrong?

   The default mode is error (errorifexists), so the write throws an exception. To replace the data you must ask for overwrite explicitly.

   Covered in Save modes: what happens when the target already exists

2. 2.mode("ignore") raises an error when data already exists, just like errorifexists.Why is that wrong?

   ignore does nothing when data exists and raises no exception, so the job succeeds even though it wrote nothing.

   Covered in Save modes: what happens when the target already exists

3. 3.save() without a format always writes Parquet.Why is that wrong?

   Parquet is the default only in Apache Spark. On Databricks, Delta Lake is the default format for reading and writing data and tables.

   Covered in One interface for every file format