Reading and Writing Data in Spark: Parquet, Delta, JSON, and JDBC

Scenario context

Parquet's columnar layout with row-group statistics enables predicate pushdown that can reduce a 500 GB scan to 8 GB. Delta Lake wraps Parquet with a JSON transaction log to add ACID semantics and time travel. JSON and CSV read every byte because they have no statistics. JDBC reads scale only with explicit numPartitions + partitionColumn. The format you choose is not cosmetic — it determines the entire I/O path…

This walkthrough focuses on The Team That Turned a 500 GB Daily Scan into an 8 GB Query.

Before making a design choice, clarify:

• A platform analytics team at a mid-size SaaS company had a daily pipeline that read the previous seven days of raw user-event data from S3 and computed a set of funnel metrics.
• The data was stored as gzipped JSON — one file per hour per region — totalling approximately 500 GB.
• The pipeline ran for ninety minutes every morning, used forty executors, and cost the team about $180 per run on their cloud cluster.

Concepts in play: spark-data-engineering, caching, data-warehousing-lakehouse, transaction-isolation-distributed-acid

Stage 1 — The DataFrameReader and DataFrameWriter API and What You Can Control

• Every Spark I/O operation passes through two symmetric APIs: DataFrameReader for reads and DataFrameWriter for writes.
• Understanding the options available at this surface is the prerequisite for understanding what the underlying format can and cannot do.
• The DataFrameReader Pattern The reader is accessed via spark.read and accepts a format identifier, optional key-value options, and a path.
• The supported built-in formats are: The reader chain works as: specify format → apply options → load a path or paths.

output/
  date=2026-01-15/
    region=us-east/
      part-00000-abc123.parquet
      part-00001-def456.parquet
    region=eu-west/
      part-00002-ghi789.parquet
  date=2026-01-16/
    region=us-east/
      part-00003-jkl012.parquet

Observable outcome: The supported built-in formats are: The reader chain works as: specify format → apply options → load a path or paths.

Stage 2 — Parquet Internals, the Delta Transaction Log, and How Partition Pruning Works

• Parquet's Columnar Layout: Row Groups, Column Chunks, and Statistics A Parquet file is divided horizontally into row groups — each row group is a horizontal slice of the data, typically 128 MB.
• Within a row group, data for each column is stored contiguously as a column chunk .
• Column chunks are further divided into pages of approximately 1 MB each, which are the unit of compression and encoding.
• Every column chunk stores column statistics in the Parquet footer: the minimum and maximum values of that column across all rows in the row group.

_delta_log/
  00000000000000000000.json   ← initial commit (table creation)
  00000000000000000001.json   ← first write
  00000000000000000002.json   ← second write (e.g., DELETE)
  00000000000000000010.json   ← tenth write
  00000000000000000010.checkpoint.parquet  ← checkpoint at commit 10
  _last_checkpoint              ← pointer to latest checkpoint

{"commitInfo":{"timestamp":1713398400000,"operation":"WRITE","operationParameters":{"mode":"Append","partitionBy":"[\"date\"]"}}}
{"add":{"path":"date=2026-01-15/part-00000-abc123-c000.snappy.parquet","partitionValues":{"date":"2026-01-15"},"size":134217728,"modificationTime":1713398400000,"dataChange":true}}
{"add":{"path":"date=2026-01-15/part-00001-def456-c000.snappy.parquet","partitionValues":{"date":"2026-01-15"},"size":127926944,"modificationTime":1713398400000,"dataChange":true}}

Observable outcome: Every column chunk stores column statistics in the Parquet footer: the minimum and maximum values of that column across all rows in the row group.

Stage 3 — Deep Dive: How Spark Plans and Executes a Parquet Read Internally

• The Internals of Parquet Reads in Spark via DataSourceV2 Since Spark 2.3, the Parquet data source (and most others) is implemented via the DataSourceV2 API .
• This API introduces a clean separation between scan planning (what to read) and scan execution (how to read it), enabling more sophisticated optimizations than the original DataSource V1 API allowed.
• When spark.read.parquet("s3://bucket/events") is called, the execution sequence is: Step 1 — File scan planning.
• Spark calls the data source's Table.newScanBuilder() method.

