206 lines
9.9 KiB
Python
206 lines
9.9 KiB
Python
|
# 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.
|