CC/media_ml
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: CC:bb36c54f961816abfb3cbcb9243dfaf68a0854c3
Branches Tags
CC:main
...
compare: CC:main
Branches Tags
CC:main

10 Commits
bb36c54f96 ... main

Author SHA1 Message Date
CC
f2782cb1a8 ml 2026-10-19 08:27:26 +01:00
CC
95a7d229e9 ml 2026-10-19 08:23:52 +01:00
CC
dbcbad6d64 descriptions 2026-10-18 23:26:25 +01:00
CC
72b501a5ab descriptions 2026-10-18 23:25:48 +01:00
CC
8fc40a9015 databricks 2026-10-18 22:59:56 +01:00
CC
7fb281f9db databricks 2026-10-18 22:55:40 +01:00
CC
7b6803b1ee local vscode 2026-10-18 22:25:12 +01:00
CC
f409cefa18 siv databricks 2026-10-18 22:00:35 +01:00
CC
f03814ab93 silver load from jdbc 2026-10-18 18:38:51 +01:00
CC
dc6b08eeeb bronze -> silver postgres 2026-10-18 18:05:53 +01:00
9 changed files with 646 additions and 1010 deletions
Expand all files Collapse all files

0
claude/business_model.md → business_model.md
View File

153
claude/bronze_foreign.py
View File

@@ -1,153 +0,0 @@
# Databricks notebook source
# MAGIC %md
# MAGIC # Bronze ingestion: Redshift foreign catalog -> Delta bronze (Trakt example)
# MAGIC The Redshift login lives in a Unity Catalog **connection**; the **foreign catalog**
# MAGIC built on it is what shows up in the Catalog Explorer UI. No secrets, JDBC URL or S3
# MAGIC tempdir in this notebook. Foreign tables are read-only, so we land into a native Delta catalog.
# COMMAND ----------
import json
import uuid
from datetime import datetime, timezone
from pyspark.sql import functions as F
dbutils.widgets.text("source_catalog", "redshift_bronze") # foreign catalog (Redshift connection)
dbutils.widgets.text("source_schema", "trakt_raw")
dbutils.widgets.text("target_catalog", "dev") # native Delta catalog
dbutils.widgets.dropdown("mode", "incremental", ["incremental", "full"])
SRC_CAT = dbutils.widgets.get("source_catalog")
SRC_SCHEMA = dbutils.widgets.get("source_schema")
TGT_CAT = dbutils.widgets.get("target_catalog")
MODE = dbutils.widgets.get("mode")
BATCH_ID = str(uuid.uuid4())
RUN_TS = datetime.now(timezone.utc)
BRONZE = f"{TGT_CAT}.bronze"
WM = f"{TGT_CAT}.ops.ingestion_watermark"
TABLES = {
"watch_history": {"key": ["id"], "cursor": "watched_at"},
"ratings": {"key": ["trakt_id", "rated_at"], "cursor": "rated_at"},
"watchlist": {"key": ["trakt_id", "listed_at"], "cursor": "listed_at"},
}
# COMMAND ----------
# MAGIC %md ## Sanity check the federation (what you see in the UI, via SQL)
# COMMAND ----------
spark.sql(f"SHOW TABLES IN {SRC_CAT}.{SRC_SCHEMA}").show(truncate=False)
spark.sql(f"DESCRIBE EXTENDED {SRC_CAT}.{SRC_SCHEMA}.watch_history").show(50, truncate=False)
# COMMAND ----------
# MAGIC %md ## Target objects
# COMMAND ----------
spark.sql(f"CREATE SCHEMA IF NOT EXISTS {BRONZE}")
spark.sql(f"CREATE SCHEMA IF NOT EXISTS {TGT_CAT}.ops")
spark.sql(f"""
CREATE TABLE IF NOT EXISTS {WM} (
source_table STRING, last_cursor TIMESTAMP,
last_batch_id STRING, updated_at TIMESTAMP
) USING DELTA
""")
spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", "true")
# COMMAND ----------
# MAGIC %md ## Helpers
# COMMAND ----------
def get_watermark(name):
if MODE == "full":
return None
row = spark.sql(f"SELECT last_cursor FROM {WM} WHERE source_table = '{name}'").first()
return row["last_cursor"] if row else None
def set_watermark(name, value):
if value is None:
return
ts = value.strftime("%Y-%m-%d %H:%M:%S.%f")
spark.sql(f"""
MERGE INTO {WM} t
USING (SELECT '{name}' AS source_table, TIMESTAMP'{ts}' AS last_cursor,
'{BATCH_ID}' AS last_batch_id, current_timestamp() AS updated_at) s
ON t.source_table = s.source_table
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *
""")
def add_audit(df, name):
return (df.withColumn("_source_table", F.lit(f"{SRC_CAT}.{SRC_SCHEMA}.{name}"))
.withColumn("_batch_id", F.lit(BATCH_ID))
.withColumn("_ingested_at", F.lit(RUN_TS))
.withColumn("_ingest_date", F.to_date(F.lit(RUN_TS))))
def merge_insert_only(df, target, keys):
if not spark.catalog.tableExists(target):
df.write.format("delta").option("delta.enableChangeDataFeed", "true").saveAsTable(target)
return
df.createOrReplaceTempView("incoming")
on_clause = " AND ".join(f"t.{k} <=> s.{k}" for k in keys)
spark.sql(f"""
MERGE INTO {target} t USING incoming s ON {on_clause}
WHEN NOT MATCHED THEN INSERT *
""")
def dq_gate(df, name, keys):
total = df.count()
null_keys = df.filter(" OR ".join(f"{k} IS NULL" for k in keys)).count()
if null_keys:
raise ValueError(f"[{name}] {null_keys} NULL business keys; failing batch {BATCH_ID}")
return {"rows_read": total, "null_keys": null_keys}
# COMMAND ----------
# MAGIC %md ## Run
# COMMAND ----------
results = {}
for name, cfg in TABLES.items():
keys, cursor = cfg["key"], cfg["cursor"]
source = f"{SRC_CAT}.{SRC_SCHEMA}.{name}"
target = f"{BRONZE}.trakt_{name}"
wm = get_watermark(name)
src_df = spark.table(source) # federated read, no connector options
if wm is not None:
src_df = src_df.filter(F.col(cursor) >= F.lit(wm)) # should push down to Redshift
df = add_audit(src_df, name).cache()
stats = dq_gate(df, name, keys)
new_wm = df.agg(F.max(cursor)).first()[0] if stats["rows_read"] else None
merge_insert_only(df, target, keys)
set_watermark(name, new_wm)
stats.update({"target": target, "watermark_after": str(new_wm)})
results[name] = stats
df.unpersist()
# COMMAND ----------
# MAGIC %md ## Verify pushdown (look for the filter inside the federated scan, not above it)
# COMMAND ----------
spark.table(f"{SRC_CAT}.{SRC_SCHEMA}.watch_history") \
.filter("watched_at >= '2026-01-01'") \
.select("id", "watched_at") \
.explain("formatted")
# COMMAND ----------
dbutils.notebook.exit(json.dumps({"batch_id": BATCH_ID, "mode": MODE, "results": results}))

195
claude/bronze_redshift.py
View File

