Apache Spark for Data Engineers: RDDs, DataFrames, and Structured Streaming

Scenario context

Apache Spark distributes Python DataFrame jobs across a cluster of executors, using lazy evaluation and the Catalyst query optimizer to process terabytes with the same code that works on gigabytes. Master partitioning, shuffle-awareness, and Structured Streaming and you have a single engine for both batch and real-time pipelines.

<!--

Picture this: a data engineer writes a clean pandas script that reads 500MB of e…

This walkthrough focuses on When Pandas Hits the Wall: The 500GB Problem.

Before making a design choice, clarify:

• Picture this: a data engineer writes a clean pandas script that reads 500MB of e-commerce event data, filters it, joins it to a product catalog, and produces a category rollup.
• It runs in 40 seconds on a laptop.
• The team ships it to production.

Concepts in play: spark-data-engineering, jvm-internals-and-garbage-collection, database-sharding, data-warehousing-lakehouse

Stage 1 — RDDs: Spark's Original Distributed Dataset Abstraction

• Before DataFrames, Spark exposed data as Resilient Distributed Datasets (RDDs) .
• An RDD is an immutable, fault-tolerant, distributed collection of records partitioned across a cluster.
• Three properties define it: - Immutable : you never modify an RDD in place; every transformation produces a new RDD.
• • Distributed : partitions live on different executor nodes and are processed in parallel.

# RDD-style wordcount — verbose and unoptimized
lines = sc.textFile("s3://bucket/logs/*.txt")
words = lines.flatMap(lambda line: line.split())
pairs = words.map(lambda w: (w, 1))
counts = pairs.reduceByKey(lambda a, b: a + b)
counts.saveAsTextFile("s3://bucket/output/wordcount")   # ← action triggers execution

Observable outcome: - Distributed : partitions live on different executor nodes and are processed in parallel.

Stage 2 — Driver, Executors, and the DAG: How Spark Distributes Work

• Understanding where your code runs is the most important mental model for a Spark engineer.
• Driver : the process running your Python script.
• It hosts the SparkSession , parses your DataFrame transformations, builds a logical plan, hands it to the Catalyst optimizer, and ultimately sends tasks to executors.
• The driver is the brain — it does no data crunching.

# DataFrame-style — shorter, and Catalyst-optimized
from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, split, col

spark = SparkSession.builder.appName("wordcount").getOrCreate()
df = spark.read.text("s3://bucket/logs/*.txt")
counts = (
    df.select(explode(split(col("value"), " ")).alias("word"))
      .groupBy("word")
      .count()
)
counts.write.parquet("s3://bucket/output/wordcount")

Observable outcome: The driver is the brain — it does no data crunching.

Stage 3 — Visualizing Spark's Execution: From Python Call to JVM Task

• The journey from a Python groupBy().count() call to bytes on disk passes through five layers: Key insight : stages are separated by shuffles — the most expensive operation in Spark, because shuffle requires moving data…
• Every groupBy , join (non-broadcast), and repartition triggers a shuffle.
• Minimizing shuffles is the single most impactful Spark performance tuning lever.

flowchart TD
    A[" Python DataFrame API (filter  join  groupBy)"] --> B["Logical Plan (unresolved operators)"]
    B --> C["Catalyst Optimizer (predicate pushdown  column pruning join reordering)"]
    C --> D["Optimized Physical Plan (scan  hash-aggregate  sort-merge-join)"]
    D --> E["DAG of Stages (split at shuffle boundaries)"]
    E --> F["Tasks (one per partition)"]
    F --> G["Executor Cores (JVM  Tungsten UnsafeRow)"]
    G --> H[" Output (S3 / HDFS / Delta Lake)"]

Observable outcome: Minimizing shuffles is the single most impactful Spark performance tuning lever.

Stage 4 — Deep Dive: Catalyst, Tungsten, and How Python DataFrames Become JVM Bytecode

• Spark's Internals: DAGScheduler, BlockManager, and Shuffle When an action fires, the DAGScheduler takes the optimized physical plan and divides it into stages.
• A new stage begins whenever a shuffle is required.
• Within a stage, tasks are independent — they can run fully in parallel without exchanging data.
• The TaskScheduler assigns tasks to available executor slots (cores × threads).

Observable outcome: The TaskScheduler assigns tasks to available executor slots (cores × threads).

Stage 5 — Processing a 1-Billion-Row E-Commerce Clickstream in PySpark

• Example: Category-Level Add-to-Cart Funnel The dataset: {user id, product id, event type, timestamp} stored as Parquet on S3, partitioned by date .
• Goal: count add to cart events per product category over the last 30 days.
• What's happening under the hood: - filter(col("date") = ...) triggers partition pruning — Spark only lists and reads the last 30 date-partitioned directories from S3, scanning perhaps 30 TB of a 365-day archive down to…
• • The products table is small enough for a broadcast join : the driver fetches a copy and pushes it to every executor, eliminating the shuffle entirely.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, to_date, current_date, date_sub

