Reading and Writing DataFrames with Schemas and Save Modes

1.Two entry points: spark.read and df.write

Every input and output question on the exam starts from one of two objects, and it matters which object owns which call. You read through SparkSession.read, which returns a DataFrameReader. You write through DataFrame.write, which returns a DataFrameWriter. A common distractor reverses them, for example offering df.read or spark.write. Neither exists, because reading creates a DataFrame from the session and writing takes an existing DataFrame out to storage.

Both builders give you two equivalent ways to state the format. You can use a generic chain such as spark.read.format("json").load(path) or df.write.format("parquet").save(path). Or you can call a format shortcut such as spark.read.json(path) or df.write.parquet(path). The shortcut does the same thing as setting format() and then calling load() or save(). Options work the same way in both styles: option(key, value) adds one setting, and options(**options) adds several at once.

Each format shortcut has a fixed output shape worth knowing. text() is the odd one out. It does not split lines into fields. Instead it returns a DataFrame with one string column per line, and that column has a fixed name.

Sources12

2.Reading DataFrames with an explicit schema

Without a schema, Spark has to work one out. Some sources, JSON among them, infer it from the data, and for CSV you can set option("inferSchema", "true"). Inference costs time, because Spark has to look at the data before the real read. DataFrameReader.schema() lets you declare the schema up front, so the source can skip that step. The method takes one of two forms: a StructType built from StructField(name, type, nullable) entries, or a DDL-formatted string such as "name STRING, age INT". Both forms describe the same thing, and the exam may show either.

from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# Define schema
schema = StructType([
    StructField("name", StringType(), True),
    StructField("age", IntegerType(), True)
])

# Read CSV with schema
df = spark.read.schema(schema).csv("path/to/file.csv")

# Read CSV with DDL-formatted string schema
df = spark.read.schema("name STRING, age INT").csv("path/to/file.csv")

You can check what you got with printSchema(). A DDL string of "col0 INT, col1 DOUBLE" prints as col0: integer and col1: double, and both are marked nullable. The same schema() call works for Parquet too. The Databricks docs recommend passing one there to avoid the overhead of schema inference.

Matching by position is the CSV-specific catch. A mismatched schema does not always fail. It can quietly shift values into the wrong fields, and the docs warn that results can differ considerably depending on which columns you access. The docs also show a partial schema, listing only review_id and rating, used to read just a subset of columns.

So what happens when a value cannot be parsed into the declared type, such as a city name in an IntegerType field? That depends on the parser mode, which you set with .option("mode", ...). This read option has nothing to do with the writer's save mode, which comes up later.

PERMISSIVE replaces bad values with nulls without telling you, so you may want to see which rows were affected. One way is to add a _corrupt_record string column to the schema you pass in, then filter for rows where it is not null. (The badRecordsPath option takes precedence over it: rows written to that path do not appear in the DataFrame.)

schema = StructType([
  StructField("review_id", StringType(), True),
  StructField("rating", IntegerType(), True),
  StructField("comment", StringType(), True),
  StructField("_corrupt_record", StringType(), True)
])

df = (spark.read
  .format("csv")
  .option("header", "true")
  .option("mode", "PERMISSIVE")
  .schema(schema)
  .load("/Volumes/<catalog>/<schema>/<volume>/reviews_csv")
)
display(df.filter(df["_corrupt_record"].isNotNull()))

Sources345

3.Writing DataFrames to files

Writing works like reading, but starts from the DataFrame. df.write returns a DataFrameWriter. You set the format with a shortcut (json(), csv(), parquet(), orc(), text()) or with format(...) followed by save(path). Options go in with option(), for example header for CSV or compression for Parquet. The writer has no schema() method, because the schema it writes is the DataFrame's own. If you want a different output schema, you change the DataFrame before writing it.

The writer also controls the directory layout. partitionBy(*cols) creates a separate directory level on the file system for each value of the named columns. The Databricks Parquet example derives year and month columns from a date column with withColumn and then partitions on them. The writer settings can be chained before the final save(), as below.

# Chain multiple configuration methods
df.write \
    .format("parquet") \
    .mode("overwrite") \
    .option("compression", "snappy") \
    .partitionBy("year", "month") \
    .save("path/to/output")

Sources24

4.Overwriting and appending: save modes

