Apache Spark DataFrames, Catalyst Optimizer, PySpark Operations, and Spark SQL

This lecture moves from low-level Spark RDDs to the schema-full DataFrame API: why DataFrames exist and how the Catalyst Optimizer rewrites plans (10.1), how to ingest data and shape columns and rows (10.2), how to filter, cast, limit, and sort (10.3), how to group and aggregate (10.4), how relational joins behave and how shuffle versus broadcast joins run on a cluster (10.5), and how Spark SQL sits on the same optimizer with plan parity to the DataFrame API (10.6).

10.1 History, Architecture, and Catalyst Optimizer of Spark DataFrames

10.1.1 Evolution from RDDs to Schema-Full DataFrames

In early versions of Apache Spark (Spark 1.x), the core abstraction for distributed computation was the Resilient Distributed Dataset (RDD). An RDD (an immutable, fault-tolerant collection of elements partitioned across worker nodes) gave you powerful low-level functional primitives such as map, flatMap, and filter. Those primitives were enough to express almost any computation, but they created two operational problems in day-to-day analytics work:

1. High learning curve. RDD programs force you to write explicit functional lambda chains. Because RDDs are schema-less, records are opaque Python or Java objects — often tuples — so you reach fields by positional index (_1, _2) instead of by name.
2. No engine-level optimization. An RDD transformation treats each element as an arbitrary object. The Spark engine cannot look inside those objects to push filters earlier, drop unused columns, or choose a cheaper join plan before execution.

To lower the barrier for people who already know Python pandas and relational SQL tables, Spark introduced DataFrames.

When to pick which: Prefer DataFrames (or Spark SQL) for structured analytics so Catalyst can optimize. Drop to RDDs when you need element-level control that does not map cleanly to columns.

Real-world domain connection: Teams that migrate pandas prototypes to production Spark keep the mental model of “tables with columns,” while Catalyst turns those expressions into cluster-efficient jobs — the same reason enterprise stacks expose one DataFrame/SQL surface to Python, Scala, and Java users (see 10.8.1).

10.1.2 Catalyst Optimizer and Cost-Based Optimization (CBO)

When a program modifies a DataFrame, Spark does not run each line immediately. DataFrame transformations are lazy: Spark builds an abstract logical execution plan — a tree of relational operators describing what you asked for, not yet how to run it.

Spark then hands that plan to the Catalyst Optimizer. Catalyst walks the tree and applies rule-based rewrites (and, with statistics, cost-based choices) to produce an optimized physical execution plan.

Starting in Spark 2.2+, Spark added a Cost-Based Optimizer (CBO). The CBO uses table and column statistics (row counts, cardinality, size) to estimate costs of alternative plans and pick a lower-cost one when rules alone are not enough.

Because PySpark DataFrames and Scala/Java DataFrames both compile through the same Catalyst pipeline, query efficiency is unified across languages. For extreme object-allocation-heavy workloads, Scala/Java can still win on memory behavior because they stay on the JVM without a Python worker process — but the optimizer path itself is shared.

Visual intuition: picture the logical plan as a tree drawn top-down — Join at the root, Filter and Scan as leaves. After optimization, a Filter that used to sit above a Join may move under it (predicate pushdown), and a skinny Project may appear early so only two columns ride the shuffle instead of fifty. Takeaway: same answer, less network and CPU.

Real-world domain connection: ETL jobs that filter early and broadcast small dimension tables routinely cut shuffle volume by orders of magnitude compared with naive RDD-style join code — the same idea as 10.5.2 and 10.8.2.

10.1.3 Spark Session vs. Spark Context Entry Points

In Spark 1.x, the cluster entry point was SparkContext (often bound to the variable sc). It connected the driver to the cluster and created RDDs.

In Spark 2.0+, Spark unified entry points into SparkSession (often spark). SparkSession encapsulates SparkContext, SQLContext, and HiveContext, and is the standard entry point for DataFrame and Spark SQL work.

Inside interactive shells such as pyspark, spark and sc are usually pre-created. In standalone scripts, Jupyter, or Colab, you must build SparkSession yourself:

from pyspark.sql import SparkSession

# Initialize a SparkSession object
spark = SparkSession.builder \
    .appName("Spark DataFrame Foundation") \
    .getOrCreate()

