Fix Skewed Viewing Events Pipeline

A hard Spark interview practice problem on DataDriven. Write the Spark code and run it against the tables, with instant feedback.

Domain

Spark

Difficulty

hard

Seniority

L5

Asked in

Databricks

Problem

Two hours into a nightly viewing_engagement run that normally takes 45 minutes against a 60-minute SLA, the on-call engineer is staring at a job that matches the 800M-row-a-day event_data table to the 2M-subscriber users dimension on user_id and counts events per event type and account status. From the Spark UI evidence, find the root cause and fix the job so the large event table is never moved across the cluster, leaving the counts unchanged.

Your nightly Spark job just paged you. One task has 40% of the data.

Expected output

Worked solution and explanation

What this problem actually is

This is a data-skew diagnosis wearing an SLA-miss costume. The job is not slow because the cluster is too small; it is slow because one shuffle partition holds most of the rows. The tell is in the Spark UI: 199 tasks in the stage finished in seconds while a single task, Task 3 on exec-1, has been running for over an hour with 220 GB of shuffle read and 38s of GC. The trap is reading '2h and climbing' as a capacity problem and reaching for more executors, more memory, or a higher spark.sql.shuffle.partitions. None of those move a mega-partition off the one reducer that owns it.

Reading the evidence

Step 1: Find the straggler and its skew

The stuck task read 802M rows and 220 GB of shuffle while the median partition is 4M rows: a 120x skew. The physical plan shows a SortMergeJoin with an Exchange hashpartitioning(user_id, 200) on both sides, so all 800M event rows were shuffled by user_id. Because the top 1% of subscribers generate about 60% of events, one partition swallowed roughly 480M rows.

Step 2: See why Catalyst chose SortMergeJoin

The log spells it out: autoBroadcastJoinThreshold=10m. users=48 MB > threshold. Chose SortMergeJoin. The users dimension is only 2M rows but 48 MB, just above the 10 MB auto-broadcast cutoff, so Spark did NOT broadcast it and fell back to a shuffle join. That single threshold decision is the root cause.

Step 3: Confirm the dimension is broadcastable

48 MB fits comfortably in executor memory: each has 64 GB and the healthy executors are using about 9 GB. A broadcast() hint overrides the threshold and forces the small side to every executor.

The fix: broadcast the dimension

from pyspark.sql import SparkSession
from pyspark.sql.functions import broadcast, count

spark = SparkSession.builder.getOrCreate()

event_data = spark.table("event_data")   # 800M rows/day, skewed on user_id
users = spark.table("users")             # 2M subscribers, ~48 MB

# users (48 MB) sits just above autoBroadcastJoinThreshold (10 MB), so Catalyst
# picked a SortMergeJoin: it hash-partitions all 800M event rows by user_id, and
# because a small set of heavy users owns most of the events, one reducer inherits
# a mega-partition, spills to disk, and strangles the stage.
# broadcast() ships the tiny dimension to every executor instead, so event_data is
# never shuffled and the user_id skew has nowhere to pile up.
engagement = (
    event_data
    .join(broadcast(users), "user_id", "inner")
    .groupBy("event_type", "account_status")
    .agg(count("*").alias("event_count"))
)

engagement.write.mode("overwrite").saveAsTable("viewing_engagement")

With a BroadcastHashJoin, the 48 MB users table is shipped to every executor once and each executor probes its local hash table as it streams event_data. There is no Exchange on the large side, so there is no partition for the hot user_id to concentrate in: the 120x skew simply stops mattering. The only shuffle left is the final groupBy on event_type and account_status, which distributes evenly across a small key space.

SortMergeJoin (before)

Both sides shuffled by user_id; the hot key lands one 480M-row partition on one reducer, which spills to disk and blows the SLA.

BroadcastHashJoin (after)

users broadcast to every executor; event_data streamed with no shuffle, so user_id skew has nowhere to pile up.

Common follow-up questions

• What if the small side were 5 GB instead of 48 MB and could not be broadcast? (Pushes toward AQE skew-join handling (`spark.sql.adaptive.skewJoin.enabled`) or salting the hot `user_id`.)
• How would you confirm the fix held on the next run without watching the UI live? (Tests whether they instrument shuffle-read size and max-task duration as SLA alarms.)