Spark SQL - Basic DataFrame Operations: Theoretical Quiz

This assessment focuses on DataFrame schema configurations, group-by aggregation internals, and type safety constraints.

Scenario 1: Schema Inference vs. Explicit Schema Enforcement

The Scenario

A data engineering team schedules a batch ETL job to ingest a daily 20-Terabyte raw CSV clickstream dataset stored in HDFS. A junior developer configures the ingestion with automatic schema inference:

# Ingest with schema inference
df = spark.read.option("header", "true") \
               .option("inferSchema", "true") \
               .csv("hdfs://cluster/raw_clicks/*.csv")

The cluster manager monitors the job and observes massive disk reads and high cluster latency before the actual transformation stages even begin.

The Questions

1. Describe the exact read execution profile of .option("inferSchema", "true") on raw files, and explain why it wastes heavy cluster resource cycles.
2. Provide the refactored PySpark code using an explicit StructType schema.

Detailed Solution & Architectural Analysis

1. Schema Inference Execution Profile

• Double-Read Penalty: When inferSchema is set to true, Spark cannot immediately initialize the execution engine. It must launch an initial dedicated read job that parses the entire 20-Terabyte dataset to inspect every string, integer, float, and timestamp to determine the best-fit column type. Once it resolves the schema, it launches the second actual read job to do the business logic. This doubles the disk I/O cost, wasting massive CPU/network cycles.
• Schema Instability: If a single malformed row in file #500 contains a string character in a column that is 99% integers, Spark will infer that entire column as a String, causing downstream mathematical filters to fail silently or crash.

2. Explicit StructType Schema Refactoring

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

# Define strict, compile-safe schema
clickstream_schema = StructType([
    StructField("click_id", StringType(), False),
    StructField("user_id", StringType(), False),
    StructField("zip_code", IntegerType(), True),
    StructField("event_time", TimestampType(), False),
    StructField("response_time_ms", IntegerType(), True)
])

# Read directly with zero read-ahead schema pass
df = spark.read.option("header", "true") \
               .schema(clickstream_schema) \
               .csv("hdfs://cluster/raw_clicks/*.csv")

This forces Spark to assume the schema instantly, launching only a single parallel pass to process the records, protecting memory stability.

Scenario 2: GroupBy Aggregation Hash Map Mechanics

The Scenario

A PySpark DataFrame job executes df.groupBy("zip_code").agg({"response_time": "avg"}). The dataset has millions of zip codes. The job frequently spills intermediate records to disk during the aggregation stage.

The Questions

1. Explain how Spark SQL manages in-memory aggregation using local Hash Maps inside executor memory blocks.
2. What causes the aggregation memory overhead to spill data to disk?

Detailed Solution & Architectural Analysis

1. In-Memory Hash Aggregation Mechanics

When Spark performs a groupBy, it does not immediately shuffle every record. It utilizes Hash Maps locally inside the executor tasks (Tungsten's binary memory layout).

• Local Accumulation: As records flow into the task thread, they are aggregated locally in a high-speed in-memory hash map (Key -> Aggregation States). For example, 94101 -> (sum=120, count=4).
• Partial Aggregation: Once the local partitions are consumed, only these partial metrics are shuffled over the network to the reducers, minimizing network traffic.

2. Spill to Disk Causes

If the group-by key has high cardinality (e.g. millions of unique user UUIDs or ZIP codes), the local in-memory hash map will expand rapidly to hold distinct keys.

• If the hash map size exceeds the allocated execution memory boundary (spark.memory.fraction), Spark's memory manager blocks further RAM allocation.
• To prevent OOM errors, Spark freezes the active hash map, serializes it, and spills the intermediate groups to the executor's local disk. This results in high disk I/O penalties and slows down the aggregation significantly.

Scenario 3: Null Value Handling in Joins and Aggregations

The Scenario

A financial ledger join sales_df.join(customers_df, "customer_id") drops 15% of transactions silently because the customer_id contains null/missing fields.

The Questions

1. Compare how Null values are treated in Inner Joins vs. Outer Joins.
2. How can we use coalesce or fillna inside PySpark DataFrame pipelines to handle null keys explicitly?

Detailed Solution & Architectural Analysis

1. Join Null Semantics

• Inner Joins: Drop Null keys completely because Null = Null evaluates to Unknown in SQL 3-valued logic. Spark never matches two null values.
• Outer Joins: Retain rows containing Null keys on the active side, filling the missing joined columns with null.

2. Explicit Null Resolution

To prevent transaction data drops, use fillna or coalesce:

import pyspark.sql.functions as F

# Fill missing customer IDs with a default placeholder string
safe_sales_df = sales_df.fillna({"customer_id": "UNKNOWN_CUSTOMER"})
safe_customers_df = customers_df.fillna({"customer_id": "UNKNOWN_CUSTOMER"})

# Join safely on placeholder key
joined_df = safe_sales_df.join(safe_customers_df, "customer_id", "inner")

Scenario 4: Global Temp Views vs. Local Temp Views Metadata Scopes

The Scenario

A data platform architect schedules multiple SparkSessions within a single cluster application. They notice that queries in Session B fail when attempting to read a temporary view created in Session A.

The Questions

1. Differentiate local Temp Views (createOrReplaceTempView) and Global Temp Views (createOrReplaceGlobalTempView) in terms of metadata scope and life-cycle.
2. What database prefix must be used to query a Global Temp view in SQL?

Detailed Solution & Architectural Analysis

1. Local Temp Views vs. Global Temp Views

• Local Temp View: Bound strictly to the SparkSession that created it. Once the session terminates or if accessed from a parallel session, the view's catalog reference is unreachable.
• Global Temp View: Bound to the shared, cluster-wide SparkContext. It remains active across all sessions running on the cluster until the entire Spark application terminates.

2. Global View Database Prefix

Global temporary views are cataloged under a system-reserved database namespace: global_temp. To query them in SQL:

spark.sql("SELECT * FROM global_temp.global_transactions_view").show()