10.2 Ingestion, Schema Definition, and Column/Row Operations

10.2.1 Data Ingestion Formats and Automatic Schema Inference

SparkSession exposes a unified reader (spark.read) for structured and semi-structured sources: CSV, JSON, Parquet, ORC, XML, JDBC databases (for example MySQL), many NoSQL connectors, and streaming sources.

You can use format-specific helpers or a generic format(...).load(...) path:

# Ingest JSON dataset using format descriptor
df_json = spark.read.format("json").load("2015-summary.json")

# Ingest CSV dataset using direct format reader
df_csv = spark.read.csv("online_retail.csv", header=True, inferSchema=True)

When you omit an explicit schema on semi-structured data (for example JSON), Spark infers types with an initial scan:

• Text fields → StringType
• Whole numbers without decimals → LongType or IntegerType

Course texts also stress Parquet/ORC/JSON as first-class Spark SQL sources: columnar formats (Parquet/ORC) store metadata that helps projection pushdown, while JSON is convenient but often costlier to parse at scale.

10.2.2 Explicit Schema Definition with StructType and StructField

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

# Define explicit custom schema
flight_schema = StructType([
    StructField("DEST_COUNTRY_NAME", StringType(), True),
    StructField("ORIGIN_COUNTRY_NAME", StringType(), True),
    StructField("count", IntegerType(), True)
])

# Load dataset enforcing the custom schema
df_flights = spark.read.format("json") \
    .schema(flight_schema) \
    .load("2015-summary.json")

Setting nullable=False on columns you know are always present lets Catalyst skip some null-check branches in generated code, which can improve physical execution when the guarantee is real.

10.2.3 DataFrame Column Access and Manipulation

PySpark’s main column constructors are col(), column(), and expr() from pyspark.sql.functions:

from pyspark.sql.functions import col, column, expr

# Column reference methods
c1 = col("DEST_COUNTRY_NAME")
c2 = column("ORIGIN_COUNTRY_NAME")
c3 = expr("count + 10")

col / column name a field; expr parses a SQL expression string into a Column. None of these alone materialize row values.

Column Projection with select() and selectExpr()

select() projects columns; selectExpr() accepts SQL expression strings:

# Projecting specific columns using select
df_flights.select(col("DEST_COUNTRY_NAME"), col("ORIGIN_COUNTRY_NAME")).show(5)

# Applying inline SQL expressions with selectExpr
df_flights.selectExpr(
    "DEST_COUNTRY_NAME as destination",
    "ORIGIN_COUNTRY_NAME as origin",
    "count",
    "count * 3 as triple_count",
    "ORIGIN_COUNTRY_NAME == DEST_COUNTRY_NAME as is_domestic"
).show(5)

lit() adds a constant value as a column:

from pyspark.sql.functions import lit

# Append a literal column 'one' containing integer 1
df_flights.select("*", lit(1).alias("one")).show(5)

Adding, Renaming, and Dropping Columns

# Add a new column 'new_count' equal to count multiplied by 3
df_with_col = df_flights.withColumn("new_count", expr("count * 3"))

# Rename 'new_count' to 'triple_count'
df_renamed = df_with_col.withColumnRenamed("new_count", "triple_count")

# Drop columns
df_dropped = df_renamed.drop("triple_count", "DEST_COUNTRY_NAME")

DataFrame Immutability

DataFrames are immutable. withColumn / drop do not mutate the old object’s buffers; they return a new DataFrame whose logical plan includes the change.

# The original df_flights remains unchanged after calling withColumn
df_flights.withColumn("test_col", lit(100))

# df_flights still contains only the original 3 columns
print(df_flights.columns) # ['DEST_COUNTRY_NAME', 'ORIGIN_COUNTRY_NAME', 'count']

# To persist transformations, reassign the result to a variable
df_flights = df_flights.withColumn("test_col", lit(100))

Immutability underpins fault tolerance: if a worker dies, Spark recomputes lost partitions from the immutable lineage back to source data.

10.2.4 Row Objects and In-Memory DataFrame Construction

from pyspark.sql import Row

# Construct individual Row instances
row1 = Row("Delhi", "Jaipur", 3)
row2 = Row("Jaipur", "Mumbai", 2)
row3 = Row("Mumbai", "Trivandrum", 1)

