205 lines
11 KiB
Python
205 lines
11 KiB
Python
|
# 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).
|