mode(saveMode) sets what the writer does when data or a table already exists at the target. Mode only matters when there is existing data. Writing to a new path works the same way under all four. The exam tests whether you know the default. Many candidates guess overwrite or append, but the writer refuses to touch existing data unless you tell it to.

Be careful with ignore. The job reports success, but nothing is written, so stale data can stay in place without anyone noticing. The reference example below writes to one directory twice. The first write uses overwrite, the second uses append, and then the directory is read back. The resulting DataFrame contains both Alice's and Sue's rows.

import tempfile
with tempfile.TemporaryDirectory(prefix="mode") as d:
    # Overwrite the path with a new Parquet file
    spark.createDataFrame(
        [{"age": 100, "name": "Alice"}]
    ).write.mode("overwrite").format("parquet").save(d)

    # Append another DataFrame into the Parquet file
    spark.createDataFrame(
        [{"age": 120, "name": "Sue"}]
    ).write.mode("append").format("parquet").save(d)

    # Read the Parquet file as a DataFrame.
    spark.read.parquet(d).show()

df.write.mode(" ? ").parquet("path/to/output.parquet")

Sources62

5.Writing and overwriting tables: saveAsTable vs insertInto

The writer can also target a table instead of a path. Two methods do this, and they behave differently in ways the exam likes to test.

saveAsTable(name, format, mode, partitionBy, **options) saves the DataFrame as the named table. If the table already exists, the save mode decides what happens, and as with files the default is to throw an exception. Two details matter. First, in overwrite mode the DataFrame's schema does not have to match the existing table's schema, so overwrite can replace the table's structure as well as its rows. Second, in append mode the existing table's format and options are used, and columns are matched by name.

insertInto(tableName, overwrite) inserts into a table that must already exist. It takes an overwrite boolean instead of relying on a mode string. Unlike saveAsTable, it does not use column names to find column positions.

spark.sql("DROP TABLE IF EXISTS tblA")
spark.createDataFrame([
    (100, "Alice"), (120, "Bob"), (140, "Tom")],
    schema=["age", "name"]
).write.saveAsTable("tblA")

# Insert into existing table
df.write.insertInto("existing_table")

# Insert into existing table with overwrite
df.write.insertInto("existing_table", overwrite=True)

Sources72

Exam traps

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

1. 1.If you don't call mode(), df.write overwrites (or appends to) data that is already at the path.Why is that wrong?

   The default save mode is error/errorifexists. Writing to an existing path without mode() throws an exception, so you have to ask for overwrite or append explicitly.

   Covered in Overwriting and appending: save modes

2. 2.With header=true, an explicit CSV schema is matched to the header column names.Why is that wrong?

   CSV fields are matched to schema fields by position. A schema in a different order from the file shifts values into the wrong columns.

   Covered in Reading DataFrames with an explicit schema

3. 3.Reading a CSV with an explicit schema fails by default when a value doesn't fit the declared type.Why is that wrong?

   The default parser mode is PERMISSIVE. It keeps the row and puts null in the field that failed. Only FAILFAST aborts the read.

   Covered in Reading DataFrames with an explicit schema

4. 4.insertInto and saveAsTable both line up DataFrame columns with table columns by name.Why is that wrong?

   Only saveAsTable uses column names to find column positions. The documentation contrasts this explicitly with insertInto.

   Covered in Writing and overwriting tables: saveAsTable vs insertInto

Practise it for real

Watch the overwrite and append save modes act on the same Parquet path, then read the result back.

1. 1.Inside a tempfile.TemporaryDirectory, write a one-row DataFrame ({"age": 100, "name": "Alice"}) with .write.mode("overwrite").format("parquet").save(d).

   Why: Overwrite replaces whatever is at the path, so this write succeeds even if the directory exists.

   You should see: The write finishes without an error.

2. 2.Write a second one-row DataFrame ({"age": 120, "name": "Sue"}) to the same path with .write.mode("append").format("parquet").save(d).

   Why: Append adds this DataFrame's contents to the existing data instead of replacing it.

   You should see: The write finishes, and the directory now holds both writes.

3. 3.Run spark.read.parquet(d).show().

   Why: Reading back shows the combined effect of the two modes.

   You should see: Two rows: Alice (100) and Sue (120).

4. 4.Write to the same path again with .write.mode("error").format("parquet").save(d).

   Why: error/errorifexists is the default mode, and it refuses to write when data already exists.

   You should see: An exception saying the data already exists.