214 lines
8.9 KiB
Python
214 lines
8.9 KiB
Python
# Databricks notebook source
|
|
"""
|
|
Bronze -> Silver for TWO users' Spotify Extended Streaming History.
|
|
|
|
ASSUMED BRONZE: {catalog}.bronze.spotify_streaming_history
|
|
raw export columns as landed (ts, ms_played, platform, conn_country,
|
|
master_metadata_track_name, master_metadata_album_artist_name,
|
|
master_metadata_album_album_name, spotify_track_uri, episode_name,
|
|
episode_show_name, spotify_episode_uri, reason_start, reason_end,
|
|
shuffle, skipped, offline, incognito_mode, ip_addr_decrypted, user_agent_decrypted)
|
|
+ ingestion columns: _user_label ('user_a'/'user_b', taken from the landing folder,
|
|
NOT from the raw username field), _source_file, _ingested_at
|
|
|
|
Key facts baked in:
|
|
* `ts` is when the play ENDED (UTC), so start = ts - ms_played.
|
|
* Exports can overlap (re-requested downloads) -> dedupe.
|
|
* Skips are signal, so flag them, don't delete them.
|
|
* IP address and user agent are dropped in silver (privacy minimization).
|
|
"""
|
|
dbutils.widgets.text("catalog", "dev")
|
|
CAT = dbutils.widgets.get("catalog")
|
|
|
|
spark.sql(f"CREATE SCHEMA IF NOT EXISTS {CAT}.silver")
|
|
spark.table(f"{CAT}.bronze.spotify_streaming_history").createOrReplaceTempView("bronze_v")
|
|
|
|
# Per-user home time zone (replace with your real config table)
|
|
spark.sql("""
|
|
CREATE OR REPLACE TEMP VIEW user_config AS
|
|
SELECT * FROM VALUES ('user_a', 'Europe/London'), ('user_b', 'America/New_York')
|
|
AS t(user_id, home_tz)
|
|
""")
|
|
|
|
# COMMAND ----------
|
|
# MAGIC %md ## Step 1: type, rename, normalize (no rows lost yet)
|
|
|
|
# COMMAND ----------
|
|
|
|
spark.sql("""
|
|
CREATE OR REPLACE TEMP VIEW typed AS
|
|
SELECT
|
|
_user_label AS user_id,
|
|
to_timestamp(ts) AS ended_at_utc,
|
|
CAST(ms_played AS BIGINT) AS ms_played,
|
|
CASE WHEN lower(platform) LIKE '%ios%' THEN 'ios'
|
|
WHEN lower(platform) LIKE '%android%' THEN 'android'
|
|
WHEN lower(platform) RLIKE 'windows|osx|mac|linux' THEN 'desktop'
|
|
WHEN lower(platform) LIKE '%web%' THEN 'web'
|
|
ELSE 'other' END AS platform,
|
|
conn_country AS country,
|
|
CASE WHEN spotify_track_uri IS NOT NULL THEN 'track'
|
|
WHEN spotify_episode_uri IS NOT NULL THEN 'episode'
|
|
ELSE 'other' END AS content_type,
|
|
NULLIF(regexp_extract(spotify_track_uri, 'spotify:track:([A-Za-z0-9]+)', 1), '') AS track_id,
|
|
NULLIF(regexp_extract(spotify_episode_uri, 'spotify:episode:([A-Za-z0-9]+)', 1), '') AS episode_id,
|
|
trim(master_metadata_track_name) AS track_name,
|
|
trim(master_metadata_album_artist_name) AS artist_name,
|
|
trim(master_metadata_album_album_name) AS album_name,
|
|
episode_show_name AS show_name,
|
|
reason_start, reason_end,
|
|
COALESCE(shuffle, false) AS shuffle,
|
|
skipped, offline,
|
|
COALESCE(incognito_mode, false) AS incognito_mode,
|
|
_source_file
|
|
FROM bronze_v
|
|
WHERE ts IS NOT NULL
|
|
AND ms_played IS NOT NULL AND ms_played >= 0 -- impossible values out
|
|
""")
|
|
|
|
# COMMAND ----------
|
|
# MAGIC %md ## Step 2: dedupe overlapping exports + derive start time and flags
|
|
|
|
# COMMAND ----------
|
|
|
|
spark.sql("""
|
|
CREATE OR REPLACE TEMP VIEW cleaned AS
|
|
SELECT
|
|
d.*,
|
|
timestamp_millis(unix_millis(ended_at_utc) - ms_played) AS started_at_utc,
|
|
ms_played >= 30000 AS is_stream, -- Spotify's 30s rule
|
|
COALESCE(skipped, reason_end = 'fwdbtn' AND ms_played < 30000) AS is_skip,
|
|
reason_end = 'trackdone' AS is_complete,
|
|
(content_type = 'track' AND ms_played > 3600000) AS is_outlier, -- >1h on one track
|
|
from_utc_timestamp(timestamp_millis(unix_millis(ended_at_utc) - ms_played), c.home_tz)
|
|
AS started_at_local
|
|
FROM (
|
|
SELECT * EXCEPT (rn) FROM (
|
|
SELECT *, row_number() OVER (
|
|
PARTITION BY user_id, ended_at_utc, track_id, episode_id
|
|
ORDER BY ms_played DESC) AS rn
|
|
FROM typed)
|
|
WHERE rn = 1
|
|
) d
|
|
LEFT JOIN user_config c USING (user_id)
|
|
""")
|
|
|
|
# COMMAND ----------
|
|
# MAGIC %md ## Step 3: sessionize (30 min gap, per user) + sequence features
|
|
|
|
# COMMAND ----------
|
|
|
|
spark.sql("""
|
|
CREATE OR REPLACE TEMP VIEW sessioned AS
|
|
SELECT *,
|
|
concat(user_id, '-', session_seq) AS session_id,
|
|
row_number() OVER (PARTITION BY user_id, session_seq ORDER BY started_at_utc) AS pos_in_session,
|
|
lag(track_id) OVER (PARTITION BY user_id ORDER BY started_at_utc) AS prev_track_id
|
|
FROM (
|
|
SELECT *,
|
|
SUM(new_session) OVER (PARTITION BY user_id ORDER BY started_at_utc
|
|
ROWS UNBOUNDED PRECEDING) AS session_seq
|
|
FROM (
|
|
SELECT *,
|
|
CASE WHEN lag(ended_at_utc) OVER (PARTITION BY user_id ORDER BY started_at_utc) IS NULL
|
|
OR started_at_utc > lag(ended_at_utc) OVER (PARTITION BY user_id ORDER BY started_at_utc)
|
|
+ INTERVAL 30 MINUTES
|
|
THEN 1 ELSE 0 END AS new_session
|
|
FROM cleaned
|
|
)
|
|
)
|
|
""")
|
|
|
|
# COMMAND ----------
|
|
# MAGIC %md ## Step 4: discovery + cross-user features (what makes two users special)
|
|
|
|
# COMMAND ----------
|
|
|
|
spark.sql(f"""
|
|
CREATE OR REPLACE TABLE {CAT}.silver.spotify_plays AS
|
|
WITH first_play AS ( -- first real listen per user and track
|
|
SELECT user_id, track_id, MIN(started_at_utc) AS first_play_utc
|
|
FROM sessioned WHERE is_stream AND track_id IS NOT NULL
|
|
GROUP BY user_id, track_id
|
|
)
|
|
SELECT s.*,
|
|
s.started_at_utc = fp.first_play_utc AS is_first_listen,
|
|
other.first_play_utc AS other_user_first_play_utc,
|
|
other.first_play_utc IS NOT NULL
|
|
AND other.first_play_utc < s.started_at_utc AS other_user_heard_first,
|
|
hour(s.started_at_local) AS hour_local,
|
|
dayofweek(s.started_at_local) IN (1, 7) AS is_weekend,
|
|
CAST(s.started_at_local AS DATE) AS play_date_local
|
|
FROM sessioned s
|
|
LEFT JOIN first_play fp ON fp.user_id = s.user_id AND fp.track_id = s.track_id
|
|
LEFT JOIN first_play other ON other.track_id = s.track_id AND other.user_id <> s.user_id
|
|
""")
|
|
|
|
# COMMAND ----------
|
|
# MAGIC %md ## Step 5: data-quality checks (fail loudly)
|
|
|
|
# COMMAND ----------
|
|
|
|
dq = spark.sql(f"""
|
|
SELECT user_id,
|
|
COUNT(*) AS rows,
|
|
MIN(started_at_utc) AS first_play, MAX(started_at_utc) AS last_play,
|
|
ROUND(AVG(CAST(track_id IS NULL AND content_type = 'track' AS INT)), 4) AS null_track_share,
|
|
SUM(CAST(is_outlier AS INT)) AS outliers,
|
|
SUM(CAST(started_at_utc > current_timestamp() AS INT)) AS future_rows
|
|
FROM {CAT}.silver.spotify_plays GROUP BY user_id
|
|
""")
|
|
dq.show(truncate=False)
|
|
assert dq.filter("future_rows > 0").count() == 0, "future timestamps found"
|
|
assert dq.count() == 2, "expected exactly two users"
|
|
|
|
# COMMAND ----------
|
|
# MAGIC %md ## Two-user analytics: taste overlap and cross-recommendation candidates
|
|
|
|
# COMMAND ----------
|
|
|
|
# Artist-level Jaccard and cosine similarity between the two users
|
|
spark.sql(f"""
|
|
WITH ua AS (
|
|
SELECT user_id, artist_name, COUNT(*) AS plays
|
|
FROM {CAT}.silver.spotify_plays
|
|
WHERE is_stream AND content_type = 'track' AND artist_name IS NOT NULL
|
|
GROUP BY 1, 2
|
|
),
|
|
shared AS (
|
|
SELECT a.artist_name, a.plays AS pa, b.plays AS pb
|
|
FROM ua a JOIN ua b ON a.artist_name = b.artist_name
|
|
WHERE a.user_id = 'user_a' AND b.user_id = 'user_b'
|
|
),
|
|
sizes AS (
|
|
SELECT SUM(CAST(user_id = 'user_a' AS INT)) AS n_a,
|
|
SUM(CAST(user_id = 'user_b' AS INT)) AS n_b,
|
|
SQRT(SUM(CASE WHEN user_id = 'user_a' THEN plays * plays END)) AS norm_a,
|
|
SQRT(SUM(CASE WHEN user_id = 'user_b' THEN plays * plays END)) AS norm_b
|
|
FROM ua
|
|
)
|
|
SELECT COUNT(*) AS shared_artists,
|
|
COUNT(*) / (MAX(n_a) + MAX(n_b) - COUNT(*)) AS jaccard,
|
|
SUM(pa * pb) / (MAX(norm_a) * MAX(norm_b)) AS cosine_plays
|
|
FROM shared CROSS JOIN sizes
|
|
""").show()
|
|
|
|
# Candidates for user A: tracks B plays a lot that A has never played,
|
|
# preferring artists A already listens to
|
|
spark.sql(f"""
|
|
WITH aff AS (
|
|
SELECT user_id, track_id, artist_name, MAX(track_name) AS track_name, COUNT(*) AS plays
|
|
FROM {CAT}.silver.spotify_plays
|
|
WHERE is_stream AND content_type = 'track' AND track_id IS NOT NULL
|
|
GROUP BY 1, 2, 3
|
|
),
|
|
a_artists AS (SELECT artist_name, SUM(plays) AS plays FROM aff WHERE user_id = 'user_a' GROUP BY 1)
|
|
SELECT b.track_name, b.artist_name, b.plays AS b_plays,
|
|
COALESCE(aa.plays, 0) AS a_plays_for_artist
|
|
FROM aff b
|
|
LEFT JOIN aff a_trk ON a_trk.user_id = 'user_a' AND a_trk.track_id = b.track_id
|
|
LEFT JOIN a_artists aa ON aa.artist_name = b.artist_name
|
|
WHERE b.user_id = 'user_b' AND a_trk.track_id IS NULL
|
|
ORDER BY (COALESCE(aa.plays, 0) > 0) DESC, b.plays DESC
|
|
LIMIT 20
|
|
""").show(truncate=False) |