Spark SQL - Data Sources & Formats: Theoretical Quiz

This assessment deep-dives into structured file format optimization, Parquet data skipping, and parallel database ingestion tuning.

Scenario 1: Columnar Storage Predicate Pushdown (Parquet)

The Scenario

An analytics pipeline queries an 80-Terabyte daily transactional log archive stored in Parquet format. The queries target a tiny fraction of columns and apply strict filters:

df = spark.read.parquet("hdfs://cluster/archive/")
result_df = df.select("transaction_id", "country") \
              .filter("transaction_date = '2026-05-26'")

A network administrator notices that only a couple of gigabytes are transferred over the network switches during this heavy scan.

The Questions

1. Detail the physical layout differences of Parquet (Columnar) vs CSV (Row-based) and explain how this impacts disk I/O.
2. Explain how Predicate Pushdown and Column Projection function inside Parquet metadata blocks to skip loading irrelevant bytes.

Detailed Solution & Architectural Analysis

1. Parquet Columnar Layout vs. CSV

• CSV Layout (Row-based): Stores records sequentially row1_col1, row1_col2, row1_col3, .... To select only two columns (transaction_id and country), the scan engine must read every row's byte stream sequentially from disk, parsing newline characters and commas, generating heavy disk I/O overhead.
• Parquet Layout (Columnar): Groups records into horizontal segments called Row Groups, and columns are stored in independent blocks within each Row Group. If a query only needs transaction_id and country, Spark reads the file metadata, determines the byte-offsets of those specific column streams, and completely skips reading the bytes for the other columns.

2. Data Skipping & Metadata Mechanics

• Column Projection: Allows Spark to select only column offsets transaction_id and country, reducing the read data volume by >90% for tables with hundreds of columns.
• Predicate Pushdown: Parquet files contain metadata headers at the file and Row Group levels containing statistical indicators: Min/Max values for each column block.
  • When the query applies transaction_date = '2026-05-26', Spark checks the Min/Max bounds of transaction_date inside each Row Group header.
  • If a Row Group's date range is ['2026-01-01', '2026-05-20'], Spark skips reading that entire Row Group from disk. This limits physical file reads to only the matching blocks.

Scenario 2: Parallelizing JDBC Database Connections

The Scenario

A developer writes a PySpark DataFrame job to pull a database table containing 40 million customer accounts from a Microsoft SQL Server database:

# Default read
df = spark.read.format("jdbc") \
               .option("url", "jdbc:sqlserver://host") \
               .option("dbtable", "customers") \
               .load()

The job runs for hours, while database CPU remains on a single thread and executors sit idle.

The Questions

1. Why does the default JDBC configuration limit ingestion to a single thread/partition?
2. Explain how to use partitionColumn, lowerBound, upperBound, and numPartitions options to divide database scans into parallel partition queries safely.

Detailed Solution & Architectural Analysis

1. The Single JDBC Connection Bottleneck

By default, when spark.read.jdbc is called without partition options, Spark initializes exactly 1 execution task running a single database socket query (SELECT * FROM customers) on one executor JVM. The other executors do not receive tasks, and the network bandwidth is bottlenecked by the processing capacity of that single database session thread.

2. Parallel JDBC Tuning & Partition Math

To divide the ingestion into parallel execution tasks, you must pass boundary configurations:

df = spark.read.format("jdbc") \
    .option("url", "jdbc:sqlserver://host") \
    .option("dbtable", "customers") \
    .option("partitionColumn", "customer_id") \
    .option("lowerBound", "1") \
    .option("upperBound", "40000000") \
    .option("numPartitions", "40") \
    .load()

• Partition Splits: Spark divides the primary integer range (1 to 40,000,000) by numPartitions (40). This establishes uniform interval bounds of size 1,000,000.
• Parallel Tasks: Spark generates 40 parallel tasks across executors, each issuing a localized range query to the database in parallel:
  • Task 1: SELECT ... WHERE customer_id >= 1 AND customer_id < 1000000
  • Task 2: SELECT ... WHERE customer_id >= 1000000 AND customer_id < 2000000
  • Task 40: SELECT ... WHERE customer_id >= 39000000 AND customer_id <= 40000000 This distributes the data loading work evenly across the cluster and accelerates ingestion 40x.

Scenario 3: Ingesting Malformed JSON/CSV Rows safely

The Scenario

A daily JSON feed from a partner firm contains malformed rows, missing closing brackets, and string elements mixed into integer fields. The standard Spark read operation fails instantly.

The Questions

1. Compare the execution profiles and error handling of PERMISSIVE, DROPMALFORMED, and FAILFAST read modes.
2. How can we use the columnNameOfCorruptRecord option to isolate bad rows into a dedicated column for auditing?

Detailed Solution & Architectural Analysis

1. Data Ingestion Fail-safe Modes

• PERMISSIVE (Default): When a malformed record is parsed, Spark does not crash. It replaces the corrupted fields with null and logs the raw corrupted string inside a user-defined column.
• DROPMALFORMED: Ignores and drops all corrupted records silently, yielding only the successfully parsed rows.
• FAILFAST: Crashes the entire Spark application instantly upon encountering the first malformed row, preventing corrupted data from entering the warehouse.

2. Corrupted Record Column Configuration

# Configure permissive parsing with a dedicated audit column
df = spark.read.option("mode", "PERMISSIVE") \
               .option("columnNameOfCorruptRecord", "_corrupt_record") \
               .json("hdfs://cluster/raw_data/*.json")

All schema violations and malformed lines will be written directly into _corrupt_record as raw strings, allowing developers to filter and audit them downstream without crashing the main ETL.

Scenario 4: Hive Directory Partition Discovery in Data Lakes

The Scenario

A data lake organizes transaction files using directory structures: .../year=2026/month=05/day=26/transactions.parquet A developer wants to know how Spark discovers these partitions automatically.

The Questions

1. Explain how Directory Partition Discovery operates when Spark scans the root path of the data lake.
2. What are the metadata performance risks of having millions of nested partition folders?

Detailed Solution & Architectural Analysis

1. Directory Partition Discovery Mechanics

When you write spark.read.parquet("hdfs://cluster/raw_transactions/"):

1. Spark scans the root directory and identifies directory names containing equal signs = (e.g. year=2026).
2. It parses these directory names as columns and infers their data types.
3. The scanned records are populated with virtual columns year, month, and day matching their physical storage folders.

2. Metadata Overhead Hazards (File Spans)

If the data lake is over-partitioned (e.g., partitioning by both customer ID and day, generating millions of directories containing only tiny 10KB files):

• To plan a query, Spark must issue recursive listing commands (RPCs) to HDFS/S3 to locate all physical files.
• This listing overhead can take minutes, stalling query compilation before execution even begins. This is known as the "Small Files metadata bottleneck".