media_ml/spotify_excercise_etl.md

# Exercise: Spotify Streaming History ETL Pipeline (Two Users)

**Scenario:** You have exported the streaming history of two Spotify users (for example, you and a friend) and want to compare their listening habits. Spotify's "Extended streaming history" export provides JSON files such as `Streaming_History_Audio_2023.json`. Place each user's files in their own folder: `data/user_a/` and `data/user_b/`.

**Goal:** Build a PySpark pipeline that combines both histories into one clean dataset, then answers questions like "who listens more?", "what artists do they share?", and "when do they listen?"

---

## Extract Phase

**Task 1.0.1: Create a SparkSession**
- Name the app and run it in local mode.

**Task 1.0.2: Read each user's JSON files**
- Use a wildcard so all yearly files for a user load at once.
- Read user A and user B into separate DataFrames.
- Inspect the schema and row counts. Note which columns are nested, null, or oddly named (the export has fields like timestamp, track name, artist name, album name, ms played, skipped, platform, and reason for starting/ending).

**Task 1.0.3: Tag the source**
- Add a `user` column to each DataFrame so you know who each row belongs to.
- Optionally capture the source file name per row.

**Task 1.0.4: Combine the two DataFrames**
- Stack them into a single DataFrame with a union, matching columns by name rather than position.
- Verify the combined row count equals the sum of both users' counts.

**(Optional) Enrich from a REST API**
- Call the Spotify Web API with a bearer token to fetch extra track metadata (genre, popularity, duration) for a sample of track IDs, and load it into a lookup DataFrame.

---

## Transform Phase

**Task 2.0.1: Rename and select columns**
- Rename awkward columns to a consistent snake_case style.
- Keep only the columns you need and drop the rest.

**Task 2.0.2: Fix data types**
- Convert the timestamp string to a proper timestamp type.
- Convert milliseconds played to a numeric type, then derive minutes played.

**Task 2.0.3: Handle dirty and missing data**
- Remove rows with null track or artist names (these are often podcasts or unknown items).
- Remove rows where playback was extremely short (for example, under 30 seconds) since Spotify does not count those as real listens. Decide and document your threshold.
- Trim whitespace and standardize casing in artist and track names.
- Drop duplicate rows.

**Task 2.0.4: Derive new columns**
- Extract year, month, day of week, and hour from the timestamp.
- Create a time-of-day bucket (morning, afternoon, evening, night).
- Create a boolean "was skipped" flag, handling nulls sensibly.
- Create a unique track key combining artist and track name, for use in joins.

**Task 2.0.5: Aggregate per user**
- Total listening time and number of plays per user.
- Top 10 artists and top 10 tracks per user (use a window function to rank within each user).
- Listening activity by hour of day and by day of week.
- Skip rate per user.

**Task 2.0.6: Join the two users**
- Build a per-user artist play table, then join both on artist name to find **shared artists**.
- Compute an overlap measure, such as the number of shared artists divided by the total number of distinct artists.
- Use a left anti join to find artists unique to one user.

**Task 2.0.7: Redo part of it with Spark SQL**
- Register the cleaned data as a temp view.
- Write SQL for at least two of the aggregations above (for example, top artists per user using `ROW_NUMBER() OVER (PARTITION BY ...)`).
- Confirm the SQL results match your DataFrame API results.

---

## Load Phase

**Task 3.0.1: Write the clean combined dataset**
- Save it as Parquet, partitioned by `user` and `year`.
- Use overwrite mode so the pipeline can be re-run safely.

**Task 3.0.2: Write the summary tables**
- Save each aggregate (top artists, hourly activity, shared artists) as its own small CSV for sharing.
- Collapse to a single output file since these are small.

**Task 3.0.3: Load to a database (optional)**
- Write the clean data and summaries to MySQL or SQLite via JDBC.
- Choose append vs. overwrite deliberately and explain why.

**Task 3.0.4: Validate the pipeline**
- Assert that the output row count equals the cleaned DataFrame's count.
- Assert there are no nulls in key columns (user, timestamp, artist, track).
- Assert both users are present in the output.
- Assert total minutes played in the summaries matches the cleaned data.

---

## Structure and Stretch Goals

**Structure:** Organize the code into `extract()`, `transform()`, and `load()` functions, plus a `main` that runs them in order and stops the Spark session at the end. Add simple logging that prints row counts after each stage so you can see where data is dropped.

**Stretch goals:**
- Add a third user and generalize your code to loop over any number of user folders.
- Detect listening "sessions" by grouping plays separated by less than 30 minutes of inactivity.
- Compute each user's most-replayed track in a single day.
- Make the pipeline incremental: only process files that were not loaded previously.
- Compare runtime with and without caching the cleaned DataFrame.

**Questions your final output should answer:**
1. Who listened more in total, and during which months?
2. Which artists do both users share, and who plays them more?
3. At what time of day does each user listen most?
4. Who skips more tracks?