One Bad Executor

A medium AI Coding interview practice problem on DataDriven.

Domain
AI Coding
Difficulty
medium
Seniority
L5
Asked in
Uber

Problem

On-call paged at 02:14 last night: the nightly orders enrichment job ran 3h 40m instead of its usual 22 minutes and one executor OOMed on stage 4. Cluster spend tripled. Volumes only grew about 8% week-over-week. The job lives in jobs/. Reproduce by running tests/test_skew_observability.py. Fix it so the test passes. Be ready to defend the trade-offs of your fix when the interviewer asks what happens with a different data shape.

One node held the whole night hostage.

Worked solution and explanation

What actually broke

This is textbook data skew wearing a cost-spike costume. The join key user_id has one monster value , 'guest' , that is about 38% of all 2.4B order rows. A plain hash-shuffle join routes rows by hash(user_id) % 1000, so every 'guest' row lands in ONE reducer partition. That partition's executor tries to hold ~900M rows and OOMs on stage 4 while the other 999 tasks finish in 90 seconds. Volume only grew 8%; the skew is what held the night hostage. Seniority here isn't spotting the skew , it's which fix you reach for, because the obvious one is a trap.

Why the obvious fix is a trap

Broadcast the dim
Skip the shuffle entirely: ship users to every executor and join in-map, so there's no reducer partition to overload. But cluster/constraints.sql records the dim at 9 GB and the broadcast ceiling at 8 GB, so F.broadcast(users) fails planner validation in production. The ~2 GB note in jobs/io.py is stale bait.
Salt the hot key
Keep the shuffle but spread the hot value: add a salt bucket 0..N-1 to the orders side, replicate each users row across all N buckets, and join on (user_id, salt). 'guest' now hashes across N partitions instead of one. Cost is an N-fold copy of the dim , cheap only because the dim is the small side.
Salt the skewed join
# jobs/tuning.py
def salt_skewed_key(df: DataFrame, key_col: str, salt_buckets: int) -> DataFrame:
    """Spread a hot key across salt_buckets partitions."""
    return df.withColumn("salt", (F.rand() * salt_buckets).cast("int"))


# jobs/enrich_orders.py
from jobs.tuning import salt_skewed_key

SALT_BUCKETS = 16

def run(spark: SparkSession) -> int:
    orders = read_orders(spark)
    users  = read_users(spark)

    # users is 9 GB > 8 GB broadcast threshold (cluster/constraints.sql),
    # so broadcast is off the table. Salt orders, then replicate the dim
    # across every bucket so each salted order still finds its user.
    salted_orders = salt_skewed_key(orders, "user_id", SALT_BUCKETS)
    salted_users = users.withColumn(
        "salt", F.explode(F.array([F.lit(i) for i in range(SALT_BUCKETS)]))
    )

    enriched = salted_orders.join(salted_users, on=["user_id", "salt"], how="left")

    daily = (
        enriched
        .groupBy("order_date", "country")
        .agg(
            F.count("*").alias("order_count"),
            F.sum("amount_cents").alias("revenue_cents"),
        )
    )
    return write_enriched(daily)

The salt column makes 'guest' hash across 16 partitions; the 16x dim copy is cheap at 9 GB but would be ruinous on a large table.

Common follow-up questions