# Assemble rows into a list sequence
row_list = [row1, row2, row3]

# Create DataFrame without explicit schema (columns default to _1, _2, _3)
df_custom_raw = spark.createDataFrame(row_list)
df_custom_raw.show()

# Assign explicit column names using toDF()
df_custom = df_custom_raw.toDF("origin", "destination", "flight_count")
df_custom.show()

Real-world domain connection: Lakehouse jobs typically pair Parquet/ORC reads with checked-in schema definitions (or metastore tables) so evolving JSON dumps cannot silently change types overnight — the same production advice as avoiding inference on multi-TB scans.

10.3 Data Filtering, Type Casting, Limiting, and Sorting

10.3.1 Row Filtering with filter() and where()

filter() and where() are synonyms in PySpark. Prefer where if you think in SQL; prefer filter if you think in pandas-style functional APIs. Both keep rows for which the predicate is true.

# Filtering rows where count is greater than 6 using col()
df_filtered = df_flights.filter(col("count") > 6)

# Equivalent filtering using where() with SQL string expression
df_filtered_where = df_flights.where("count > 200")

# Combining multiple logical predicates
df_multi_filter = df_flights.where(
    (col("count") > 100) & (col("ORIGIN_COUNTRY_NAME") != col("DEST_COUNTRY_NAME"))
)

Combine Column predicates with & (and), | (or), and ~ (not). Parentheses matter because Python operator precedence is not SQL precedence.

10.3.2 Strict Type Casting with cast()

from pyspark.sql.types import IntegerType

# Explicitly cast string column 'count_str' to IntegerType
df_casted = df_flights.select(
    col("count").cast(IntegerType()).alias("count_int"),
    col("count").cast("string").alias("count_str")
)

10.3.3 Row Truncation with limit() vs. show()

# limit(3) returns a new DataFrame object with 3 rows
df_top3 = df_flights.limit(3)

# show(3) executes the plan and prints formatted output to console
df_flights.show(3)

10.3.4 Deduplication with distinct()

distinct() keeps unique full-row combinations (after any prior select).

# Get distinct origin countries
df_unique_origins = df_flights.select("ORIGIN_COUNTRY_NAME").distinct()

print("Total raw rows:", df_flights.count())          # e.g., 256
print("Unique origin count:", df_unique_origins.count()) # e.g., 125

distinct is a wide transformation (shuffle) in general — expensive on huge frames; prefer keyed aggregations when you only need uniqueness for metrics.

10.3.5 Sorting and Ordering with sort() and orderBy()

sort() and orderBy() behave the same. Control direction with asc / desc helpers or .asc() / .desc() on Columns.

from pyspark.sql.functions import asc, desc

# Sort by count in descending order
df_sorted_desc = df_flights.orderBy(desc("count"))

# Multi-column sorting: count descending, origin country ascending
df_multi_sort = df_flights.sort(
    col("count").desc(),
    col("ORIGIN_COUNTRY_NAME").asc()
)

# Find top destination with maximum flights
df_flights.orderBy(expr("count").desc()).show(1)

Real-world domain connection: Retail and flight-analytics notebooks typically filter bad rows, cast price/quantity columns, then sort for “top country” dashboards — the same building blocks before grouped KPIs in 10.4.

10.4 Data Grouping and Aggregation

10.4.1 Full-Table vs. Grouped Aggregations

# Compute total record count across dataset
total_records = df_flights.count()

from pyspark.sql.functions import sum, avg, max, min, count

# Group flights by origin country and calculate total flight count
df_grouped_count = df_flights.groupBy("ORIGIN_COUNTRY_NAME").count()

# Group online retail transactions by Country and compute multiple summary metrics
df_retail_summary = df_retail.groupBy("Country").agg(
    count("InvoiceNo").alias("total_invoices"),
    sum("Quantity").alias("total_quantity"),
    avg("UnitPrice").alias("avg_unit_price"),
    max("Quantity").alias("max_single_order")
)

When groupBy runs, Spark hashes/partitions by key, may aggregate partially on the map side, shuffles so identical keys co-locate, then finishes reductions. That shuffle is why unselective grouping on high-cardinality keys can dominate job time — the same wide-transform cost story as joins (10.5).

Visual intuition: draw countries on the x-axis and total_quantity on the y-axis as a bar chart; each bar is one group’s reducer output. Tall bars are heavy keys; many tiny bars mean high cardinality and more shuffle groups.

