Reading and Writing Data Sources

Analogy categoryAnalogies

The DataFrameReader and DataFrameWriter APIs

Every read starts from spark.read, a DataFrameReader you configure with .format("parquet"), a chain of .option() calls for format-specific settings, and finally .load(path) or a format-specific shortcut like spark.read.parquet(path). Writing mirrors this exactly: df.write.format("parquet").mode("overwrite").save(path), where .mode() controls what happens if the destination already has data — append, overwrite, error (the default, also called errorifexists), or ignore, which silently does nothing if the path exists.

🏏

Cricket analogy: It is like a groundsman preparing a pitch: choose the pitch type, set specific conditions (grass length, moisture), then either lay a fresh pitch (overwrite), extend the existing one (append), or refuse to touch it if one's already there (error).

Choosing a Format: Parquet, CSV, and JSON

Parquet is Spark's default and preferred format for internal pipeline stages: it's columnar, stores schema and statistics in the file's footer, supports predicate and column pruning so queries skip reading unneeded data, and compresses well because similar values sit adjacent within a column. CSV and JSON are row-oriented, human-readable, and common at data ingestion boundaries, but they lack embedded schema (CSV entirely, JSON only loosely via inferred types), so reading them typically requires either .option("inferSchema", "true") — which triggers an extra pass over the data — or an explicitly supplied StructType schema, which is both faster and safer for production pipelines.

🏏

Cricket analogy: It is like a stat-analysis system storing every player's strike rates in one contiguous column for instant filtering, versus a handwritten scorebook where you must flip page by page (row by row) to find the same figures.

python

from pyspark.sql.types import StructType, StructField, StringType, DoubleType, TimestampType

# Explicit schema avoids a costly inferSchema pass over the data
csv_schema = StructType([
    StructField("order_id", StringType(), nullable=False),
    StructField("customer_id", StringType(), nullable=False),
    StructField("amount", DoubleType(), nullable=True),
    StructField("order_ts", TimestampType(), nullable=True),
])

orders_df = (
    spark.read
         .format("csv")
         .option("header", "true")
         .schema(csv_schema)
         .load("s3://data/raw/orders/")
)

# Write as partitioned Parquet for efficient downstream querying
(
    orders_df.write
        .format("parquet")
        .mode("overwrite")
        .partitionBy("order_date")
        .option("compression", "snappy")
        .save("s3://data/curated/orders/")
)

# JDBC read from a relational database
jdbc_df = (
    spark.read
        .format("jdbc")
        .option("url", "jdbc:postgresql://db-host:5432/salesdb")
        .option("dbtable", "public.orders")
        .option("user", "reader")
        .option("password", dbutils.secrets.get("scope", "db-password"))
        .load()
)

Partitioned Writes and JDBC Sources

partitionBy("order_date") on a write splits the output into a directory-per-value layout (order_date=2026-07-01/, order_date=2026-07-02/, ...), which lets downstream readers skip entire directories via partition pruning when a query filters on that column, at the cost of potentially creating many small files if the partition column has high cardinality relative to data volume. Reading from a JDBC source pulls data through a single connection by default unless you supply partitionColumn, lowerBound, upperBound, and numPartitions options, which let Spark split the read into multiple parallel queries against numeric ranges of that column.

🏏

Cricket analogy: It is like organizing a stadium's archive by match date into separate labeled boxes — finding all of last Tuesday's scorecards means grabbing one box, not searching every box in the warehouse; but a box per single delivery would create far too many tiny folders.

For JDBC reads of large tables, always set numPartitions along with partitionColumn, lowerBound, and upperBound — without them, Spark reads the entire table through one JDBC connection sequentially, which is often the single biggest bottleneck in an otherwise well-tuned pipeline.

Partitioning a write by a high-cardinality column, such as a raw timestamp or a UUID, can produce an enormous number of tiny partition directories and files — this small-file problem slows down both the write itself and every future read, since the driver must list and plan around thousands of files. Partition by coarse-grained columns like date or region instead.

spark.read/.write use a consistent .format().option().load()/.save() pattern across all data sources.
write mode controls collision behavior: append, overwrite, error (default), or ignore.
Parquet is columnar, self-describing, and supports predicate/column pruning, making it Spark's preferred internal format.
CSV and JSON lack robust embedded schemas; supplying an explicit StructType avoids a costly inferSchema pass.
partitionBy() on write enables partition pruning on read but risks the small-file problem with high-cardinality columns.
JDBC reads run through a single connection unless partitionColumn/lowerBound/upperBound/numPartitions are set for parallelism.
Choose partition columns with coarse, well-distributed cardinality such as date or region.