media_ml/spotify_bronze_silver.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "e9bf1d26",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession, DataFrame\n",
    "from pyspark.sql import functions as F\n",
    "from pathlib import Path\n",
    "import glob\n",
    "import os\n",
    "from dotenv import load_dotenv, find_dotenv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "11bb5dc2",
   "metadata": {},
   "outputs": [],
   "source": [
    "                            ## Extract ##\n",
    "\n",
    "BASE = Path.cwd() / \"ingest\"\n",
    "\n",
    "def start_spark(app_name):\n",
    "    jar_path = str(Path.cwd() / \"postgresql.jar\")\n",
    "    spark = SparkSession.builder.appName(app_name).config(\"spark.jars\", jar_path).config(\"spark.driver.extraClassPath\", jar_path).getOrCreate()\n",
    "    spark.sparkContext.setLogLevel(\"ERROR\")\n",
    "    return spark\n",
    "\n",
    "def music_source_paths():\n",
    "    return sorted(glob.glob(str(BASE / \"*_sesh\" / \"*.json\")))\n",
    "\n",
    "def video_source_paths():\n",
    "    return sorted(glob.glob(str(BASE / \"trakt_*\" / \"*.json\")))\n",
    "\n",
    "def read_json(spark, paths):\n",
    "    if not paths:\n",
    "        raise FileNotFoundError(\"No JSON files matched the pattern\")\n",
    "    return (\n",
    "        spark.read\n",
    "        .format(\"json\")\n",
    "        .option(\"multiLine\", True)\n",
    "        .json(paths)                                  # list of paths, one read\n",
    "        # .withColumn(\"source_file\", F.input_file_name())\n",
    "        # .withColumn(\n",
    "        #     \"source_folder\",\n",
    "        #     F.regexp_extract(\"source_file\", r\"/ingest/([^/]+)/\", 1),\n",
    "        # )\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "a898326e",
   "metadata": {},
   "outputs": [],
   "source": [
    "                                ## Transform ##\n",
    "def null_counts_by_column(df):\n",
    "    null_counts = df.select([\n",
    "        F.count(F.when(F.col(c).isNull(), c)).alias(c)\n",
    "        for c in df.columns\n",
    "    ])\n",
    "    return null_counts\n",
    "\n",
    "def metadata_df(df):\n",
    "    return df.withColumnRenamed(\"master_metadata_album_album_name\", \"album_name\").withColumnRenamed(\"master_metadata_album_artist_name\", \"artist_name\").withColumnRenamed(\"master_metadata_track_name\", \"track_name\")\n",
    "\n",
    "def rename_columns(df, rename_map, drop_columns):\n",
    "    missing_rename = set(rename_map) - set(df.columns)\n",
    "    missing_dropped = set(drop_columns) - set(df.columns)\n",
    "\n",
    "    if missing_rename:\n",
    "        print(f\"Warning: columns not found: {missing_rename}\")\n",
    "    if missing_dropped:\n",
    "        print(f\"Warning: columns not found: {missing_dropped}\")\n",
    "\n",
    "    df = df.drop(*drop_columns)\n",
    "    \n",
    "    return df.toDF(*[rename_map.get(c, c) for c in df.columns])\n",
    "\n",
    "def track_df(df):\n",
    "    return df.where(df[\"track_name\"].isNotNull())\n",
    "\n",
    "def parse_timestamps(df: DataFrame, columns) -> DataFrame:\n",
    "    for column in columns:\n",
    "        df = df.withColumn(column, F.to_timestamp(column))\n",
    "    return df\n",
    "\n",
    "def parse_booleans(df: DataFrame, columns) -> DataFrame:\n",
    "    df = df.select([F.col(c).cast(\"string\") for c in columns])\n",
    "    return df\n",
    "\n",
    "def add_minutes_played(df: DataFrame, src, dest) -> DataFrame:\n",
    "    return (df.withColumn(dest, F.round(F.col(src) / 60000, 2)))\n",
    "\n",
    "def add_time_parts(df: DataFrame, ts_col) -> DataFrame:\n",
    "    return (df\n",
    "            .withColumn(\"ts\", F.to_timestamp(\"ts\"))\n",
    "            .withColumn(\"offline_timestamped\", F.to_timestamp(\"offline_timestamp\"))\n",
    "            .withColumn('time', F.date_format('ts', 'HH:mm:ss'))\n",
    "            .withColumn('date', F.date_format('ts', 'yyyy-MM-dd'))\n",
    "            .withColumn('offline_time', F.date_format('offline_timestamped', 'HH:mm:ss'))\n",
    "            .withColumn('offline_date', F.date_format('offline_timestamped', 'yyyy-MM-dd'))            \n",
    "            )\n",
    "\n",
    "def clean_track_df(df: DataFrame) -> DataFrame:\n",
    "    return(df\n",
    "            .transform(rename_columns, SPOTIFY_RENAMES, MUSIC_DROP_COLUMNS)\n",
    "            .transform(track_df)\n",
    "            # .transform(parse_timestamps, [\"ts\", \"offline_timestamp\"])\n",
    "            # .transform(parse_booleans, [\"offline\", \"shuffle\", \"skipped\"]) ##returns only the selected columns\n",
    "            .transform(add_minutes_played, \"ms_played\", \"ms_played\")\n",
    "            .transform(add_time_parts, \"ts\")\n",
    "            )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "817475cf",
   "metadata": {},
   "outputs": [],
   "source": [
    "                                        ## Load ##\n",
    "def get_jdbc_config():\n",
    "    \"\"\"Load env vars and return a Postgres URL\"\"\"\n",
    "    load_dotenv(find_dotenv())\n",
    "    \n",
    "    username=os.getenv(\"DATABASE_USERNAME\")\n",
    "    password=os.getenv(\"DATABASE_PASSWORD\")\n",
    "    host=os.getenv(\"DATABASE_IP\")\n",
    "    port=int(os.getenv(\"DATABASE_PORT\", 5432))\n",
    "    database=os.getenv(\"DATABASE_NAME\")\n",
    "    \n",
    "    missing = [k for k, v in {\n",
    "        \"DATABASE_USERNAME\": username,\n",
    "        \"DATABASE_PASSWORD\": password,\n",
    "        \"DATABASE_IP\": host,\n",
    "        \"DATABASE_NAME\": database,}.items() if not v]\n",
    "    if missing:\n",
    "        raise ValueError(f\"Missing env vars: {missing}\")\n",
    "\n",
    "    url = f\"jdbc:postgresql://{host}:{port}/{database}\"\n",
    "    properties = {\n",
    "        \"user\": username,\n",
    "        \"password\": password,\n",
    "        \"driver\": \"org.postgresql.Driver\"\n",
    "    }\n",
    "    return url, properties\n",
    "\n",
    "def load_data(df, table_name):\n",
    "    \"\"\"Load the data into the database.\"\"\"\n",
    "    url, properties = get_jdbc_config()\n",
    "    df.write.jdbc(url=url, table=table_name, mode=\"overwrite\", properties=properties)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "c609291c",
   "metadata": {},
   "outputs": [],
   "source": [
    "## Config \n",
    "SPOTIFY_RENAMES = {\n",
    "    \"master_metadata_album_album_name\": \"album_name\",\n",
    "    \"master_metadata_album_artist_name\": \"artist_name\",\n",
    "    \"master_metadata_track_name\": \"track_name\",\n",
    "    \"spotify_episode_uri\" : \"episode_uri\",\n",
    "    \"spotify_track_uri\" : \"track_uri\"\n",
    "}\n",
    "\n",
    "TRAKT_RENAMES = {\"watched_at\": \"ts\", \"title\": \"video_title\"}\n",
    "\n",
    "MUSIC_DROP_COLUMNS = [\"ip_addr_decrypted\", \"user_agent_decrypted\", \"incognito_mode\", \"conn_country\", \"spotify_episode_uri\", \"episode_name\", \"episode_show_name\"]\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4cc14bd0",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "WARNING: Using incubator modules: jdk.incubator.vector\n",
      "Using Spark's default log4j profile: org/apache/spark/log4j2-defaults.properties\n",
      "26/10/18 17:57:09 WARN Utils: Your hostname, MacBook-Neo-232.local, resolves to a loopback address: 127.0.0.1; using 192.168.10.130 instead (on interface en0)\n",
      "26/10/18 17:57:09 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address\n",
      "26/10/18 17:57:09 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n",
      "Using Spark's default log4j profile: org/apache/spark/log4j2-defaults.properties\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n",
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      " Raw DF Rows: 105477\n",
      " Transformed DF Rows: 102626\n",
      " Row difference is 2851\n"
     ]
    }
   ],
   "source": [
    "                                    ## MAIN ## \n",
    "def main():\n",
    "    app_name = \"ml_media\"\n",
    "    spark = start_spark(app_name)\n",
    "\n",
    "    table_name = \"spotify_data\"\n",
    "    raw_df = read_json(spark, music_source_paths())\n",
    "    transformed_df = clean_track_df(raw_df)\n",
    "    load_data(transformed_df, table_name)\n",
    "\n",
    "    print(f\" Raw DF Rows: {raw_df.count()}\")\n",
    "    print(f\" Transformed DF Rows: {transformed_df.count()}\")\n",
    "    print(f\" Row difference is {raw_df.count() - transformed_df.count()}\") \n",
    "    spark.stop()\n",
    "if __name__ == \"__main__\":\n",
    "    main()\n",
    "\n",
    "\n",
    "# transformed_df.groupBy(\"username\").count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "faa6a8cb",
   "metadata": {},
   "outputs": [
    {
     "ename": "NameError",
     "evalue": "name 'music_df' is not defined",
     "output_type": "error",
     "traceback": [
      "\u001b[31m---------------------------------------------------------------------------\u001b[39m",
      "\u001b[31mNameError\u001b[39m                                 Traceback (most recent call last)",
      "\u001b[36mCell\u001b[39m\u001b[36m \u001b[39m\u001b[32mIn[7]\u001b[39m\u001b[32m, line 2\u001b[39m\n\u001b[32m      1\u001b[39m \u001b[38;5;66;03m## Potential Test Help\u001b[39;00m\n\u001b[32m----> \u001b[39m\u001b[32m2\u001b[39m print(metadata_df(music_df).rdd.getNumPartitions())  \u001b[38;5;66;03m# Partitions for Pyspark\u001b[39;00m\n\u001b[32m      3\u001b[39m print(len(metadata_df(music_df).columns)) \u001b[38;5;66;03m# # of columns in JSON\u001b[39;00m\n\u001b[32m      4\u001b[39m null_counts = null_counts_by_column(metadata_df(music_df)) \u001b[38;5;66;03m# Null counts by column\u001b[39;00m\n\u001b[32m      5\u001b[39m null_counts.show(vertical=\u001b[38;5;28;01mTrue\u001b[39;00m) \u001b[38;5;66;03m# Print Null counts by column_df\u001b[39;00m\n",
      "\u001b[31mNameError\u001b[39m: name 'music_df' is not defined"
     ]
    }
   ],
   "source": [
    "## Potential Test Help\n",
    "print(metadata_df(music_df).rdd.getNumPartitions())  # Partitions for Pyspark\n",
    "print(len(metadata_df(music_df).columns)) # # of columns in JSON\n",
    "null_counts = null_counts_by_column(metadata_df(music_df)) # Null counts by column\n",
    "null_counts.show(vertical=True) # Print Null counts by column_df\n",
    "total_rows = music_df.count\n",
    "\n",
    "df.count()-track_df.count() ## assert the difference in rows for tracks is equal to the nulls count\n",
    "                            ## same can be done for podcasts\n",
    "\n",
    "df.show(5)\n",
    "print(f\"Original Rows {df.count()}\")\n",
    "\n",
    "track_df.show(5)\n",
    "print(f\"Track Rows {track_df.count()}\")\n",
    "\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.13.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}