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

You are the on-call data engineer at a streaming company and the nightly viewing_engagement Spark job just paged you: it normally finishes in 45 minutes but has been running for over two hours and is still stuck. The job joins a large event_data table (800M rows/day of viewing, playback, and interaction events) against a small users dimension (2M subscribers) on user_id, then produces daily engagement counts by event type and account status, and its SLA is 60 minutes. Read the Spark UI evidence to find the root cause and fix the job so it meets SLA.

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.)