@@ -1,195 +0,0 @@
# Databricks notebook source
# MAGIC %md
# MAGIC # Bronze ingestion: Redshift -> Unity Catalog (Trakt.tv example)
# MAGIC Assumes Trakt personal data (watch history, ratings, watchlist) was already
# MAGIC exported from the Trakt API into Redshift schema `trakt_raw`.
# MAGIC Pattern: incremental read -> audit columns -> idempotent MERGE -> watermark -> DQ gate.
# COMMAND ----------
import json
import uuid
from datetime import datetime, timezone
from pyspark.sql import functions as F
# ---- Parameters (set per environment by the job / bundle) ----
dbutils.widgets.text("catalog", "dev")
dbutils.widgets.text("source_schema", "trakt_raw")
dbutils.widgets.dropdown("mode", "incremental", ["incremental", "full"])
CATALOG = dbutils.widgets.get("catalog")
SRC_SCHEMA = dbutils.widgets.get("source_schema")
MODE = dbutils.widgets.get("mode")
BATCH_ID = str(uuid.uuid4())
RUN_TS = datetime.now(timezone.utc)
BRONZE = f"{CATALOG}.bronze"
OPS = f"{CATALOG}.ops"
WM = f"{OPS}.ingestion_watermark"
# ---- Source table config: business key + incremental cursor column ----
TABLES = {
"watch_history": {"key": ["id"], "cursor": "watched_at"},
"ratings": {"key": ["trakt_id", "rated_at"], "cursor": "rated_at"},
"watchlist": {"key": ["trakt_id", "listed_at"], "cursor": "listed_at"},
}
# COMMAND ----------
# MAGIC %md ## Secrets and connection (no credentials in code)
# COMMAND ----------
JDBC_URL = dbutils.secrets.get("redshift", "jdbc_url") # jdbc:redshift://host:5439/db
RS_USER = dbutils.secrets.get("redshift", "user")
RS_PASSWORD = dbutils.secrets.get("redshift", "password")
IAM_ROLE = dbutils.secrets.get("redshift", "unload_iam_role")
TEMPDIR = dbutils.secrets.get("redshift", "tempdir") # s3://bucket/redshift-tmp/ (UC external location)
def read_redshift(query: str):
"""Redshift connector unloads via S3 (faster than plain JDBC for large reads)."""
return (
spark.read.format("redshift")
.option("url", JDBC_URL)
.option("user", RS_USER)
.option("password", RS_PASSWORD)
.option("query", query)
.option("tempdir", TEMPDIR)
.option("aws_iam_role", IAM_ROLE)
.load()
)
# COMMAND ----------
# MAGIC %md ## Unity Catalog objects (Spark SQL DDL)
# COMMAND ----------
spark.sql(f"CREATE SCHEMA IF NOT EXISTS {BRONZE} COMMENT 'Raw landed data, minimal transformation'")
spark.sql(f"CREATE SCHEMA IF NOT EXISTS {OPS} COMMENT 'Pipeline control tables'")
spark.sql(f"""
CREATE TABLE IF NOT EXISTS {WM} (
source_table STRING,
last_cursor TIMESTAMP,
last_batch_id STRING,
updated_at TIMESTAMP
) USING DELTA
""")
# Allow new source columns to flow into bronze instead of failing the job
spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", "true")
# COMMAND ----------
# MAGIC %md ## Helpers: watermark, audit columns, idempotent write, DQ
# COMMAND ----------
def get_watermark(name: str):
if MODE == "full":
return None
row = spark.sql(
f"SELECT last_cursor FROM {WM} WHERE source_table = '{name}'"
).first()
return row["last_cursor"] if row else None
def set_watermark(name: str, cursor_value):
if cursor_value is None:
return
ts = cursor_value.strftime("%Y-%m-%d %H:%M:%S.%f")
spark.sql(f"""
MERGE INTO {WM} t
USING (SELECT '{name}' AS source_table,
TIMESTAMP'{ts}' AS last_cursor,
'{BATCH_ID}' AS last_batch_id,
current_timestamp() AS updated_at) s
ON t.source_table = s.source_table
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *
""")
def add_audit_columns(df, name: str):
return (
df.withColumn("_source_table", F.lit(f"{SRC_SCHEMA}.{name}"))
.withColumn("_batch_id", F.lit(BATCH_ID))
.withColumn("_ingested_at", F.lit(RUN_TS))
.withColumn("_ingest_date", F.to_date(F.lit(RUN_TS)))
)
def merge_insert_only(df, target: str, keys: list):
"""Idempotent: re-running a batch (or overlapping windows) never duplicates rows."""
if not spark.catalog.tableExists(target):
(df.write.format("delta")
.option("delta.enableChangeDataFeed", "true")
.saveAsTable(target))
return
df.createOrReplaceTempView("incoming")
on_clause = " AND ".join(f"t.{k} <=> s.{k}" for k in keys) # null-safe equality
spark.sql(f"""
MERGE INTO {target} t
USING incoming s
ON {on_clause}
WHEN NOT MATCHED THEN INSERT *
""")
def dq_gate(df, name: str, keys: list):
total = df.count()
null_keys = df.filter(" OR ".join(f"{k} IS NULL" for k in keys)).count()
if null_keys > 0:
raise ValueError(f"[{name}] {null_keys} rows with NULL business key; failing batch {BATCH_ID}")
return {"rows_read": total, "null_keys": null_keys}
# COMMAND ----------
# MAGIC %md ## Run: one pass per source table
# COMMAND ----------
results = {}
for name, cfg in TABLES.items():
keys, cursor = cfg["key"], cfg["cursor"]
target = f"{BRONZE}.trakt_{name}"
wm = get_watermark(name)
query = f"SELECT * FROM {SRC_SCHEMA}.{name}"
if wm is not None:
# >= (not >) so late-arriving rows at the boundary aren't missed;
# the MERGE de-duplicates the overlap
query += f" WHERE {cursor} >= '{wm.strftime('%Y-%m-%d %H:%M:%S')}'"
df = read_redshift(query)
df = add_audit_columns(df, name).cache()
stats = dq_gate(df, name, keys)
new_wm = df.agg(F.max(cursor)).first()[0] if stats["rows_read"] > 0 else None
merge_insert_only(df, target, keys)
set_watermark(name, new_wm) # only after a successful write
stats.update({"target": target, "watermark_before": str(wm), "watermark_after": str(new_wm)})
results[name] = stats
df.unpersist()
# COMMAND ----------
# MAGIC %md ## Governance (Unity Catalog grants) and post-load checks
# COMMAND ----------
spark.sql(f"GRANT USE SCHEMA, SELECT ON SCHEMA {BRONZE} TO `data-engineers`")
spark.sql(f"""
SELECT _source_table, _batch_id, COUNT(*) AS rows_loaded
FROM {BRONZE}.trakt_watch_history
GROUP BY _source_table, _batch_id
ORDER BY MAX(_ingested_at) DESC
LIMIT 5
""").show(truncate=False)
# COMMAND ----------
dbutils.notebook.exit(json.dumps({"batch_id": BATCH_ID, "mode": MODE, "results": results}))

214
claude/spotify_silver.py
View File

@@ -1,214 +0,0 @@
# 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)

205
claude/trakt_models.py
View File