graph TD
    A[SparkSession.read.parquet - path]
    B[DataSourceV2 scan builder]
    C[FileIndex - list partition directories]
    D[PartitionPruning - evaluate partition filters]
    E[Footer metadata read - row group statistics per file]
    F[Row group filtering - skip groups where predicate cannot match]
    G[Column projection - fetch only requested column chunks]
    H[Vectorized decode - 4096 rows per columnar batch]
    I[DataFrame - InternalRow columnar batches]

    A --> B
    B --> C
    C --> D
    D --> E
    E --> F
    F --> G
    G --> H
    H --> I

Observable outcome: Spark calls the data source's Table.newScanBuilder() method.

Stage 4 — Tracing the Parquet Read Path: From Storage to DataFrame

• The following diagram shows both the planning path and the execution path for a partitioned Parquet table read with predicate pushdown.
• The left column shows planning-time decisions (no data read); the right column shows execution-time I/O.
• The point where control transfers from planning to execution is the FileScan plan node — once that node exists, Spark knows exactly which files and row groups to read, and no planning-time decision can be revisited with…
• The planning phase (top half of the diagram) operates entirely on metadata: directory listings, footer statistics, and schema information.

graph TD
    QP[Query with WHERE predicates]
    CA[Catalyst Analyzer]
    CO[Catalyst Optimizer]
    PP[Partition pruning at FileIndex]
    FP[Filter pushdown to DataSourceV2]
    CP[Column pruning - ReadSchema]
    FS[FileScan plan node - pruned file list]
    DA[DAGScheduler creates FileScanRDD tasks]
    T1[Task 1 - reads file1 footer]
    T2[Task 2 - reads file2 footer]
    TN[Task N - reads fileN footer]
    RG[Row group filter - skip non-matching row groups]
    CD[Column data fetch - only projected columns]
    VD[Vectorized decode - columnar batches]
    DF[DataFrame result]

    QP --> CA
    CA --> CO
    CO --> PP
    PP --> FP
    FP --> CP
    CP --> FS
    FS --> DA
    DA --> T1
    DA --> T2
    DA --> TN
    T1 --> RG
    T2 --> RG
    TN --> RG
    RG --> CD
    CD --> VD
    VD --> DF

Observable outcome: The planning phase (top half of the diagram) operates entirely on metadata: directory listings, footer statistics, and schema information.

Stage 5 — Delta Lake Merges, JDBC Parallel Reads, and Dynamic Partition Overwrites

• Delta Lake MERGE: The Upsert Pattern for Mutable Data Lakes The most common mutation pattern in data lake workloads is the upsert (insert or update): for each incoming record, update the existing row if a matching key e…
• Relational databases handle this natively with SQL MERGE or INSERT ...
• For Parquet-based lakes, implementing upsert without Delta requires reading all existing data, merging the changes in memory, and rewriting entire partitions — an O(table size) operation regardless of change volume.
• Delta Lake implements this as a first-class MERGE INTO operation backed by the transaction log: Internally, Delta executes a MERGE as two phases: first, it identifies which data files contain rows matching the ON condit…

MERGE INTO target_table AS target
USING source_data AS source
ON target.user_id = source.user_id
WHEN MATCHED THEN UPDATE SET
    target.last_seen = source.event_ts,
    target.session_count = target.session_count + 1
WHEN NOT MATCHED THEN INSERT *

Query 1: WHERE order_id >= 1         AND order_id < 1000001
Query 2: WHERE order_id >= 1000001   AND order_id < 2000001
...
Query N: WHERE order_id >= (N-1)*1M  AND order_id <= 4000000

Observable outcome: Delta Lake implements this as a first-class MERGE INTO operation backed by the transaction log: Internally, Delta executes a MERGE as two phases: first, it identifies which data files contain rows matching the ON condit…