Real-world domain connection: Country-level invoice dashboards in retail ETL are textbook Spark SQL/DataFrame workloads — the same grouped aggregates appear again as SQL GROUP BY once temporary views are registered (10.6).

10.5 Relational Joins and Distributed Execution Mechanics

10.5.1 Relational Join Types and Behavior

Demo tables:

1. person — columns id, name, graduate_program, spark_status

• (0, "Gita", 0, [100, 200])
• (1, "Rita", 1, [500])
• (2, "Meeta", 1, [600])
• (3, "Ganesh", 3, [300, 400]) — program 3 is missing from the catalog

1. graduate_program — columns id, school, degree

• (0, "IIT Madras", "M.Tech")
• (1, "IISc Bangalore", "M.Tech")
• (2, "BITS Pilani", "M.Tech") — no student references program 2

# Generic join invocation syntax
df_joined = df_left.join(df_right, join_expression, join_type)

# Join predicate matching graduate_program ID to school ID
join_expr = person["graduate_program"] == graduate_program["id"]

1. Inner Join ("inner")

Keeps rows whose keys exist in both sides; drops orphans; output includes columns from both tables.

person.join(graduate_program, join_expr, "inner").show()

Result people: Gita (IIT Madras), Rita (IISc Bangalore), Meeta (IISc Bangalore). Ganesh and BITS Pilani are out; no null-padded unmatched rows.

2. Outer / Full Outer Join ("outer" / "full_outer")

Keeps all rows from both sides; fills missing opposite-side attributes with null.

person.join(graduate_program, join_expr, "outer").show()

Result: Gita, Rita, Meeta, Ganesh (null school fields), and BITS Pilani (null student fields).

3. Left Outer Join ("left" / "left_outer")

Keeps all left rows; right attributes null when unmatched; drops right-only rows.

person.join(graduate_program, join_expr, "left").show()

Result: Gita, Rita, Meeta, Ganesh (null school). BITS Pilani excluded.

4. Right Outer Join ("right" / "right_outer")

Keeps all right rows; left attributes null when unmatched; drops left-only rows.

person.join(graduate_program, join_expr, "right").show()

Result: Gita, Rita, Meeta, and BITS Pilani (null student fields). Ganesh excluded.

5. Left Semi Join ("left_semi")

Keeps left rows that have a match on the right. Output schema = left columns only.

person.join(graduate_program, join_expr, "left_semi").show()

Result: Gita, Rita, Meeta with columns [id, name, graduate_program, spark_status] only — like an existence filter, not a column merge.

6. Left Anti Join ("left_anti")

Keeps left rows that lack a right match. Output schema = left columns only.

person.join(graduate_program, join_expr, "left_anti").show()

Result: Only Ganesh (program key 3 missing).

7. Cross Join ("cross")

Cartesian product: every left row with every right row.

person.crossJoin(graduate_program).show()

For sizes \(M\) and \(N\), output size is \(M \times N\). Here \(4 \times 3 = 12\) rows.

When to pick which: Use semi/anti when you only need existence logic; use outer forms when unmatched entities must remain visible; avoid cross joins unless you truly need \(M \times N\) pairs.

10.5.2 Distributed Execution Strategies: Shuffle Join vs. Broadcast Join

# Inspect physical execution plan for join operation
person.join(graduate_program, join_expr, "inner").explain()

+-----------------------------------------------------------------------------------+
|                            Distributed Join Strategies                            |
|                                                                                   |
|  1. Shuffle Join (Big-Table to Big-Table)                                         |
|     Worker Node 1 [Keys 0,1] ------------Shuffle Network------------> Executor A   |
|     Worker Node 2 [Keys 0,1] ------------Shuffle Network------------> Executor B   |
|                                                                                   |
|  2. Broadcast Join (Big-Table to Small-Table)                                     |
|     Driver Node ---> Broadcasts Small Table Once ---> Worker Node 1 (Local Join)  |
|                                                  ---> Worker Node 2 (Local Join)  |
+-----------------------------------------------------------------------------------+

Real-world domain connection: Nightly warehouse ETL joins billion-row fact tables to small store/currency dimensions with broadcast joins to cut shuffle — expanded in 10.8.2.