@@ -1,205 +0,0 @@
# Databricks notebook source
"""
Modeling examples on Trakt.tv data (silver layer).
ASSUMED SILVER TABLES (rename to match yours)
watch_history : id, trakt_id, type ('movie'|'episode'), title, watched_at, runtime_min
watched_movies : trakt_id, title, release_year, plays, last_watched_at, runtime_min, genres
watched_shows : trakt_id, title, release_year, plays, aired_episodes, n_seasons,
avg_runtime_min, first_watched_at, last_watched_at
watched_playback: id, trakt_id, type, progress (0-100), paused_at, runtime_min, release_year
NOTE: Trakt personal data is ONE user. The code works as is, but the business framing
("what drives engagement") is really "what drives my habits". The same models scale to a
multi-user streaming service by adding user_id and fitting on all users.
All "EXAMPLE OUTPUT" blocks below are ILLUSTRATIVE numbers, not real results.
"""
from pyspark.sql import functions as F, Window
from pyspark.ml.feature import VectorAssembler, StandardScaler
from pyspark.ml.regression import GeneralizedLinearRegression, AFTSurvivalRegression
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.clustering import KMeans
from pyspark.ml.evaluation import BinaryClassificationEvaluator, ClusteringEvaluator
SILVER = "dev.silver"
for t in ["watch_history", "watched_movies", "watched_shows", "watched_playback"]:
spark.table(f"{SILVER}.{t}").createOrReplaceTempView(t)
# COMMAND ----------
# =====================================================================
# 1. COUNT MODEL: "What drives how much I watch per day?"
# Framework: GLM (Poisson). Question types: demand forecasting, seasonality, trend.
# =====================================================================
start = spark.sql("SELECT MIN(CAST(watched_at AS DATE)) d FROM watch_history").first()["d"]
daily = spark.sql(f"""
WITH cal AS (
SELECT explode(sequence(DATE'{start}', current_date(), INTERVAL 1 DAY)) AS d
),
act AS (
SELECT CAST(watched_at AS DATE) AS d, COUNT(*) AS plays, SUM(runtime_min) AS minutes
FROM watch_history GROUP BY 1
)
SELECT cal.d,
COALESCE(act.plays, 0) AS plays,
COALESCE(act.minutes, 0) AS minutes,
CASE WHEN dayofweek(cal.d) IN (1, 7) THEN 1.0 ELSE 0.0 END AS is_weekend,
CASE WHEN month(cal.d) IN (11, 12, 1, 2) THEN 1.0 ELSE 0.0 END AS is_winter,
datediff(cal.d, DATE'{start}') / 365.0 AS trend_yrs
FROM cal LEFT JOIN act USING (d)
""").cache()
daily.createOrReplaceTempView("daily")
feat1 = ["is_weekend", "is_winter", "trend_yrs"]
d1 = VectorAssembler(inputCols=feat1, outputCol="features").transform(daily)
pois = GeneralizedLinearRegression(
family="poisson", link="log", labelCol="plays", featuresCol="features", regParam=0.0
).fit(d1)
print(dict(zip(feat1, pois.coefficients)), "intercept:", pois.intercept)
print("dispersion:", pois.summary.dispersion, "| AIC:", pois.summary.aic)
print("std errors:", pois.summary.coefficientStandardErrors())
# EXAMPLE OUTPUT (illustrative)
# intercept 0.05 -> baseline = exp(0.05) = 1.05 plays on a weekday, non-winter, day 0
# is_weekend 0.62 -> exp(0.62) = 1.86x : ~86% more plays on weekends
# is_winter 0.21 -> exp(0.21) = 1.23x : ~23% more plays Nov-Feb
# trend_yrs -0.18 -> exp(-0.18) = 0.84x : plays/day shrink ~16% per year
# dispersion 1.9 -> well above 1 = overdispersed (binge days). Std errors are too
# optimistic: move to negative binomial / quasi-Poisson.
# Model "constants": the intercept is the baseline rate; dispersion is the variance
# multiplier; AIC lets you compare this against a version with extra features.
# COMMAND ----------
# =====================================================================
# 2. CLASSIFICATION: "Will I finish what I start?"
# Framework: GLM (logistic). Question types: completion/abandonment drivers,
# content-length strategy, scorecards.
# =====================================================================
pb = spark.sql("""
SELECT paused_at,
CAST(progress >= 90 AS DOUBLE) AS completed,
runtime_min,
CASE WHEN type = 'movie' THEN 1.0 ELSE 0.0 END AS is_movie,
CAST(hour(paused_at) AS DOUBLE) AS hour_of_day,
CASE WHEN dayofweek(paused_at) IN (1, 7) THEN 1.0 ELSE 0.0 END AS is_weekend,
CAST(release_year AS DOUBLE) AS release_year
FROM watched_playback
WHERE runtime_min IS NOT NULL
""")
# time-based split (never random for behavior over time)
pb = pb.withColumn("pr", F.percent_rank().over(Window.orderBy("paused_at")))
feat2 = ["runtime_min", "is_movie", "hour_of_day", "is_weekend", "release_year"]
asm2 = VectorAssembler(inputCols=feat2, outputCol="features")
train2 = asm2.transform(pb.filter("pr < 0.8"))
test2 = asm2.transform(pb.filter("pr >= 0.8"))
logit = LogisticRegression(
labelCol="completed", featuresCol="features",
regParam=0.01, elasticNetParam=0.5, # elastic net: some shrinkage, some zeroing
standardization=True # coefficients reported on ORIGINAL scale
).fit(train2)
auc = BinaryClassificationEvaluator(labelCol="completed", metricName="areaUnderROC") \
.evaluate(logit.transform(test2))
print(dict(zip(feat2, logit.coefficients)), "intercept:", logit.intercept, "| test AUC:", auc)
# EXAMPLE OUTPUT (illustrative)
# intercept 2.10 -> baseline log-odds; p = 1/(1+e^-2.1) = 0.89 completion at zeros
# runtime_min -0.011 -> per +30 min: exp(-0.33) = 0.72 odds ratio (28% lower odds)
# is_movie -0.45 -> exp(-0.45) = 0.64 : movies completed less often than episodes
# hour_of_day 0.06 -> exp(0.06) = 1.06 : each later hour +6% odds (late = committed)
# is_weekend 0.30 -> exp(0.30) = 1.35 : weekends +35% odds
# release_year -0.004 -> ~0 : elastic net shrinks it; effectively no signal
# test AUC 0.74 -> decent ranking ability; 0.5 is a coin flip
# Model "constants": regParam / elasticNetParam (regularization strength / L1 mix),
# 0.5 default decision threshold (tune it if completions and abandonments are imbalanced).
# COMMAND ----------
# =====================================================================
# 3. SURVIVAL: "How long until I abandon a show?"
# Framework: Survival analysis (AFT). Question types: churn timing, retention curves,
# which content holds attention longest. Handles shows still in progress (censoring).
# =====================================================================
shows = spark.sql("""
SELECT trakt_id, title,
GREATEST(CAST(datediff(last_watched_at, first_watched_at) AS DOUBLE), 1.0) AS days_active,
CASE
WHEN plays < aired_episodes
AND last_watched_at < date_sub(current_date(), 60) THEN 1.0 -- abandoned = event
ELSE 0.0 -- finished/active = censored
END AS event,
CAST(n_seasons AS DOUBLE) AS n_seasons,
CAST(avg_runtime_min AS DOUBLE) AS avg_runtime_min,
CAST(release_year AS DOUBLE) AS release_year
FROM watched_shows
WHERE first_watched_at IS NOT NULL AND last_watched_at IS NOT NULL
""")
feat3 = ["n_seasons", "avg_runtime_min", "release_year"]
d3 = VectorAssembler(inputCols=feat3, outputCol="features").transform(shows)
aft = AFTSurvivalRegression(
labelCol="days_active", censorCol="event", featuresCol="features",
quantileProbabilities=[0.25, 0.5, 0.75], quantilesCol="quantiles"
).fit(d3)
print(dict(zip(feat3, aft.coefficients)), "intercept:", aft.intercept, "scale:", aft.scale)
aft.transform(d3).select("title", "days_active", "event", "quantiles").show(5, truncate=False)
# EXAMPLE OUTPUT (illustrative)
# intercept 4.20 -> baseline time scale exp(4.2) = ~67 days
# n_seasons 0.10 -> exp(0.10) = 1.11x : each extra season stretches time-to-abandon 11%
# avg_runtime_min -0.015-> per +30 min: exp(-0.45) = 0.64x : hour-long shows abandoned ~36% sooner
# release_year 0.00 -> no effect
# scale (sigma) 1.30 -> Weibull shape = 1/1.30 = 0.77 (<1): hazard FALLS over time,
# i.e. most abandonment happens early; survivors tend to stay.
# quantiles [9, 31, 88] -> predicted days until 25% / 50% / 75% chance of abandoning.
# Model "constants": intercept (baseline time), scale (shape of hazard over time),
# and the quantile probabilities you ask for.
# COMMAND ----------
# =====================================================================
# 4. SEGMENTATION: "What kinds of viewing days do I have?"
# Framework: Clustering (K-means). Question types: behavioral segments,
# personas, targeting, anomaly baselines.
# =====================================================================
days = spark.sql("""
SELECT CAST(watched_at AS DATE) AS d,
CAST(COUNT(*) AS DOUBLE) AS plays,
CAST(SUM(runtime_min) AS DOUBLE) AS minutes,
CAST(AVG(hour(watched_at)) AS DOUBLE) AS avg_hour,
AVG(CASE WHEN type = 'episode' THEN 1.0 ELSE 0.0 END) AS episode_share
FROM watch_history GROUP BY 1
""")
feat4 = ["plays", "minutes", "avg_hour", "episode_share"]
d4 = VectorAssembler(inputCols=feat4, outputCol="raw").transform(days)
scaler = StandardScaler(inputCol="raw", outputCol="features", withMean=True, withStd=True).fit(d4)
d4s = scaler.transform(d4)
best = None
for k in range(2, 8):
m = KMeans(k=k, seed=42, featuresCol="features").fit(d4s)
sil = ClusteringEvaluator(featuresCol="features").evaluate(m.transform(d4s))
print(f"k={k} silhouette={sil:.3f} WSSSE={m.summary.trainingCost:.1f}")
if best is None or sil > best[1]:
best = (m, sil, k)
model, sil, k = best
mu, sd = scaler.mean.toArray(), scaler.std.toArray()
for i, c in enumerate(model.clusterCenters()):
print(f"cluster {i}:", dict(zip(feat4, (c * sd + mu).round(2)))) # back to real units
# EXAMPLE OUTPUT (illustrative)
# k chosen = 4, silhouette = 0.52 (>0.5 = reasonably separated), WSSSE falls with k so
# use the elbow plus silhouette, not WSSSE alone.
# cluster 0: plays 1.2, minutes 45, avg_hour 21.0, episode_share 1.0 -> "weeknight episode"
# cluster 1: plays 1.0, minutes 125, avg_hour 20.5, episode_share 0.0 -> "movie night"
# cluster 2: plays 5.5, minutes 240, avg_hour 15.0, episode_share 0.95 -> "weekend binge"
# cluster 3: plays 2.0, minutes 95, avg_hour 0.8, episode_share 0.7 -> "late-night wind-down"
# Model "constants": k, the cluster centers (the personas), and the scaler's mean/std
# (needed to score new days the same way).

206
claude/trakt_silver.py
View File

