# Who Holds Attention Canonical URL: Domain: Spark · Difficulty: hard · Seniority: mid · Asked in: TikTok ## Problem Our short-video platform keeps a small video catalog in `content_items` while the view log in `content_views` grows fast, so the creator team wants the catalog broadcast to the views rather than the view log moved for the lookup. Pair each view with its video's creator and report each creator's total seconds watched and number of views, keeping only creators whose videos drew more than 1000 seconds in total. Videos with no creator on record count together as one row with a blank creator, and the creators can come back in any order. ## Expected output First 5 of 46 rows: | creator_id | total_watch_seconds | view_count | | --- | --- | --- | | NULL | 30528 | 26 | | 100 | 3616 | 2 | | 197 | 4670 | 2 | | 294 | 8441 | 4 | | 391 | 11654 | 7 | ## Worked solution and explanation This is a fact-into-dimension join wearing a creator-analytics costume, and it has two places to lose. The logical one: the 1000-second bar belongs to a creator's summed watch time, so a filter on individual views throws away every modest view before it can count, and creators whose many short views add up simply vanish. The physical one: the catalog is a tiny dimension that just carries `content_id` to `creator_id`, yet a default join keys both sides by `content_id` and drags the whole view log across the network to meet it. Broadcast the catalog and the fact never moves. Miss that and your one cheap join becomes the most expensive stage in the job. --- ### The two shuffles, and which one you can kill #### Step 1: Broadcast the catalog, not the view log `content_items` is small enough to fit in executor memory, so wrap it in `F.broadcast()`. Spark ships one copy to every executor and joins locally against whatever view partitions already live there. No shuffle on the big side. Without the hint, Spark may still pick a sort-merge join and repartition the whole view log by `content_id` first. #### Step 2: Accept the `groupBy` shuffle, then watch it for skew Grouping by `creator_id` is a wide transformation: it has to exchange rows so every creator lands on one partition. That shuffle is unavoidable here, but it is where a viral creator with millions of views melts one task while the rest idle. Knowing the difference between the join shuffle (avoidable) and the aggregation shuffle (intrinsic) is the senior tell. #### Step 3: Put the bar on the creator, after the `agg` The more-than-1000-seconds rule reads the summed `total_watch_seconds`, so it cannot run until the aggregation completes. Filter on the aggregated column, never on raw `watch_seconds`. Skip a final sort as well: the answer is a set of creators, and an `orderBy` would buy a second exchange the question never asked for. **Bar on each view** `views.filter(F.col('watch_seconds') > 1000)` before the `groupBy` keeps only single views longer than 1000 seconds. A creator with seven views of 400 seconds each has 2800 seconds of watch time and disappears entirely. **Bar on each creator** `.filter(F.col('total_watch_seconds') > 1000)` after the `agg` reads the summed column, so every view counts toward its creator before the cut. Same one-line filter, one step later, a different answer. --- ### The job, stage by stage **Pipeline stages** - content_views (fact) (source) - content_items (small) (source) - broadcast join (transform) - groupBy creator_id (transform) - post-agg filter (transform) - per-creator result (consumer) **Data flow** - content_items (small) -> broadcast join - content_views (fact) -> broadcast join - broadcast join -> groupBy creator_id - groupBy creator_id -> post-agg filter - post-agg filter -> per-creator result *The broadcast join keeps the fact in place; the only network exchange is the `groupBy`.* ### The solution **Per-creator watch summary** ```python from pyspark.sql import functions as F views = spark.table('content_views') catalog = spark.table('content_items') result = ( views .join(F.broadcast(catalog), 'content_id') .groupBy('creator_id') .agg( F.sum('watch_seconds').alias('total_watch_seconds'), F.count('view_id').alias('view_count'), ) .filter(F.col('total_watch_seconds') > 1000) ) ``` *`F.broadcast()` pins the small catalog; the only shuffle left is the unavoidable `groupBy`.* > **Two big shuffles become one small one** > > At tens of millions of view rows against a catalog that fits in executor memory, a sort-merge join would sort and shuffle both sides: two exchanges, both dominated by the fact. The broadcast join removes the fact-side exchange entirely. The catalog, tens of MB once projected to `content_id` and `creator_id`, is sent once per executor and the views stay on their original partitions. The one remaining shuffle moves only the aggregated key space. > **The blank creator is a real row** > > Some catalog items carry no `creator_id`, yet their views still match on `content_id`. The inner join keeps them, and `groupBy('creator_id')` gathers them under one `NULL` key, which clears the bar comfortably on this data. An `isNotNull` filter would quietly delete that row, and the question never asked for that. > **Name the broadcast side and the threshold behind it** > > Say out loud why the catalog is the broadcast side (it is the small one), and name `spark.sql.autoBroadcastJoinThreshold` (default 10MB), which decides whether Spark does this for you. Then volunteer that the `groupBy` still shuffles and could skew on a viral creator. That sentence separates knowing the join API from understanding the physical plan. > **Broadcast the small side, never the fact** > > Calling `F.broadcast(views)` on the view log tries to collect it to the driver and ship it everywhere, which blows the driver heap or silently falls back to a regular join. Broadcast is for the SMALL table only. --- ## Common follow-up questions - One viral creator owns 30% of all views. The `groupBy` stage hangs on a single task. What do you do? _(Tests skew handling: salting the key, AQE skewed-partition handling, or pre-aggregating.)_ - What happens if you join this one large view log against many small dimension tables at once? _(Tests whether they know each broadcast is cheap alone, but many concurrent broadcasts pressure the driver; chaining vs. a single denormalized broadcast.)_ - The catalog grows past the broadcast threshold. How does the plan change and how would you keep it fast? _(Tests the fallback to sort-merge join and mitigations like bucketing both tables on `content_id` to avoid the shuffle.)_ ## Related - [All practice problems](https://datadriven.io/problems) - [Mock interview mode](https://datadriven.io/interview/who_holds_attention) - [PySpark Interview Questions](https://datadriven.io/pyspark-interview-questions) - [Data Engineering Interview Prep Guide](https://datadriven.io/data-engineer-interview-prep) - [Daily Challenge](https://datadriven.io/daily) --- Source: DataDriven (https://datadriven.io). DataDriven is the data engineering interview community. Live code execution in SQL, Python, and Spark sandboxes. Every feature is open to every member.