spark = (
    SparkSession.builder
    .appName("clickstream-category-funnel")
    .config("spark.sql.shuffle.partitions", "400")          # tune for ~128 MB per partition
    .config("spark.sql.autoBroadcastJoinThreshold", "50mb") # broadcast the small products table
    .getOrCreate()
)

# ── 1. Read partitioned Parquet — Spark only scans the last-30-days date partitions
events = (
    spark.read
    .parquet("s3://data-lake/clickstream/")
    .filter(col("event_type") == "add_to_cart")
    .filter(col("date") >= date_sub(current_date(), 30))    # partition pruning
)

# ── 2. Load the products dimension table (small — fits in broadcast memory)
products = spark.read.parquet("s3://data-lake/products/")   # ~200 MB — triggers broadcast join

# ── 3. Join to get category — broadcast join avoids a shuffle on the products side
enriched = events.join(products, on="product_id", how="inner")

# ── 4. Aggregate: count add_to_cart events per category
result = (
    enriched
    .groupBy("category")
    .count()
    .withColumnRenamed("count", "add_to_cart_events")
    .orderBy(col("add_to_cart_events").desc())
)

# ── 5. Write as partitioned Parquet — downstream jobs can prune by run_date
result.write.mode("overwrite").parquet("s3://data-lake/reports/category_funnel/")

spark.stop()

Observable outcome: - The products table is small enough for a broadcast join : the driver fetches a copy and pushes it to every executor, eliminating the shuffle entirely.

Stage 6 — Real-World Applications: Spark SQL and Structured Streaming at Scale

• Spark SQL: Running SQL on DataFrames Any DataFrame can be registered as a temporary view and queried with standard SQL.
• This is not a conversion — it is the same Catalyst optimizer, the same physical plan.
• SQL and DataFrame API are interchangeable: Spark SQL integrates with the Hive Metastore: you can define tables with schemas, partitioning, and SerDes, then query them with spark.sql("SELECT ...") without ever specifying…
• This is how most warehouse-style ETL jobs are structured in practice.

# Register the enriched DataFrame as a SQL-queryable view
enriched.createOrReplaceTempView("clickstream_enriched")

# Run SQL — Catalyst compiles it identically to the DataFrame API
top_categories = spark.sql("""
    SELECT category,
           COUNT(*) AS add_to_cart_events
    FROM   clickstream_enriched
    WHERE  event_type = 'add_to_cart'
    GROUP  BY category
    ORDER  BY add_to_cart_events DESC
    LIMIT  20
""")
top_categories.show()

# Streaming word count from Kafka — same DataFrame API, continuous execution
words = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker:9092")
    .option("subscribe", "logs-topic")
    .load()
    .selectExpr("CAST(value AS STRING) AS line")
    .selectExpr("explode(split(line, ' ')) AS word")
)

word_counts = words.groupBy("word").count()

query = (
    word_counts.writeStream
    .format("delta")                          # Delta Lake for ACID output
    .outputMode("update")                     # emit only changed rows each micro-batch
    .option("checkpointLocation", "s3://checkpoints/wordcount/")
    .start("s3://data-lake/wordcount/")
)

query.awaitTermination()

Observable outcome: This is how most warehouse-style ETL jobs are structured in practice.

End-to-end outcome

After working through the stages, you should be able to explain:

• Don't use the default shuffle partition count ( spark.sql.shuffle.partitions=200 ).
• For a 10 GB dataset that's 50 MB per partition — fine.

Alternatives considered

Partitioning Strategy: The 128MB Rule, repartition vs coalesce, and Column Partitioning

Partition count controls parallelism.

Decision Guide: When to Use Spark vs Alternatives

Decision Guide: When to Use Spark vs Alternatives

Failure and degraded modes

Performance vs. Operational Complexity

• Detection: Spark delivers horizontal scale but introduces non-trivial operational overhead: cluster sizing, executor memory tuning, shuffle partition selection, and network-intensive shuffle operations all requ…
• User or operator impact: Spark delivers horizontal scale but introduces non-trivial operational overhead: cluster sizing, executor memory tuning, shuffle partition selection, and network-intensive shuffle operations all requ…

Failure Modes

• Detection: Data skew — one partition is 10× larger than the rest because the join key is highly concentrated (e.g., a single user id generated 40% of events).
• User or operator impact: Data skew — one partition is 10× larger than the rest because the join key is highly concentrated (e.g., a single user id generated 40% of events).

Mitigation Patterns

• Detection: - Profile with Spark UI (port 4040): examine stage timelines, task duration distributions, and shuffle read/write sizes.
• User or operator impact: - Profile with Spark UI (port 4040): examine stage timelines, task duration distributions, and shuffle read/write sizes.