@@ -1,206 +0,0 @@
# Databricks notebook source
"""
Bronze -> Silver for ONE user's Trakt export, where watch HISTORY is the main table.
ASSUMED BRONZE (nested JSON landed as-is; if you load from Redshift the columns are already
flat, so skip Step 1 and start from your flat table):
{cat}.bronze.trakt_history id, watched_at, action, type, movie{title,year,ids{trakt,imdb,tmdb}},
show{...same...}, episode{season,number,title,ids{trakt}}
{cat}.bronze.trakt_watched_movies plays, last_watched_at, movie{...}
{cat}.bronze.trakt_watched_shows plays, last_watched_at, reset_at, show{...}, seasons[{number, episodes[{number, plays}]}]
{cat}.bronze.trakt_playback id, progress, paused_at, type, movie{...}/episode{...}/show{...}
{cat}.bronze.trakt_title_metadata OPTIONAL enrichment (extended=full): title_kind ('movie'|'show'),
title_id, runtime_min, genres (array<string>), aired_episodes
Trakt-specific traps handled below:
* Movie ids and show ids are DIFFERENT id spaces -> always key on (title_kind, title_id).
* History has no runtime or duration -> runtime comes from metadata, or stays NULL.
* watched_at can be backdated/bulk-imported -> flag unreliable timestamps, don't use them
for time-of-day or daily-rate models.
* Rewatches are legitimate separate rows -> dedupe on history id only, never on title.
"""
dbutils.widgets.text("catalog", "dev")
dbutils.widgets.text("home_tz", "Europe/London")
CAT = dbutils.widgets.get("catalog")
TZ = dbutils.widgets.get("home_tz")
spark.sql(f"CREATE SCHEMA IF NOT EXISTS {CAT}.silver")
for t in ["trakt_history", "trakt_watched_movies", "trakt_watched_shows", "trakt_playback"]:
spark.table(f"{CAT}.bronze.{t}").createOrReplaceTempView(f"b_{t}")
has_meta = spark.catalog.tableExists(f"{CAT}.bronze.trakt_title_metadata")
if has_meta:
spark.table(f"{CAT}.bronze.trakt_title_metadata").createOrReplaceTempView("b_meta")
else: # empty stand-in so the join still works
spark.sql("""CREATE OR REPLACE TEMP VIEW b_meta AS
SELECT CAST(NULL AS STRING) title_kind, CAST(NULL AS BIGINT) title_id,
CAST(NULL AS INT) runtime_min, CAST(NULL AS ARRAY<STRING>) genres,
CAST(NULL AS INT) aired_episodes WHERE 1 = 0""")
# COMMAND ----------
# MAGIC %md ## Step 1: flatten nested JSON into one event row per history entry
# COMMAND ----------
spark.sql("""
CREATE OR REPLACE TEMP VIEW h_flat AS
SELECT
CAST(id AS BIGINT) AS history_id,
try_to_timestamp(watched_at) AS watched_at_utc, -- NULL if a placeholder string
watched_at AS watched_at_raw,
lower(action) AS action, -- watch | scrobble | checkin
lower(type) AS media_type, -- movie | episode
CASE WHEN lower(type) = 'movie' THEN 'movie' ELSE 'show' END AS title_kind,
COALESCE(movie.ids.trakt, show.ids.trakt) AS title_id, -- movie id OR show id
episode.ids.trakt AS episode_id,
COALESCE(movie.title, show.title) AS title,
COALESCE(movie.year, show.year) AS release_year,
episode.season AS season,
episode.number AS episode_number,
COALESCE(movie.ids.imdb, show.ids.imdb) AS imdb_id,
COALESCE(movie.ids.tmdb, show.ids.tmdb) AS tmdb_id,
_source_file
FROM b_trakt_history
WHERE id IS NOT NULL
""")
# COMMAND ----------
# MAGIC %md ## Step 2: dedupe on history id, flag unreliable timestamps, enrich, localize
# COMMAND ----------
spark.sql(f"""
CREATE OR REPLACE TEMP VIEW h_clean AS
WITH dedup AS (
SELECT * EXCEPT (rn) FROM (
SELECT *, row_number() OVER (PARTITION BY history_id ORDER BY _source_file DESC) AS rn
FROM h_flat)
WHERE rn = 1
),
flagged AS (
SELECT *,
CASE WHEN media_type = 'movie' THEN concat('m:', title_id)
ELSE concat('e:', episode_id) END AS content_key,
watched_at_utc IS NULL AS is_placeholder_ts,
action IN ('scrobble', 'checkin') AS is_live_action,
date_format(watched_at_utc, 'HH:mm:ss') = '00:00:00' AS is_midnight_ts,
COUNT(*) OVER (PARTITION BY date_trunc('minute', watched_at_utc)) >= 10 AS is_bulk_import
FROM dedup
)
SELECT f.*,
from_utc_timestamp(watched_at_utc, '{TZ}') AS watched_at_local,
m.runtime_min, m.genres, m.aired_episodes,
-- only trust clock-of-day features when the event looks like a real-time watch
(is_live_action AND NOT is_midnight_ts AND NOT is_bulk_import
AND NOT is_placeholder_ts) AS is_reliable_time
FROM flagged f
LEFT JOIN b_meta m ON m.title_kind = f.title_kind AND m.title_id = f.title_id
WHERE NOT f.is_placeholder_ts -- keep placeholders in a quarantine table in real pipelines
""")
# COMMAND ----------
# MAGIC %md ## Step 3: rewatch, show-sequence and session features
# COMMAND ----------
spark.sql(f"""
CREATE OR REPLACE TABLE {CAT}.silver.trakt_watch_events AS
WITH seq AS (
SELECT *,
row_number() OVER (PARTITION BY content_key ORDER BY watched_at_utc) AS play_number,
lag(watched_at_utc) OVER (PARTITION BY title_kind, title_id ORDER BY watched_at_utc) AS prev_same_title_utc,
lag(season) OVER (PARTITION BY title_kind, title_id ORDER BY watched_at_utc) AS prev_season,
lag(episode_number) OVER (PARTITION BY title_kind, title_id ORDER BY watched_at_utc) AS prev_episode,
lag(watched_at_utc) OVER (ORDER BY watched_at_utc) AS prev_any_utc
FROM h_clean
),
sess AS (
SELECT *,
SUM(CASE WHEN prev_any_utc IS NULL
OR watched_at_utc > prev_any_utc + INTERVAL 4 HOURS THEN 1 ELSE 0 END)
OVER (ORDER BY watched_at_utc ROWS UNBOUNDED PRECEDING) AS session_seq
FROM seq
)
SELECT *,
play_number > 1 AS is_rewatch,
datediff(watched_at_utc, prev_same_title_utc) AS days_since_prev_same_title,
media_type = 'episode' AND (
(season = prev_season AND episode_number = prev_episode + 1)
OR (episode_number = 1 AND season = prev_season + 1)) AS is_next_in_order,
count(*) OVER (PARTITION BY session_seq) AS plays_in_session,
hour(watched_at_local) AS hour_local,
dayofweek(watched_at_local) IN (1, 7) AS is_weekend,
CAST(watched_at_local AS DATE) AS watch_date_local
FROM sess
""")
# COMMAND ----------
# MAGIC %md ## Step 4: small tables (playback, titles) and a daily panel for count models
# COMMAND ----------
spark.sql(f"""
CREATE OR REPLACE TABLE {CAT}.silver.trakt_playback AS
SELECT CAST(id AS BIGINT) AS playback_id,
lower(type) AS media_type,
CAST(progress AS DOUBLE) AS progress_pct,
try_to_timestamp(paused_at) AS paused_at_utc,
COALESCE(movie.ids.trakt, show.ids.trakt) AS title_id,
episode.ids.trakt AS episode_id,
progress >= 90 AS is_completed_proxy
FROM b_trakt_playback
""")
spark.sql(f"""
CREATE OR REPLACE TABLE {CAT}.silver.trakt_titles AS
SELECT title_kind, title_id, MAX(title) AS title, MAX(release_year) AS release_year,
MAX(imdb_id) AS imdb_id, MAX(tmdb_id) AS tmdb_id,
MAX(runtime_min) AS runtime_min, MAX(genres) AS genres, MAX(aired_episodes) AS aired_episodes
FROM {CAT}.silver.trakt_watch_events GROUP BY title_kind, title_id
""")
# Daily panel: full calendar, zeros included, bulk imports excluded from the behavioral count
first_day = spark.sql(f"SELECT MIN(watch_date_local) d FROM {CAT}.silver.trakt_watch_events").first()["d"]
spark.sql(f"""
CREATE OR REPLACE TABLE {CAT}.silver.trakt_daily AS
WITH cal AS (SELECT explode(sequence(DATE'{first_day}', current_date(), INTERVAL 1 DAY)) AS d),
act AS (
SELECT watch_date_local AS d, COUNT(*) AS plays_live, SUM(runtime_min) AS minutes_live
FROM {CAT}.silver.trakt_watch_events
WHERE NOT is_bulk_import AND NOT is_midnight_ts
GROUP BY 1
)
SELECT cal.d, COALESCE(act.plays_live, 0) AS plays, COALESCE(act.minutes_live, 0) AS minutes
FROM cal LEFT JOIN act ON act.d = cal.d
""")
# COMMAND ----------
# MAGIC %md ## Step 5: quality checks and reconciliation against the aggregate files
# COMMAND ----------
spark.sql(f"""
SELECT COUNT(*) AS events,
MIN(watched_at_utc) AS first_event, MAX(watched_at_utc) AS last_event,
ROUND(AVG(CAST(is_bulk_import AS INT)), 3) AS bulk_share,
ROUND(AVG(CAST(is_midnight_ts AS INT)), 3) AS midnight_share,
ROUND(AVG(CAST(is_reliable_time AS INT)), 3) AS reliable_time_share,
ROUND(AVG(CAST(runtime_min IS NULL AS INT)), 3) AS missing_runtime_share
FROM {CAT}.silver.trakt_watch_events
""").show(truncate=False)
recon = spark.sql(f"""
SELECT 'movies' AS kind,
(SELECT COUNT(*) FROM {CAT}.silver.trakt_watch_events WHERE media_type = 'movie') AS history_plays,
(SELECT SUM(plays) FROM b_trakt_watched_movies) AS aggregate_plays
UNION ALL
SELECT 'episodes',
(SELECT COUNT(*) FROM {CAT}.silver.trakt_watch_events WHERE media_type = 'episode'),
(SELECT SUM(e.plays) FROM b_trakt_watched_shows
LATERAL VIEW explode(seasons) st AS s
LATERAL VIEW explode(s.episodes) et AS e)
""")
recon.withColumn("diff_pct", F.round((F.col("history_plays") - F.col("aggregate_plays"))
/ F.col("aggregate_plays") * 100, 2)).show()
# Small differences are normal (show resets via reset_at, removed history).
# Large ones mean the history export is incomplete: stop and investigate before modeling.