Stage 6 — Reading the Evidence: Explain Output and a Delta Transaction Log Walk-Through

• This section demonstrates two diagnostic skills: reading a Spark explain() output to understand what predicate pushdown is doing, and reading a Delta transaction log to understand what a write actually committed.
• Example 1 — Interpreting a Parquet Scan's explain() Output Consider a query selecting purchase events for a single date from a partitioned Parquet table with 47 schema columns.
• The explain(true) physical plan output contains the following FileScan node: Reading this output section by section tells the full I/O story: Location shows a single path after partition pruning.
• The original table has 365 date partitions but only date=2026-01-15 survived the PartitionFilters evaluation — Spark's file index listed 365 directories and included exactly one.

== Physical Plan ==
*(1) ColumnarToRow
+- FileScan parquet events[event_type#5, user_id#6, revenue#7]
   Batched: true,
   DataFilters: [isnotnull(event_type#5), (event_type#5 = purchase)],
   Format: Parquet,
   Location: InMemoryFileIndex(1 paths)[s3://data-lake/events/date=2026-01-15],
   PartitionFilters: [isnotnull(date#0), (date#0 = 2026-01-15)],
   PushedFilters: [IsNotNull(event_type), EqualTo(event_type,purchase)],
   ReadSchema: struct<event_type:string,user_id:string,revenue:double>

{"commitInfo":{"timestamp":1713484800000,"operation":"MERGE","operationParameters":{"predicate":"(target.user_id = source.user_id)","matchedPredicates":"[{\"actionType\":\"update\"}]","notMatchedPredicates":"[{\"actionType\":\"insert\"}]"},"readVersion":41,"isolationLevel":"Serializable","isBlindAppend":false}}
{"remove":{"path":"date=2026-01-14/part-00003-old-abc.snappy.parquet","deletionTimestamp":1713484800000,"dataChange":true}}
{"remove":{"path":"date=2026-01-15/part-00007-old-def.snappy.parquet","deletionTimestamp":1713484800000,"dataChange":true}}
{"add":{"path":"date=2026-01-14/part-00003-new-xyz.snappy.parquet","partitionValues":{"date":"2026-01-14"},"size":131072000,"modificationTime":1713484800000,"dataChange":true}}
{"add":{"path":"date=2026-01-15/part-00007-new-uvw.snappy.parquet","partitionValues":{"date":"2026-01-15"},"size":127926944,"modificationTime":1713484800000,"dataChange":true}}

Observable outcome: The original table has 365 date partitions but only date=2026-01-15 survived the PartitionFilters evaluation — Spark's file index listed 365 directories and included exactly one.

End-to-end outcome

A platform analytics team at a mid-size SaaS company had a daily pipeline that read the previous seven days of raw user-event data from S3 and computed a set of funnel metrics.

Alternatives considered

Choosing the Right Format: A Decision Guide for Spark I/O

Use this table when choosing a data format for a new Spark pipeline or migrating an existing one.

Failure and degraded modes

The Small File Problem in Streaming Writes

• Detection: Structured Streaming writes to Parquet produce one or more files per micro-batch per partition.
• User or operator impact: Structured Streaming writes to Parquet produce one or more files per micro-batch per partition.

Parquet vs Delta: When the Transaction Log Cost Matters

• Detection: Delta Lake is strictly more capable than plain Parquet — it adds ACID, time travel, schema enforcement, and MERGE support.
• User or operator impact: Delta Lake is strictly more capable than plain Parquet — it adds ACID, time travel, schema enforcement, and MERGE support.

JSON and CSV for Production Analytical Workloads

• Detection: JSON and CSV exist in production Spark pipelines primarily as landing zone formats — the format in which raw data arrives from external sources.
• User or operator impact: JSON and CSV exist in production Spark pipelines primarily as landing zone formats — the format in which raw data arrives from external sources.

Lessons Learned from Production Spark I/O Failures

• Detection: Lesson 1: Every JSON query is a full table scan, always.
• User or operator impact: Lesson 1: Every JSON query is a full table scan, always.