From 95a7d229e9efeafb65e52ad671615aa5583e5dd4 Mon Sep 17 00:00:00 2001 From: CC Date: Mon, 19 Oct 2026 08:23:52 +0100 Subject: [PATCH] ml --- claude/bronze_foreign.py | 153 ------------ claude/bronze_redshift.py | 195 --------------- claude/business_model.md | 43 ---- claude/etl_spark.md | 490 -------------------------------------- claude/spotify_silver.py | 214 ----------------- claude/trakt_models.py | 205 ---------------- claude/trakt_silver.py | 206 ---------------- spotify_silver_gold.ipynb | 259 +++++++++++++++++++- 8 files changed, 257 insertions(+), 1508 deletions(-) delete mode 100644 claude/bronze_foreign.py delete mode 100644 claude/bronze_redshift.py delete mode 100644 claude/business_model.md delete mode 100644 claude/etl_spark.md delete mode 100644 claude/spotify_silver.py delete mode 100644 claude/trakt_models.py delete mode 100644 claude/trakt_silver.py diff --git a/claude/bronze_foreign.py b/claude/bronze_foreign.py deleted file mode 100644 index 3accb3d..0000000 --- a/claude/bronze_foreign.py +++ /dev/null @@ -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})) \ No newline at end of file diff --git a/claude/bronze_redshift.py b/claude/bronze_redshift.py deleted file mode 100644 index 27280b7..0000000 --- a/claude/bronze_redshift.py +++ /dev/null @@ -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})) \ No newline at end of file diff --git a/claude/business_model.md b/claude/business_model.md deleted file mode 100644 index 3ab2cb8..0000000 --- a/claude/business_model.md +++ /dev/null @@ -1,43 +0,0 @@ -Yes. Most business questions map to a model family once you work out the unit of analysis (who or what is being counted) and the type of outcome. Here's the mapping using your Trakt and Spotify examples. - -**Who** -- *Questions:* Who are my heavy skippers? Who is most like whom? -- *Frameworks:* clustering (K-means) for segments, logistic models for propensity (e.g. who completes shows), similarity scores (Jaccard, cosine) for how alike two users are. -- *Example:* Spotify user A vs. B overlap, or the viewing-day personas from K-means. - -**What** -- *Questions:* What should we recommend? What drives completion? -- *Frameworks:* recommenders (item-item, Word2Vec), association rules (FP-growth), and GLM coefficients that rank drivers. -- *Example:* tracks B plays that A hasn't heard, or runtime's effect on finishing a show. - -**When** -- *Questions:* When will demand peak? How long until someone abandons a show? When did habits change? -- *Frameworks:* time series (ETS, Prophet) for forecasts, survival analysis for time-to-event, Poisson with seasonality terms, change-point detection. -- *Example:* predicted days until a 50% chance of abandoning a series. - -**Where** -- *Questions:* Which region, device or channel? -- *Frameworks:* group-level GLMs with country or platform terms, hierarchical models to shrink small-sample regions toward the average. -- *Example:* Spotify's country and platform fields. Trakt has almost no "where", so there it's mostly a time-zone question. - -**How many / how much** -- *Frameworks:* Poisson or negative binomial for counts, gamma for skewed amounts, with an offset for exposure. -- *Example:* plays per day on weekends vs. weekdays. - -**Why** -- *Frameworks:* GLM coefficients show association. Causal designs (A/B tests, difference-in-differences) show effect. -- *Example:* "Did B hearing it first cause A's plays?" needs more than a coefficient. - -**The shortcut:** ask what the outcome looks like. -- A count → Poisson -- Yes/no → logistic -- Time until something → survival -- A skewed amount → gamma -- Groups of similar things → clustering -- Next best item → recommender - -**Two cautions** -- Plain SQL answers many who/what/when/where questions descriptively (top artists, busiest days). Reach for a model when you need prediction, comparison with uncertainty, or an effect size. -- Models give probabilities and rates, not certainties, and sparse data (one or two users) limits which ones are valid. - -Want to take one real business question from your data and walk it through question → outcome type → model → features? \ No newline at end of file diff --git a/claude/etl_spark.md b/claude/etl_spark.md deleted file mode 100644 index 34da1bd..0000000 --- a/claude/etl_spark.md +++ /dev/null @@ -1,490 +0,0 @@ -# Building an ETL Pipeline with PySpark and Spark SQL - -**Setup:** `pip install pyspark`. Java 8, 11, or 17 must be installed. - ---- - -## Extract Phase - -The extract phase is the first step in an ETL pipeline, where raw data is collected from files, databases, and APIs. In PySpark, every extract starts with a `SparkSession`, the entry point for DataFrames and Spark SQL. - -In this task you will learn how to: - -- Create a SparkSession -- Read flat files into Spark DataFrames -- Read data from a SQL database over JDBC -- Read data from REST APIs into Spark DataFrames - -To get started open the `extract.py` file. - -### Task 1.0.1: Create a SparkSession - -```python -from pyspark.sql import SparkSession - -spark = ( - SparkSession.builder - .appName("etl_pipeline") - .config("spark.jars.packages", "com.mysql:mysql-connector-j:8.3.0") # MySQL JDBC driver, used in Task 1.0.3 - .getOrCreate() -) -``` - -`.master("local[*]")` runs Spark on your machine. On a real cluster you would remove `.master()` and let the cluster manager supply it. - -0 / 1 checks: Create a SparkSession named `spark`. - -### Task 1.0.2: Extracting Data from Flat Files - -Spark reads these formats natively: - -- `.csv` -- `.json` -- `.parquet` - -Excel is not supported natively, so you would read it with Pandas and convert it (see Task 1.0.4). - -For this task: - -1. Read `data/example.csv` into `df1` with `spark.read.csv`. Set `header=True` and `inferSchema=True`, otherwise every column is read as a string. -2. Read `data/example.json` into `df2` with `spark.read.json`. -3. Read `data/example.parquet` into `df3` with `spark.read.parquet`. - -```python -df1 = spark.read.csv("data/example.csv", header=True, inferSchema=True) -df2 = spark.read.json("data/example.json") -df3 = spark.read.parquet("data/example.parquet") - -df1.printSchema() -df1.show(5) -``` - -0 / 1 checks: Read different file types into Spark DataFrames using `spark.read`. - -### Task 1.0.3: Loading Multiple Files with Wildcards - -Spark is built for parallel processing, so reading many files at once is a core use case. Spark will split the files into partitions and process them across your cores or cluster. - -1. Observe the `dask_data_*.csv` files in the `data` folder. -2. Assign `df4` the result of: - -```python -df4 = spark.read.csv("data/dask_data_*.csv", header=True, inferSchema=True) -``` - -3. Optionally inspect metadata about the DataFrame: - -```python -print(df4.count()) # row count (triggers a job) -print(len(df4.columns)) # column count -print(df4.rdd.getNumPartitions()) # how many partitions Spark created -``` - -0 / 1 checks: Read the `dask_data` files in `data` into the `df4` DataFrame. - -### Task 1.0.4: Extract Data from Database Tables (JDBC) - -Spark reads databases through JDBC. Instead of a SQLAlchemy engine, you pass connection options to `spark.read.format("jdbc")`. - -1. Assign the query to the `query` variable: - -```python -query = "SELECT TABLE_NAME FROM information_schema.TABLES" -``` - -2. Read the results into `df5`: - -```python -df5 = ( - spark.read.format("jdbc") - .option("url", "jdbc:mysql://localhost:3306/mydb") - .option("query", query) - .option("user", "root") - .option("password", "your_password") - .option("driver", "com.mysql.cj.jdbc.Driver") - .load() -) -df5.show() -``` - -For large tables, add `partitionColumn`, `lowerBound`, `upperBound`, and `numPartitions` so Spark reads in parallel rather than through a single connection. - -0 / 1 checks: Read the provided query results into the `df5` Spark DataFrame. - -### Task 1.0.5: Extracting REST API Data with Requests - -Spark has no built-in REST reader. The common pattern is to fetch the JSON with `requests` on the driver, then convert it into a DataFrame. - -1. Open a new terminal tab and run `python3 scripts/app.py` to start the Flask app. -2. Fetch the data and create a DataFrame: - -```python -import requests - -url = "http://localhost:5000/random-user" -r = requests.get(url) -print(r.json()) - -df6 = spark.createDataFrame([r.json()]) # a dict becomes a one-row DataFrame -df6.show() -``` - -For a list of records, pass the list directly: `spark.createDataFrame(r.json())`. - -### Task 1.0.6: Authenticating with Requests - -Add a Bearer token to the request headers to authenticate to the `/protected` endpoint. - -1. In the `headers` dictionary, add the key `Authorization`. -2. Set the value to `Bearer mysecrettoken123`. -3. Assign the result to `response`: - -```python -headers = {"Authorization": "Bearer mysecrettoken123"} -response = requests.get("http://localhost:5000/protected", headers=headers) -df7 = spark.createDataFrame([response.json()]) -``` - -0 / 1 checks: Assign the header values and authenticate to the `/protected` endpoint, assigning the results to `response`. - -### (Optional) Reading JSON strings with a schema - -`createDataFrame` infers types, which can be fragile with nested or inconsistent API data. Define a schema explicitly for predictable results: - -```python -from pyspark.sql.types import StructType, StructField, StringType, IntegerType - -schema = StructType([ - StructField("name", StringType(), True), - StructField("age", IntegerType(), True), -]) -df8 = spark.createDataFrame(response.json(), schema=schema) -``` - ---- - -## Transform Phase - -The transform phase is where raw data is cleansed, standardized, aggregated, and joined. PySpark gives you two interchangeable APIs for this: the **DataFrame API** (`pyspark.sql.functions`) and **Spark SQL**. Both compile to the same optimized execution plan, so use whichever reads better. - -In this task you will learn how to: - -- Convert data types and date formats -- Replace values and clean dirty data with regex -- Filter, aggregate, and create calculated columns -- Join DataFrames -- Run the same logic with Spark SQL using temp views - -**Example Data* - -```python -data = { - "Name": ["Adam", "Bob", "Charlie"], - "Children": [1, 2, 0], - "ZipCode":[10001,90001,60601], - "City": ["New York", "Los Angeles", "Chicago"], - "Birthdate": [pd.Timestamp('20000101'),pd.Timestamp('19900418'),pd.Timestamp('20180310')], - "CharacterSeq": ['****','____','****'], - "UserId":['A111','B222','C333'], - "TShirtSize":['M','XL','XL'], - "Salary":[50000,60000,70000] -} -``` - -```python -from pyspark.sql import functions as F -from pyspark.sql.types import StringType, DoubleType -``` - -### Task 2.0.1: Convert Column Data Types - -Observe the data dictionary. Notice that `ZipCode` is an integer. Convert it to a string with `cast`: - -```python -df = df.withColumn("ZipCode", F.col("ZipCode").cast(StringType())) -df.printSchema() -``` - -DataFrames are immutable in Spark. `withColumn` returns a new DataFrame, so you must assign the result back. - -0 / 1 checks: Convert the `ZipCode` column to a string data type. - -### Task 2.0.2: Convert Date Formats - -`Birthdate` is in `yyyy-MM-dd` format. We want `MM-dd-yyyy`. Parse the string into a date, then format it back out: - -```python -df = df.withColumn( - "Birthdate", - F.date_format(F.to_date("Birthdate", "yyyy-MM-dd"), "MM-dd-yyyy") -) -df.select("Birthdate").show(5) -``` - -Spark date patterns are case-sensitive. `MM` is month and `mm` is minutes. - -0 / 1 checks: Convert the `Birthdate` column to the `MM-dd-yyyy` format. - -### Task 2.0.3: Replace Values with a Mapping - -1. Observe the `replace_map` dictionary. The value being replaced comes first, followed by the replacement. -2. Apply it to the `City` column: - -```python -replace_map = {"NYC": "New York", "LA": "Los Angeles"} -df = df.replace(replace_map, subset=["City"]) -``` - -0 / 1 checks: Convert the existing `City` values using `df.replace()`. - -### Task 2.0.4: Using Regex to Replace Dirty Data - -Use `regexp_replace` to find and replace patterns. `*` is a special regex character, so it must be escaped. - -1. Observe the `CharacterSeq` column. -2. Write a regex pattern that matches four asterisks: `\*{4}` (or `\*\*\*\*`). -3. Replace them with underscores: - -```python -df = df.withColumn("CharacterSeq", F.regexp_replace("CharacterSeq", r"\*{4}", "____")) -``` - -0 / 1 checks: Use a regex on `CharacterSeq` to replace `****` with `____`. - -### (Optional) DataFrame Filtering - -```python -# Column expression syntax -df.filter(F.col("Name") == "Adam") - -# SQL string syntax -df.filter("Name = 'Adam'") - -# Assign to a new DataFrame for a sub-dataset -adams = df.filter(F.col("Name") == "Adam") -adams.show() -``` - -`where()` is an alias for `filter()`. - -### Task 2.0.5: Creating Custom Calculated Columns - -1. Observe the `cost_of_living_adjust` function, which takes an original value and an `inc_percent`. -2. Wrap it in a UDF (user-defined function) and create `Salary_COL`: - -```python -def cost_of_living_adjust(value, inc_percent): - return value * (1 + inc_percent / 100) if value is not None else None - -col_udf = F.udf(lambda s: cost_of_living_adjust(s, 5), DoubleType()) -df = df.withColumn("Salary_COL", col_udf(F.col("Salary"))) -``` - -Python UDFs are slow because every row is serialized between the JVM and Python. When you can, use native column expressions, which Spark can optimize: - -```python -df = df.withColumn("Salary_COL", F.col("Salary") * (1 + 5 / 100)) -``` - -0 / 1 checks: Create a new `Salary_COL` column by applying `cost_of_living_adjust` to `Salary` with `inc_percent=5`. - -### (Optional) DataFrame Aggregating - -```python -# Whole-DataFrame aggregate -df.agg( - F.count("Salary").alias("count"), - F.min("Salary").alias("min"), - F.max("Salary").alias("max"), - F.mean("Salary").alias("mean"), - F.sum("Salary").alias("sum"), -).show() - -# Grouped aggregate -df.groupBy("TShirtSize").agg(F.mean("Salary").alias("avg_salary")).show() -``` - -### Task 2.0.6: Joining Data - -Spark's `.join()` mirrors SQL joins. The join types are `inner`, `left`, `right`, `outer`, `left_semi`, and `left_anti`. - -1. Observe both DataFrames (`df` and `df2`). -2. Use `UserId` as the `on` parameter. -3. Use `left` as the `how` parameter. -4. Assign the result to `df_merged`: - -```python -df_merged = df.join(df2, on="UserId", how="left") -df_merged.show() -``` - -Passing `on="UserId"` (a string) keeps a single `UserId` column. Joining on `df.UserId == df2.UserId` keeps both, which causes ambiguous-column errors later. - -0 / 1 checks: Use `.join` to combine data in `df` and `df2` into `df_merged`. - -### Task 2.0.7: Transforming with Spark SQL - -Spark SQL lets you query DataFrames with plain SQL. You first register a DataFrame as a temporary view: - -1. Register the view: - -```python -df_merged.createOrReplaceTempView("users") -``` - -2. Query it with `spark.sql()`, which returns a DataFrame: - -```python -result = spark.sql(""" - SELECT - City, - COUNT(*) AS user_count, - ROUND(AVG(Salary_COL), 2) AS avg_salary - FROM users - WHERE Salary_COL IS NOT NULL - GROUP BY City - HAVING COUNT(*) > 1 - ORDER BY avg_salary DESC -""") -result.show() -``` - -3. Because the result is a DataFrame, you can switch back to the DataFrame API at any point and mix the two styles freely. - -0 / 1 checks: Register `df_merged` as a temp view named `users` and aggregate it with `spark.sql()`. - ---- - -## Loading Data - -Data writing in PySpark refers to saving a Spark DataFrame to files or databases through the `df.write` interface. Spark supports `csv()`, `json()`, `parquet()`, `orc()`, and `jdbc()`, allowing you to export data to files, data lakes, or relational databases. - -There are three key differences from Pandas. Spark writes a **directory** of part files (one per partition) rather than a single file. You control overwrite behavior with **save modes** (`overwrite`, `append`, `ignore`, `errorifexists`). And **Parquet** is the preferred format for analytics because it is columnar, compressed, and keeps the schema. - -To get started open the `load.py` file. - -### Task 3.0.1: Loading Data to a CSV File - -1. Create a dictionary with the following python code: - -```python -data = { - 'Transaction_ID': ['T1001', 'T1002', 'T1003', 'T1004'], - 'Customer_Name': ['Alice Johnson', 'Bob Lee', 'Charlie Kim', 'David Clark'], - 'Product': ['Laptop', 'Smartphone', 'Tablet', 'Smartwatch'], - 'Quantity': [1, 2, 3, 1], - 'Price_Per_Unit': [1200, 800, 400, 200], - 'Total_Sale': [1200, 1600, 1200, 200], - 'Sale_Date': ['2025-03-20', '2025-03-21', '2025-03-21', '2025-03-22'] -} -``` - -2. Convert the dictionary into a Spark DataFrame variable `df`. Spark expects a list of rows, so reshape the dictionary first: - -```python -import pandas as pd - -df = spark.createDataFrame(pd.DataFrame(data)) -``` - -3. Use `df.write.csv` to write the contents to `data/Transactions`. Include the following options: - - `header=True` - - `mode="overwrite"` - -```python -( - df.coalesce(1) # collapse to one partition, so one part file - .write - .mode("overwrite") - .option("header", True) - .csv("data/Transactions") -) -``` - -Spark creates a folder named `Transactions` containing `part-00000-....csv` and a `_SUCCESS` marker. `coalesce(1)` is fine for small data, but avoid it on large datasets because it funnels everything through one task. - -0 / 1 checks: Convert the dictionary values to a DataFrame and load it to `Transactions` in the `data` directory. - -### Task 3.0.2: Loading Data to Parquet - -```python -df.write.mode("overwrite").parquet("data/Transactions_parquet") -``` - -Partitioning by a column splits the output into subfolders, which speeds up later filtered reads: - -```python -df.write.mode("overwrite").partitionBy("Sale_Date").parquet("data/Transactions_by_date") -``` - -0 / 1 checks: Write the DataFrame to Parquet. - -### Task 3.0.3: Loading Data to a Database with JDBC - -```python -( - df.write.format("jdbc") - .option("url", "jdbc:mysql://localhost:3306/mydb") - .option("dbtable", "transactions") - .option("user", "root") - .option("password", "your_password") - .option("driver", "com.mysql.cj.jdbc.Driver") - .mode("append") - .save() -) -``` - -Use `append` to add rows. `overwrite` drops and recreates the table unless you also set `.option("truncate", True)`. - -0 / 1 checks: Load the DataFrame into a `transactions` table. - -### (Optional) Validating Data Operations - -Use `assert` statements to validate that the pipeline finished successfully. Check that row counts, column types, and totals match the expected results. - -```python -csv_df = spark.read.csv("data/Transactions", header=True, inferSchema=True) - -assert csv_df.count() == 4, "Row counts do not match expected values!" -print(f"Row counts match: {csv_df.count()} rows") - -assert csv_df.agg(F.sum("Total_Sale")).first()[0] == 4200, "Total sales mismatch!" - -expected_cols = {"Transaction_ID", "Customer_Name", "Product", "Quantity", - "Price_Per_Unit", "Total_Sale", "Sale_Date"} -assert set(csv_df.columns) == expected_cols, "Schema mismatch!" -``` - -To compare against the database table, read it back with `spark.read.format("jdbc")` and compare `.count()` to the source DataFrame. - ---- - -## Putting It All Together - -Wrap the three phases in functions so the pipeline is testable and reusable: - -```python -def extract(spark): - return spark.read.csv("data/example.csv", header=True, inferSchema=True) - -def transform(df): - return ( - df.withColumn("ZipCode", F.col("ZipCode").cast("string")) - .withColumn("Birthdate", F.date_format(F.to_date("Birthdate"), "MM-dd-yyyy")) - .withColumn("Salary_COL", F.col("Salary") * 1.05) - .dropDuplicates() - ) - -def load(df, path): - df.write.mode("overwrite").parquet(path) - -if __name__ == "__main__": - spark = SparkSession.builder.appName("etl").master("local[*]").getOrCreate() - raw = extract(spark) - clean = transform(raw) - load(clean, "data/output") - assert spark.read.parquet("data/output").count() == clean.count() - spark.stop() -``` - -I can also put this into a downloadable file or turn it into a Jupyter-style notebook. \ No newline at end of file diff --git a/claude/spotify_silver.py b/claude/spotify_silver.py deleted file mode 100644 index 79eaff9..0000000 --- a/claude/spotify_silver.py +++ /dev/null @@ -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) \ No newline at end of file diff --git a/claude/trakt_models.py b/claude/trakt_models.py deleted file mode 100644 index c2ba4a4..0000000 --- a/claude/trakt_models.py +++ /dev/null @@ -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). \ No newline at end of file diff --git a/claude/trakt_silver.py b/claude/trakt_silver.py deleted file mode 100644 index 2c35952..0000000 --- a/claude/trakt_silver.py +++ /dev/null @@ -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), 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. \ No newline at end of file diff --git a/spotify_silver_gold.ipynb b/spotify_silver_gold.ipynb index 96275e9..f23bfbb 100644 --- a/spotify_silver_gold.ipynb +++ b/spotify_silver_gold.ipynb @@ -138,9 +138,9 @@ "\n", " bronze_table_name = \"bronze_spotify_data\"\n", " silver_table_name = \"silver_spotify_data\"\n", - " silver_df = read_jdbc(spark, bronze_table_name)\n", + " bronze_df = read_jdbc(spark, bronze_table_name)\n", "\n", - " silver_df.groupBy(\"track_name\", \"username\").count().show()\n", + " bronze_df.groupBy(\"track_name\", \"username\").count().show()\n", "\n", " #spark.stop()\n", "if __name__ == \"__main__\":\n", @@ -165,6 +165,261 @@ "\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": "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": {