0
claude/etl_spark.md → etl_spark.md
View File

155
spotify.ipynb → spotify_bronze_silver.ipynb
View File

@@ -2,7 +2,7 @@
"cells": [
{
"cell_type": "code",
"execution_count": 8,
"execution_count": 14,
"id": "e9bf1d26",
"metadata": {},
"outputs": [],
@@ -17,7 +17,7 @@
},
{
"cell_type": "code",
"execution_count": 9,
"execution_count": null,
"id": "11bb5dc2",
"metadata": {},
"outputs": [],
@@ -27,18 +27,21 @@
"BASE = Path.cwd() / \"ingest\"\n",
"\n",
"def start_spark(app_name):\n",
" \"\"\"Initialize and return a SparkSession with the PostgreSQL JDBC driver.\"\"\"\n",
" jar_path = str(Path.cwd() / \"postgresql.jar\")\n",
" spark = SparkSession.builder.appName(app_name).config(\"spark.jars\", jar_path).config(\"spark.driver.extraClassPath\", jar_path).getOrCreate()\n",
" spark.sparkContext.setLogLevel(\"ERROR\")\n",
" return spark\n",
"\n",
"def music_source_paths():\n",
" \"\"\"Return sorted list of Spotify listening history JSON file paths.\"\"\"\n",
" return sorted(glob.glob(str(BASE / \"*_sesh\" / \"*.json\")))\n",
"\n",
"def video_source_paths():\n",
" \"\"\"Return sorted list of Trakt video watching history JSON file paths.\"\"\"\n",
" return sorted(glob.glob(str(BASE / \"trakt_*\" / \"*.json\")))\n",
"\n",
"def read_json(spark, paths):\n",
" \"\"\"Read a list of JSON files into a multi-line Spark DataFrame.\"\"\"\n",
" if not paths:\n",
" raise FileNotFoundError(\"No JSON files matched the pattern\")\n",
" return (\n",
@@ -63,6 +66,7 @@
"source": [
" ## Transform ##\n",
"def null_counts_by_column(df):\n",
" \"\"\"Count nulls in each column of the DataFrame.\"\"\"\n",
" null_counts = df.select([\n",
" F.count(F.when(F.col(c).isNull(), c)).alias(c)\n",
" for c in df.columns\n",
@@ -70,9 +74,11 @@
" return null_counts\n",
"\n",
"def metadata_df(df):\n",
" \"\"\"Rename verbose album/artist/track metadata columns to short names.\"\"\"\n",
" return df.withColumnRenamed(\"master_metadata_album_album_name\", \"album_name\").withColumnRenamed(\"master_metadata_album_artist_name\", \"artist_name\").withColumnRenamed(\"master_metadata_track_name\", \"track_name\")\n",
"\n",
"def rename_columns(df, rename_map, drop_columns):\n",
" \"\"\"Drop specified columns and rename remaining ones per the rename map.\"\"\"\n",
" missing_rename = set(rename_map) - set(df.columns)\n",
" missing_dropped = set(drop_columns) - set(df.columns)\n",
"\n",
@@ -86,31 +92,39 @@
" return df.toDF(*[rename_map.get(c, c) for c in df.columns])\n",
"\n",
"def track_df(df):\n",
" \"\"\"Filter to rows where track_name is not null.\"\"\"\n",
" return df.where(df[\"track_name\"].isNotNull())\n",
"\n",
"def parse_timestamps(df: DataFrame, columns) -> DataFrame:\n",
" \"\"\"Cast each column in the list to timestamp type.\"\"\"\n",
" for column in columns:\n",
" df = df.withColumn(column, F.to_timestamp(column))\n",
" return df\n",
"\n",
"def parse_booleans(df: DataFrame, columns) -> DataFrame:\n",
" \"\"\"Cast each column in the list to string type.\"\"\"\n",
" df = df.select([F.col(c).cast(\"string\") for c in columns])\n",
" return df\n",
"\n",
"def add_minutes_played(df: DataFrame, src, dest) -> DataFrame:\n",
" \"\"\"Convert milliseconds played to minutes and round to 2 decimals.\"\"\"\n",
" return (df.withColumn(dest, F.round(F.col(src) / 60000, 2)))\n",
"\n",
"def add_time_parts(df: DataFrame, ts_col) -> DataFrame:\n",
" \"\"\"Add time/date/day-of-week/hour parts derived from the timestamp columns.\"\"\"\n",
" return (df\n",
" .withColumn(\"ts\", F.to_timestamp(\"ts\"))\n",
" .withColumn(\"offline_timestamped\", F.to_timestamp(\"offline_timestamp\"))\n",
" .withColumn('time', F.date_format('ts', 'HH:mm:ss'))\n",
" .withColumn('date', F.date_format('ts', 'yyyy-MM-dd'))\n",
" .withColumn('day_of_week', F.day('ts'))\n",
" .withColumn('hour', F.hour('ts'))\n",
" .withColumn('offline_time', F.date_format('offline_timestamped', 'HH:mm:ss'))\n",
" .withColumn('offline_date', F.date_format('offline_timestamped', 'yyyy-MM-dd')) \n",
" )\n",
"\n",
"def clean_track_df(df: DataFrame) -> DataFrame:\n",
" \"\"\"Apply the full transform pipeline: rename, filter tracks, add minutes and time parts.\"\"\"\n",
" return(df\n",
" .transform(rename_columns, SPOTIFY_RENAMES, MUSIC_DROP_COLUMNS)\n",
" .transform(track_df)\n",
@@ -123,7 +137,7 @@
},
{
"cell_type": "code",
"execution_count": 11,
"execution_count": 17,
"id": "817475cf",
"metadata": {},
"outputs": [],
@@ -155,15 +169,21 @@
" }\n",
" return url, properties\n",
"\n",
"def load_data(df, table_name):\n",
"def jdbc_write(df, table_name):\n",
" \"\"\"Load the data into the database.\"\"\"\n",
" url, properties = get_jdbc_config()\n",
" df.write.jdbc(url=url, table=table_name, mode=\"overwrite\", properties=properties)"
" df.write.jdbc(url=url, table=table_name, mode=\"overwrite\", properties=properties)\n",
"\n",
"def databricks_write(df, table_name):\n",
" \"\"\"Load the data into bronze catalog.\"\"\"\n",
" catalog = \"workspace\"\n",
" schema = \"default\"\n",
" return df.write.format(\"delta\").mode(\"overwrite\").saveAsTable(f\"{catalog}.{schema}.{table_name}\")"
]
},
{
"cell_type": "code",
"execution_count": 12,
"execution_count": 18,
"id": "c609291c",
"metadata": {},
"outputs": [],
@@ -185,7 +205,7 @@
},
{
"cell_type": "code",
"execution_count": 13,
"execution_count": 19,
"id": "4cc14bd0",
"metadata": {},
"outputs": [
@@ -200,6 +220,22 @@
"name": "stdout",
"output_type": "stream",
"text": [
"+----------------------------+--------------+\n",
"|reason_end |skipped_values|\n",
"+----------------------------+--------------+\n",
"|NULL |[] |\n",
"|backbtn |[true] |\n",
"|endplay |[true] |\n",
"|fwdbtn |[true] |\n",
"|logout |[false] |\n",
"|remote |[false] |\n",
"|trackdone |[false] |\n",
"|trackerror |[false] |\n",
"|unexpected-exit |[false] |\n",
"|unexpected-exit-while-paused|[false] |\n",
"|unknown |[false] |\n",
"+----------------------------+--------------+\n",
"\n",
" Raw DF Rows: 105477\n",
" Transformed DF Rows: 102626\n",
" Row difference is 2851\n"
@@ -211,16 +247,44 @@
"def main():\n",
" app_name = \"ml_media\"\n",
" spark = start_spark(app_name)\n",
" spark.sparkContext.setLogLevel(\"ERROR\")\n",
"\n",
" raw_table_name = \"raw_spotify_data\"\n",
" bronze_table_name = \"bronze_spotify_data\"\n",
" silver_table_name = \"silver_spotify_data\"\n",
" \n",
" table_name = \"spotify_data\"\n",
" raw_df = read_json(spark, music_source_paths())\n",
"\n",
" jdbc_write(raw_df, raw_table_name)\n",
"\n",
" transformed_df = clean_track_df(raw_df)\n",
" load_data(transformed_df, table_name)\n",
" \n",
" jdbc_write(transformed_df, bronze_table_name)\n",
" # transformed_df.show(5)\n",
" # skip_reason_df = transformed_df.groupBy(\"skipped\", \"reason_end\").count().orderBy(F.desc(\"reason_end\")).show()\n",
" # reason_df = transformed_df.groupBy(\"reason_end\").orderBy(F.desc(\"count\")).show(5)\n",
" # skip_df = transformed_df.groupBy(\"skipped\").count().orderBy(F.desc(\"count\")).show(5)\n",
" # skip_join_bool = transformed_df.select(\"reason_end\", \"skipped\").distinct().orderBy(\"reason_end\").show(100)\n",
" skip_reason_bool = (transformed_df.groupBy(\"reason_end\")\n",
" .agg(F.collect_set(\"skipped\").alias(\"skipped_values\"))\n",
" .orderBy(\"reason_end\")\n",
" .show(truncate=False))\n",
" #join(transformed_df.F.col(\"reason_end\"))\n",
"\n",
" # missing_boolean = set(rename_map) - set(df.columns)\n",
"\n",
"\n",
"# Assuming 'df' is your DataFrame and 'timestamp_column' is the column with timestamps\n",
" # df = df.withColumn(\"time_of_day\", \n",
" # F.when((F.hour(\"timestamp_column\") >= 5) & (F.hour(\"timestamp_column\") < 12), \"morning\")\n",
" # .when((F.hour(\"timestamp_column\") >= 12) & (F.hour(\"timestamp_column\") < 18), \"afternoon\")\n",
" # .when((F.hour(\"timestamp_column\") >= 18) & (F.hour(\"timestamp_column\") < 22), \"evening\")\n",
" # .otherwise(\"night\"))\n",
"\n",
" print(f\" Raw DF Rows: {raw_df.count()}\")\n",
" print(f\" Transformed DF Rows: {transformed_df.count()}\")\n",
" print(f\" Row difference is {raw_df.count() - transformed_df.count()}\") \n",
" spark.stop\n",
" spark.stop()\n",
"if __name__ == \"__main__\":\n",
" main()\n",
"\n",
@@ -230,38 +294,55 @@
},
{
"cell_type": "code",
"execution_count": 14,
"execution_count": 20,
"id": "35878e16",
"metadata": {},
"outputs": [],
"source": [
"## Databricks\n",
"\n",
" ## MAIN ## \n",
"\n",
"# app_name = \"ml_media\"\n",
"# spark.log.level(\"ERROR\")\n",
"\n",
"# raw_table_name = \"raw_spotify_data\"\n",
"# bronze_table_name = \"bronze_spotify_data\"\n",
"# silver_table_name = \"silver_spotify_data\"\n",
"# raw_df = read_json(spark, music_source_paths())\n",
"# databricks_write(raw_df, raw_table_name)\n",
"# transformed_df = clean_track_df(raw_df)\n",
"\n",
"# databricks_write(transformed_df, bronze_table_name)\n",
"\n",
"# skip_reason_bool = (transformed_df.groupBy(\"reason_end\")\n",
"# .agg(F.collect_set(\"skipped\").alias(\"skipped_values\"))\n",
"# .orderBy(\"reason_end\")\n",
"# .show(truncate=False))\n"
]
},
{
"cell_type": "code",
"execution_count": 21,
"id": "faa6a8cb",
"metadata": {},
"outputs": [
{
"ename": "NameError",
"evalue": "name 'music_df' is not defined",
"output_type": "error",
"traceback": [
"\u001b[31m---------------------------------------------------------------------------\u001b[39m",
"\u001b[31mNameError\u001b[39m Traceback (most recent call last)",
"\u001b[36mCell\u001b[39m\u001b[36m \u001b[39m\u001b[32mIn[14]\u001b[39m\u001b[32m, line 2\u001b[39m\n\u001b[32m 1\u001b[39m \u001b[38;5;66;03m## Potential Test Help\u001b[39;00m\n\u001b[32m----> \u001b[39m\u001b[32m2\u001b[39m print(metadata_df(music_df).rdd.getNumPartitions()) \u001b[38;5;66;03m# Partitions for Pyspark\u001b[39;00m\n\u001b[32m 3\u001b[39m print(len(metadata_df(music_df).columns)) \u001b[38;5;66;03m# # of columns in JSON\u001b[39;00m\n\u001b[32m 4\u001b[39m null_counts = null_counts_by_column(metadata_df(music_df)) \u001b[38;5;66;03m# Null counts by column\u001b[39;00m\n\u001b[32m 5\u001b[39m null_counts.show(vertical=\u001b[38;5;28;01mTrue\u001b[39;00m) \u001b[38;5;66;03m# Print Null counts by column_df\u001b[39;00m\n",
"\u001b[31mNameError\u001b[39m: name 'music_df' is not defined"
]
}
],
"outputs": [],
"source": [
"## Potential Test Help\n",
"print(metadata_df(music_df).rdd.getNumPartitions()) # Partitions for Pyspark\n",
"print(len(metadata_df(music_df).columns)) # # of columns in JSON\n",
"null_counts = null_counts_by_column(metadata_df(music_df)) # Null counts by column\n",
"null_counts.show(vertical=True) # Print Null counts by column_df\n",
"total_rows = music_df.count\n",
"# ## Potential Test Help\n",
"# print(metadata_df(music_df).rdd.getNumPartitions()) # Partitions for Pyspark\n",
"# print(len(metadata_df(music_df).columns)) # # of columns in JSON\n",
"# null_counts = null_counts_by_column(metadata_df(music_df)) # Null counts by column\n",
"# null_counts.show(vertical=True) # Print Null counts by column_df\n",
"# total_rows = music_df.count\n",
"\n",
"df.count()-track_df.count() ## assert the difference in rows for tracks is equal to the nulls count\n",
" ## same can be done for podcasts\n",
"# df.count()-track_df.count() ## assert the difference in rows for tracks is equal to the nulls count\n",
"# ## same can be done for podcasts\n",
"\n",
"df.show(5)\n",
"print(f\"Original Rows {df.count()}\")\n",
"# df.show(5)\n",
"# print(f\"Original Rows {df.count()}\")\n",
"\n",
"track_df.show(5)\n",
"print(f\"Track Rows {track_df.count()}\")\n",
"# track_df.show(5)\n",
"# print(f\"Track Rows {track_df.count()}\")\n",
"\n"
]
}

528
spotify_silver_gold.ipynb Normal file
View File

@@ -0,0 +1,528 @@
{
"cells": [
{
"cell_type": "code",
"execution_count": 13,
"id": "09a05cc1",
"metadata": {},
"outputs": [],
"source": [
"from pyspark.sql import SparkSession, DataFrame\n",
"from pyspark.sql import functions as F\n",
"from pathlib import Path\n",
"import glob\n",
"import os\n",
"from dotenv import load_dotenv, find_dotenv"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "2c1aeb30",
"metadata": {},
"outputs": [],
"source": [
" ## Extract (Load in) ##\n",
"def start_spark(app_name):\n",
" \"\"\"Initialize and return a SparkSession with the PostgreSQL JDBC driver.\"\"\"\n",
" jar_path = str(Path.cwd() / \"postgresql.jar\")\n",
" spark = SparkSession.builder.appName(app_name).config(\"spark.jars\", jar_path).config(\"spark.driver.extraClassPath\", jar_path).getOrCreate()\n",
" return spark\n",
"\n",
"def get_jdbc_config():\n",
" \"\"\"Load env vars and return a Postgres URL\"\"\"\n",
" load_dotenv(find_dotenv())\n",
" \n",
" username=os.getenv(\"DATABASE_USERNAME\")\n",
" password=os.getenv(\"DATABASE_PASSWORD\")\n",
" host=os.getenv(\"DATABASE_IP\")\n",
" port=int(os.getenv(\"DATABASE_PORT\", 5432))\n",
" database=os.getenv(\"DATABASE_NAME\")\n",
" \n",
" missing = [k for k, v in {\n",
" \"DATABASE_USERNAME\": username,\n",
" \"DATABASE_PASSWORD\": password,\n",
" \"DATABASE_IP\": host,\n",
" \"DATABASE_NAME\": database,}.items() if not v]\n",
" if missing:\n",
" raise ValueError(f\"Missing env vars: {missing}\")\n",
"\n",
" url = f\"jdbc:postgresql://{host}:{port}/{database}\"\n",
" properties = {\n",
" \"user\": username,\n",
" \"password\": password,\n",
" \"driver\": \"org.postgresql.Driver\"\n",
" }\n",
" return url, properties\n",
"\n",
"def read_jdbc(spark, table_name):\n",
" \"\"\"Read a table from Postgres via JDBC into a Spark DataFrame.\"\"\"\n",
" url, properties = get_jdbc_config()\n",
" return spark.read.jdbc(url=url, table=table_name, properties=properties)\n",
"\n",
"def databricks_read(df, table_name):\n",
" \"\"\"Read the data from catalog.\"\"\"\n",
" catalog = \"workspace\"\n",
" schema = \"default\"\n",
" path = f\"{catalog}.{schema}.{table_name}\"\n",
" return df.read.table(path)"
]
},
{
"cell_type": "code",
"execution_count": 15,
"id": "9631ba07",
"metadata": {},
"outputs": [],
"source": [
" ## Transform ##"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "f7673c72",
"metadata": {},
"outputs": [
{
"name": "stderr",
"output_type": "stream",
"text": [
"[Stage 0:> (0 + 1) / 1]\r"
]
},
{
"name": "stdout",
"output_type": "stream",
"text": [
"+--------------------+--------------------+-----+\n",
"| track_name| username|count|\n",
"+--------------------+--------------------+-----+\n",
"| Mujeriego| csm3-us| 93|\n",
"| Angelito| csm3-us| 16|\n",
"|Navidad en Mi Barrio| csm3-us| 1|\n",
"| Como Perro Y Gato| csm3-us| 1|\n",
"| The Taste of Ink|fm30ajetkgrrkbvae...| 2|\n",
"| Hasta el Amanecer| csm3-us| 4|\n",
"| Tu Foto| csm3-us| 1|\n",
"|Love Is Not Over ...| csm3-us| 3|\n",
"|Like I'm Gonna Lo...|fm30ajetkgrrkbvae...| 2|\n",
"| Lo Siento BB:/| csm3-us| 5|\n",
"| Superbeast| csm3-us| 1|\n",
"| Amor de Hospital| csm3-us| 11|\n",
"| All Right Now|fm30ajetkgrrkbvae...| 2|\n",
"|Dancing On The Ce...|fm30ajetkgrrkbvae...| 4|\n",
"|Prologue / The Da...|fm30ajetkgrrkbvae...| 4|\n",
"|That Way (feat. J...| csm3-us| 11|\n",
"| Blame| csm3-us| 29|\n",
"| Speak Now|fm30ajetkgrrkbvae...| 2|\n",
"|A Little Respect ...|fm30ajetkgrrkbvae...| 3|\n",
"| Rockstar|fm30ajetkgrrkbvae...| 4|\n",
"+--------------------+--------------------+-----+\n",
"only showing top 20 rows\n"
]
},
{
"name": "stderr",
"output_type": "stream",
"text": [
" \r"
]
}
],
"source": [
" ## MAIN ## \n",
"def main():\n",
" app_name = \"ml_media_silver\"\n",
" spark = start_spark(app_name)\n",
"\n",
" bronze_table_name = \"bronze_spotify_data\"\n",
" silver_table_name = \"silver_spotify_data\"\n",
" bronze_df = read_jdbc(spark, bronze_table_name)\n",
"\n",
" bronze_df.groupBy(\"track_name\", \"username\").count().show()\n",
"\n",
" #spark.stop()\n",
"if __name__ == \"__main__\":\n",
" main()\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "c9a695ae",
"metadata": {},
"outputs": [],
"source": [
"## Databricks ##\n",
" ## MAIN ## \n",
"\n",
"# app_name = \"ml_media_silver\"\n",
"# bronze_table_name = \"bronze_spotify_data\"\n",
"# silver_table_name = \"silver_spotify_data\"\n",
"# silver_df = databricks_read(spark, bronze_table_name)\n",
"# silver_df.show(1)\n",
"\n",
"\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "4367ec82",
"metadata": {},
"outputs": [],
"source": [
"## --- ETL: Bronze → Silver (simple functions) --- ##\n",
"\n",
"def clean_platform(df):\n",
" \"\"\"Normalize verbose platform strings into 5 buckets.\"\"\"\n",
" return df.withColumn(\n",
" \"platform\",\n",
" F.when(F.lower(F.col(\"platform\")).contains(\"ios\"), \"ios\")\n",
" .when(F.lower(F.col(\"platform\")).contains(\"android\"), \"android\")\n",
" .when(F.lower(F.col(\"platform\")).rlike(\"windows|osx|mac|linux\"), \"desktop\")\n",
" .when(F.lower(F.col(\"platform\")).contains(\"web\"), \"web\")\n",
" .otherwise(\"other\"),\n",
" )\n",
"\n",
"def derive_skip_label(df):\n",
" \"\"\"Create a binary skip label from reason_end + ms_played.\n",
" A 'skip' = user forwarded or hit back before 30 seconds (Spotify's stream rule).\"\"\"\n",
" return df.withColumn(\n",
" \"is_skip\",\n",
" F.when(F.col(\"reason_end\").isin(\"fwdbtn\", \"backbtn\") & (F.col(\"ms_played\") < 30000), 1)\n",
" .otherwise(0),\n",
" )\n",
"\n",
"def extract_track_id(df):\n",
" \"\"\"Pull the Spotify track ID from the track_uri (spotify:track:XXXX).\"\"\"\n",
" return df.withColumn(\n",
" \"track_id\",\n",
" F.regexp_extract(F.col(\"track_uri\"), r\"spotify:track:([A-Za-z0-9]+)\", 1),\n",
" )\n",
"\n",
"def add_time_features(df):\n",
" \"\"\"Derive hour-of-day and day-of-week from the timestamp.\"\"\"\n",
" return (df\n",
" .withColumn(\"ts\", F.to_timestamp(\"ts\"))\n",
" .withColumn(\"hour\", F.hour(\"ts\"))\n",
" .withColumn(\"dow\", F.dayofweek(\"ts\"))\n",
" )\n",
"\n",
"\n",
"bronze_df = spark.table(\"workspace.default.bronze_spotify_data\")\n",
"\n",
"silver_df = (bronze_df\n",
" .transform(clean_platform)\n",
" .transform(derive_skip_label)\n",
" .transform(extract_track_id)\n",
" .transform(add_time_features)\n",
")\n",
"\n",
"print(f\"Bronze rows: {bronze_df.count()}\")\n",
"print(f\"Silver rows: {silver_df.count()}\")\n",
"silver_df.groupBy(\"is_skip\").count().show()"
]
},
{
"cell_type": "markdown",
"id": "cfe0a7c6",
"metadata": {},
"source": [
"## Feature Engineering (Silver → Gold / ML features)\n",
"\n",
"The silver table feeds into **feature engineering** — the bridge between ETL and ML.\n",
"We build per-play features that a classifier can consume."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "3db7ecd1",
"metadata": {},
"outputs": [],
"source": [
"## --- Feature Engineering for Skip Prediction --- ##\n",
"\n",
"from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler\n",
"from pyspark.ml import Pipeline\n",
"\n",
"# Select features relevant to skipping behaviour\n",
"feature_cols = [\"ms_played\", \"platform\", \"reason_start\", \"shuffle\", \"hour\", \"dow\"]\n",
"\n",
"gold_df = silver_df.select(\n",
" \"is_skip\",\n",
" F.col(\"ms_played\").cast(\"double\").alias(\"ms_played\"),\n",
" \"platform\", \"reason_start\",\n",
" F.col(\"shuffle\").cast(\"string\").alias(\"shuffle\"),\n",
" \"hour\", \"dow\",\n",
").na.drop()\n",
"\n",
"# Encode categoricals\n",
"categorical_cols = [\"platform\", \"reason_start\", \"shuffle\"]\n",
"indexers = [StringIndexer(inputCol=c, outputCol=f\"{c}_idx\", handleInvalid=\"keep\") for c in categorical_cols]\n",
"encoders = [OneHotEncoder(inputCol=f\"{c}_idx\", outputCol=f\"{c}_oh\") for c in categorical_cols]\n",
"\n",
"assembler = VectorAssembler(\n",
" inputCols=[\"ms_played\", \"platform_oh\", \"reason_start_oh\", \"shuffle_oh\", \"hour\", \"dow\"],\n",
" outputCol=\"features\",\n",
")\n",
"\n",
"feature_pipeline = Pipeline(stages=indexers + encoders + [assembler])\n",
"feature_model = feature_pipeline.fit(gold_df)\n",
"ml_ready = feature_model.transform(gold_df)\n",
"\n",
"print(f\"ML-ready rows: {ml_ready.count()}\")\n",
"ml_ready.select(\"is_skip\", \"features\").show(3, truncate=False)"
]
},
{
"cell_type": "markdown",
"id": "a3ed64ca",
"metadata": {},
"source": [
"## Model Training\n",
"\n",
"**Business question:** _Will a user skip a track within the first 30 seconds?_\n",
"\n",
"This is a **binary classification** problem. We train a Logistic Regression model on the ML-ready features and evaluate with AUC-ROC."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "b48e91df",
"metadata": {},
"outputs": [],
"source": [
"## --- Model Training --- ##\n",
"from pyspark.ml.classification import LogisticRegression\n",
"from pyspark.ml.evaluation import BinaryClassificationEvaluator\n",
"\n",
"# Split 80/20\n",
"train_df, test_df = ml_ready.randomSplit([0.8, 0.2], seed=42)\n",
"\n",
"lr = LogisticRegression(featuresCol=\"features\", labelCol=\"is_skip\", maxIter=20)\n",
"lr_model = lr.fit(train_df)\n",
"\n",
"predictions = lr_model.transform(test_df)\n",
"\n",
"evaluator = BinaryClassificationEvaluator(labelCol=\"is_skip\", metricName=\"areaUnderROC\")\n",
"auc = evaluator.evaluate(predictions)\n",
"\n",
"print(f\"Test AUC-ROC: {auc:.4f}\")\n",
"print(f\"Train rows: {train_df.count()} | Test rows: {test_df.count()}\")\n",
"predictions.select(\"is_skip\", \"prediction\", \"probability\").show(5, truncate=False)"
]
},
{
"cell_type": "markdown",
"id": "2c429452",
"metadata": {},
"source": [
"## Inference (Batch Scoring)\n",
"\n",
"Once trained, the model scores new plays to predict skip probability. This is **batch inference** — the same feature pipeline transforms new bronze data and the model produces predictions.\n",
"\n",
"In production this would run on a schedule (e.g., daily) or as a streaming job."
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "c225f0e7",
"metadata": {},
"outputs": [],
"source": [
"## --- Inference: Score new plays --- ##\n",
"# Simulate new data: take the last 1000 rows from silver as \"new\" plays\n",
"new_plays = silver_df.orderBy(F.desc(\"ts\")).limit(1000)\n",
"\n",
"# Apply the same feature pipeline\n",
"new_features = feature_model.transform(\n",
" new_plays.select(\n",
" F.col(\"ms_played\").cast(\"double\").alias(\"ms_played\"),\n",
" \"platform\", \"reason_start\",\n",
" F.col(\"shuffle\").cast(\"string\").alias(\"shuffle\"),\n",
" \"hour\", \"dow\",\n",
" ).na.drop()\n",
")\n",
"\n",
"# Score\n",
"scored = lr_model.transform(new_features)\n",
"\n",
"from pyspark.sql.functions import udf\n",
"from pyspark.sql.types import DoubleType\n",
"skip_prob_udf = udf(lambda v: float(v[1]), DoubleType())\n",
"\n",
"display(scored.select(\n",
" \"ms_played\", \"platform\", \"reason_start\", \"shuffle\",\n",
" skip_prob_udf(\"probability\").alias(\"skip_prob\"),\n",
" \"prediction\",\n",
").orderBy(F.desc(\"skip_prob\")).limit(10))"
]
},
{
"cell_type": "code",
"execution_count": null,
"id": "6cd50786",
"metadata": {},
"outputs": [],
"source": [
"## --- Tests for ETL Functions --- ##\n",
"import pytest\n",
"\n",
"# Helper: collect a single column into a Python list\n",
"def col_vals(df, colname):\n",
" return [row[colname] for row in df.select(colname).collect()]\n",
"\n",
"# ── Test 1: clean_platform ──\n",
"def test_clean_platform():\n",
" \"\"\"Platform strings are normalized into 5 buckets.\"\"\"\n",
" sample = spark.createDataFrame([\n",
" (\"iOS 14.2 (iPhone)\",),\n",
" (\"Android OS 11 API 30\",),\n",
" (\"Windows 10\",),\n",
" (\"macOS 12.0\",),\n",
" (\"web player\",),\n",
" (\"Partner amazon_salmon\",),\n",
" ], [\"platform\"])\n",
" result = clean_platform(sample)\n",
" vals = col_vals(result, \"platform\")\n",
" assert vals == [\"ios\", \"android\", \"desktop\", \"desktop\", \"web\", \"other\"], f\"Got {vals}\"\n",
" print(\"[x] test_clean_platform passed\")\n",
"\n",
"# ── Test 2: derive_skip_label ──\n",
"def test_derive_skip_label():\n",
" \"\"\"Skip = fwdbtn/backbtn with ms_played < 30s; everything else is not a skip.\"\"\"\n",
" sample = spark.createDataFrame([\n",
" (\"fwdbtn\", 500.0),\n",
" (\"backbtn\", 200.0),\n",
" (\"trackdone\", 180000.0),\n",
" (\"fwdbtn\", 60000.0), # fwdbtn but >= 30s → not a skip\n",
" (\"endplay\", 100.0), # wrong reason → not a skip\n",
" ], [\"reason_end\", \"ms_played\"])\n",
" result = derive_skip_label(sample)\n",
" vals = col_vals(result, \"is_skip\")\n",
" assert vals == [1, 1, 0, 0, 0], f\"Got {vals}\"\n",
" print(\"[x] test_derive_skip_label passed\")\n",
"\n",
"# ── Test 3: extract_track_id ──\n",
"def test_extract_track_id():\n",
" \"\"\"Track ID is extracted from the spotify:track: URI.\"\"\"\n",
" sample = spark.createDataFrame([\n",
" (\"spotify:track:0cKk8BKEi7zXbdrYdyqBP5\",),\n",
" (\"spotify:track:2VgPA0WnwvE2nbltxa3BHz\",),\n",
" (\"\",), # empty URI → empty string\n",
" ], [\"track_uri\"])\n",
" result = extract_track_id(sample)\n",
" vals = col_vals(result, \"track_id\")\n",
" assert vals[0] == \"0cKk8BKEi7zXbdrYdyqBP5\"\n",
" assert vals[1] == \"2VgPA0WnwvE2nbltxa3BHz\"\n",
" assert vals[2] == \"\"\n",
" print(\"[x] test_extract_track_id passed\")\n",
"\n",
"# ── Test 4: add_time_features ──\n",
"def test_add_time_features():\n",
" \"\"\"Hour and day-of-week are derived from the ts timestamp.\"\"\"\n",
" sample = spark.createDataFrame([\n",
" (\"2021-10-27 20:17:11\",), # Wed, 8pm\n",
" (\"2022-01-03 07:30:00\",), # Mon, 7am\n",
" ], [\"ts\"])\n",
" result = add_time_features(sample)\n",
" hours = col_vals(result, \"hour\")\n",
" dows = col_vals(result, \"dow\")\n",
" assert hours == [20, 7], f\"Got hours {hours}\"\n",
" assert dows == [4, 2], f\"Got dows {dows}\" # Sun=1, Mon=2, …, Wed=4\n",
" print(\"[x] test_add_time_features passed\")\n",
"\n",
"# ── Run all tests ──\n",
"test_clean_platform()\n",
"test_derive_skip_label()\n",
"test_extract_track_id()\n",
"test_add_time_features()\n",
"print(\"\\n🎉 All 4 ETL tests passed!\")"
]
},
{
"cell_type": "markdown",
"id": "23c744a6",
"metadata": {},
"source": [
"## Data Framework & Business Question\n",
"\n",
"### Framework: Medallion Architecture (Lakehouse)\n",
"\n",
"This pipeline uses the **Medallion Architecture** — Databricks' multi-layer data design pattern:\n",
"\n",
"```\n",
"Raw JSON Export (Spotify)\n",
" |\n",
" v\n",
" ┌─────────┐ ┌───────────┐ ┌──────────┐ ┌──────────┐\n",
" │ RAW │→ │ BRONZE │→ │ SILVER │→ │ GOLD │→ ML Model\n",
" │ (exact) │ │ (cleaned) │ │ (conform)│ │ (aggreg) │ (train/infer)\n",
" └─────────┘ └───────────┘ └──────────┘ └──────────┘\n",
"```\n",
"\n",
"| Layer | What Happens | This Project |\n",
"| --- | --- | --- |\n",
"| **Raw** | Land the data exactly as exported | `raw_spotify_data` — 105K rows, original verbose column names, PII included |\n",
"| **Bronze** | Clean names, drop PII, add time columns | `bronze_spotify_data` — 103K rows, renamed columns, derived `hour`/`date` |\n",
"| **Silver** | Conform types, dedupe, sessionize, derive flags | `silver_df` (in-memory) — platform normalized, `is_skip` label, `track_id` extracted |\n",
"| **Gold** | Feature engineering for ML | `ml_ready` — encoded categorical features, assembled vector, ready for training |\n",
"\n",
"### How It Answers a Business Question\n",
"\n",
"**Question:** _Which tracks will a user skip, and can we predict it before it happens?_\n",
"\n",
"**Why it matters:** Skip rate is a key engagement metric for music platforms. If we can predict skips:\n",
"\n",
"* **Recommendation engines** can avoid surfaging tracks a user will skip\n",
"* **Playlist curation** can order tracks to minimize early-exit\n",
"* **A/B testing** can measure how UI changes affect skip behaviour\n",
"\n",
"**How the framework supports this:**\n",
"\n",
"1. **Bronze ETL** (`spotify_bronze_silver`) — lands and cleans the raw export so the data is queryable\n",
"2. **Silver ETL** (this notebook) — derives the `is_skip` label and normalizes features like `platform`\n",
"3. **Feature Engineering** (Gold) — encodes categoricals and assembles a feature vector\n",
"4. **Model Training** — Logistic Regression learns which feature patterns lead to skips\n",
"5. **Inference** — Scores new plays in batch; high `skip_prob` flags tracks to avoid recommending\n",
"\n",
"### ML Engineering Connection\n",
"\n",
"The ETL functions aren't just data plumbing — they are the **ML feature pipeline**:\n",
"\n",
"* `clean_platform()` → categorical feature `platform` (one-hot encoded)\n",
"* `derive_skip_label()` → **target variable** `is_skip` (what the model predicts)\n",
"* `extract_track_id()` → enables per-track aggregation and cold-start handling\n",
"* `add_time_features()` → temporal features `hour` and `dow` (when do users skip most?)\n",
"\n",
"In a production ML system, steps 2–3 would be packaged as a **feature pipeline** (e.g., via Databricks Feature Store or a scheduled job), the model in step 4 would be registered in **MLflow**, and step 5 would run as a **Model Serving endpoint** for real-time scoring."
]
}
],
"metadata": {
"kernelspec": {
"display_name": ".venv",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.13.7"
}
},
"nbformat": 4,
"nbformat_minor": 5
}