# 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), 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) 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.