10.6 Spark SQL Interoperability and In-Memory Execution

10.6.1 Spark SQL Architecture and Workflow

Course materials describe Spark SQL as the structured-data component of the Spark stack: DataFrames (historically SchemaRDDs), SQL/Hive contexts (now under SparkSession), JDBC/ODBC connectivity, and support for sources such as Hive, JSON, and Parquet — with Catalyst optimization and Tungsten-style execution underneath.

# 1. Ingest dataset
df_retail = spark.read.csv("online_retail.csv", header=True, inferSchema=True)

# 2. Register temporary view
df_retail.createOrReplaceTempView("invoice_table")

# 3. Execute SQL query
df_sql_result = spark.sql("""
    SELECT 
        Country, 
        AVG(Quantity) as avg_qty, 
        SUM(Quantity) as total_qty 
    FROM invoice_table 
    GROUP BY Country 
    HAVING SUM(Quantity) > 1000
    ORDER BY total_qty DESC
""")

# 4. Display results (spark.sql returns a DataFrame object)
df_sql_result.show(5)

10.6.2 Relational Joins via Spark SQL

Once views exist, multi-table joins are ordinary SQL:

# Register temporary tables for person and graduate_program DataFrames
person.createOrReplaceTempView("person_table")
graduate_program.createOrReplaceTempView("graduate_table")

# Execute standard relational join in SQL
df_joined_sql = spark.sql("""
    SELECT 
        p.id as student_id, 
        p.name, 
        g.school, 
        g.degree 
    FROM person_table p
    JOIN graduate_table g ON p.graduate_program = g.id
""")

df_joined_sql.show()

This SQL JOIN matches the DataFrame "inner" join on join_expr from 10.5.1 (Gita, Rita, Meeta with school names). Outer/semi/anti forms are also available via SQL keywords where supported by Spark’s dialect.

10.6.3 Plan Parity Between DataFrame API and Spark SQL

DataFrames API (PySpark/Scala)  \
                                 ==> Logical Plan ==> Catalyst Optimizer ==> RDD Execution
Raw SQL Queries (spark.sql)     /

Real-world domain connection: Mixed teams put engineers on PySpark ETL, analysts on Spark SQL dashboards, and still share one optimized runtime — the multi-language story in 10.8.1.

10.7 Exam Guidance Summary

10.7.1 Assignment Submission and Video Demo Requirements

Document structure — include screenshots that verify:

1. Local environment connection and setup commands.
2. Program source sections that show the required algorithms.
3. Console commands and intermediate outputs.
4. Input parameters and final console outputs.

Observer clarity: Use clean, annotated layouts. Do not submit unformatted wall-of-text dumps.

Video demo format: Keep videos short and focused on command execution and runtime outputs (directory navigation, Sqoop ingestion, Hive queries, MapReduce results on the prompt). Avoid long unedited recordings.

Infrastructure guidance: Native Hadoop 3.x on Windows 10 often fails on WinUtils path issues. Prefer Hadoop 2.7/2.6, Linux/Ubuntu, or managed clouds (AWS EMR, Databricks).

Evaluation weightage: Command execution typically carries about 40%–50% of credit; wiring an end-to-end pipeline (for example Sqoop → Hive → MapReduce analytics) carries about 50%–60%.

10.8 Key Industry Applications

10.8.1 Multi-Language Enterprise Data Analytics

Enterprise analytics platforms use Spark DataFrames so mixed teams share one optimized runtime. Data engineers often build ETL in Scala or Java for tight JVM integration; data scientists use PySpark; business analysts query the same lakes with Spark SQL. All three surfaces compile through Catalyst into optimized RDD execution, which is why language choice is mostly about team skill and packaging — not a different optimizer. This unified architecture is a central theme in Spark: The Definitive Guide and matches the plan-parity story in 10.6.3.

10.8.2 Broadcast Join Optimization in Large-Scale ETL

In production data lakes, joining multi-billion-row transaction (fact) tables to small dimension tables — store locations, currency rates, product status codes — via shuffle joins floods the network. Broadcasting the small dimension to every worker lets each node join locally against its fact partitions, cutting shuffle I/O and speeding nightly batch ETL, often by orders of magnitude. That is the same broadcast-vs-shuffle mechanism introduced in 10.5.2, applied at warehouse scale.