# The Hours They Stayed Canonical URL: Domain: Spark · Difficulty: medium · Seniority: junior · Asked in: Tata Consultancy Services ## Problem Each row of content_views is one play of a catalog item, carrying who watched and how many seconds they watched, while the item's format lives in content_items. Report the total watch time and the unique viewer count of every content type that drew more than 28 unique viewers. ## Expected output First 4 of 4 rows: | content_type | total_watch_seconds | unique_viewers | | --- | --- | --- | | article | 57916 | 30 | | livestream | 55672 | 37 | | short | 59890 | 29 | | video | 60583 | 29 | ## Worked solution and explanation ### What this problem is really testing This is a per-format audience count dressed up as a watch-time report, and the whole answer turns on one word: unique. Every row of `content_views` is a play, not a person. The same `user_id` shows up on several plays and several `device_id`s, and 28 plays carry no `user_id` at all. Count rows and every format looks bigger than it is. On this seed podcast sits at exactly 28 unique viewers, so a row count or a `>= 28` slips it past a bar it does not clear. The second trap pulls the other way: drop the anonymous plays, or deduplicate plays before summing, and the watch-time total shrinks, because those seconds were really watched. --- ### Build it in three moves #### Step 1: Attach the format from `content_items` `watch_seconds` and `user_id` live on the play; `content_type` lives on the catalog row. Inner-join on `content_id` so every play picks up its format. The catalog holds one row per `content_id`, so the join never duplicates a play and the watch-time sum stays honest. #### Step 2: Collapse to one row per `content_type` Compute `F.sum("watch_seconds")` and `F.countDistinct("user_id")` in the same `agg`. `countDistinct` skips the `NULL` viewers on its own while `sum` still adds their seconds. That is exactly the split the business wants: anonymous time counts, but anonymous plays add no audience. Name the result `per_type`. #### Step 3: Hold the bar against `unique_viewers` The 'more than 28' rule is about a per-format figure that does not exist until after the `agg`, so the filter goes on `per_type`. That is the DataFrame API's HAVING. Put it before the `groupBy` and there is no `unique_viewers` column to test. --- ### The solution **Aggregate first, then hold the bar** ```python from pyspark.sql import functions as F views = spark.table("content_views") items = spark.table("content_items") per_type = ( views .join(items, "content_id") .groupBy("content_type") .agg( F.sum("watch_seconds").alias("total_watch_seconds"), F.countDistinct("user_id").alias("unique_viewers"), ) ) result = per_type.filter(F.col("unique_viewers") > 28) ``` *The bar is tested on `per_type`, after the `agg` has produced `unique_viewers`.* **Counting plays** `F.count("user_id")` counts every play with a viewer attached. Repeat plays by one person, or one person on two devices, each add one. Every format's audience is overstated, and podcast can clear a bar it never reached. **Counting people** `F.countDistinct("user_id")` counts each person once per format, whatever the device and however many replays. Podcast lands on 28 and is correctly left out; the other four formats keep their true audience. **Pipeline stages** - plays (source) - catalog (source) - attach format (transform, Spark) - per-format totals (transform, Spark) - audience bar (transform, Spark) - result (consumer) **Data flow** - plays -> attach format - catalog -> attach format - attach format -> per-format totals - per-format totals -> audience bar - audience bar -> result *Spark broadcasts the small catalog on its own; the one real shuffle is the aggregation on `content_type`.* > **The catalog travels, the plays stay put** > > `content_items` is a couple hundred rows, far under `spark.sql.autoBroadcastJoinThreshold` (10MB by default). Spark therefore broadcasts it to every executor, and the join runs map-side without moving `content_views`. The only exchange left is the `groupBy`, and `countDistinct` turns it into a two-level aggregation keyed on `content_type` plus `user_id`. With five formats that shuffle is tiny. At 40M plays, the partial aggregation before the exchange is what keeps it that way. > **`>= 28` is a different answer** > > Podcast sits at exactly 28 unique viewers, so the boundary is not hypothetical here. 'More than 28' is strict: `> 28` drops podcast and returns four formats; `>= 28` keeps it and returns five. > **Explain what the anonymous plays do** > > Strong candidates spot the plays with no `user_id` and say out loud that `countDistinct` ignores them while `sum` keeps their seconds. Weaker ones reach for `F.col("user_id").isNotNull()` before the `agg`, which silently shrinks every total. ## Common follow-up questions - One `content_type` takes 90% of plays, and its `groupBy` task runs for an hour while the rest finish in seconds. What do you change? _(Tests reading skew in the stage view and spreading a hot aggregation key across tasks.)_ - If `content_items` grew to 5GB, how would the plan and its cost change? _(Tests when a broadcast stops being viable and a sort-merge join takes over.)_ ## Related - [All practice problems](https://datadriven.io/problems) - [Mock interview mode](https://datadriven.io/interview/the_hours_they_stayed) - [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.