{ "cells": [ { "cell_type": "code", "execution_count": 68, "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": 69, "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": 70, "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('day of week', F.day('ts'))\n", " .withColumn('hour', F.hour('ts'))\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": 71, "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": 72, "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": "stdout", "output_type": "stream", "text": [ "+------------+------------+-----------------+--------------+-----------------+--------------------------------+---------------------------------+--------------------------+---------+-------+-----------------+--------+--------------------+------------+-------+-------+-------------------+--------------------+--------------------+--------------------+--------+\n", "|conn_country|episode_name|episode_show_name|incognito_mode|ip_addr_decrypted|master_metadata_album_album_name|master_metadata_album_artist_name|master_metadata_track_name|ms_played|offline|offline_timestamp|platform| reason_end|reason_start|shuffle|skipped|spotify_episode_uri| spotify_track_uri| ts|user_agent_decrypted|username|\n", "+------------+------------+-----------------+--------------+-----------------+--------------------------------+---------------------------------+--------------------------+---------+-------+-----------------+--------+--------------------+------------+-------+-------+-------------------+--------------------+--------------------+--------------------+--------+\n", "| GB| NULL| NULL| false| 31.94.22.158| Last Day Of Summer| Summer Walker| Girls Need Love| 80505| false| 1680818230| ios|unexpected-exit-w...| trackdone| true| false| NULL|spotify:track:644...|2023-04-06T21:58:31Z| unknown| csm3-us|\n", "+------------+------------+-----------------+--------------+-----------------+--------------------------------+---------------------------------+--------------------------+---------+-------+-----------------+--------+--------------------+------------+-------+-------+-------------------+--------------------+--------------------+--------------------+--------+\n", "only showing top 1 row\n", "+-------+--------------------+-----+\n", "|skipped| reason_end|count|\n", "+-------+--------------------+-----+\n", "| false| unknown| 99|\n", "| NULL| unknown| 375|\n", "| false|unexpected-exit-w...| 1243|\n", "| NULL|unexpected-exit-w...| 432|\n", "| false| unexpected-exit| 36|\n", "| NULL| unexpected-exit| 51|\n", "| false| trackerror| 69|\n", "| NULL| trackerror| 33|\n", "| false| trackdone|36102|\n", "| NULL| trackdone|24077|\n", "| false| remote| 121|\n", "| NULL| remote| 859|\n", "| false| logout| 491|\n", "| NULL| logout| 745|\n", "| true| fwdbtn|16571|\n", "| NULL| fwdbtn|11445|\n", "| true| endplay| 2943|\n", "| NULL| endplay| 1447|\n", "| true| backbtn| 650|\n", "| NULL| backbtn| 402|\n", "+-------+--------------------+-----+\n", "only showing top 20 rows\n", "+----------------------------+-----+\n", "|reason_end |count|\n", "+----------------------------+-----+\n", "|trackdone |60179|\n", "|fwdbtn |28016|\n", "|NULL |4435 |\n", "|endplay |4390 |\n", "|unexpected-exit-while-paused|1675 |\n", "|logout |1236 |\n", "|backbtn |1052 |\n", "|remote |980 |\n", "|unknown |474 |\n", "|trackerror |102 |\n", "|unexpected-exit |87 |\n", "+----------------------------+-----+\n", "\n", " 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", " raw_df.show(1)\n", " transformed_df = clean_track_df(raw_df)\n", " #load_data(transformed_df, table_name)\n", " # transformed_df.show(5)\n", " skip_reason_df = transformed_df.groupBy(\"skipped\", \"reason_end\").count().orderBy(F.desc(\"reason_end\")).show()\n", " # reason_df = transformed_df.groupBy(\"reason_end\").orderBy(F.desc(\"count\")).show(5)\n", " # skip_df = transformed_df.groupBy(\"skipped\").count().orderBy(F.desc(\"count\")).show(5)\n", " skip_join_bool = transformed_df.select(\"reason_end\", \"was_skipped\").distinct().orderBy(\"reason_end\").show(truncate=False))\n", " #join(transformed_df.F.col(\"reason_end\"))\n", "\n", " # missing_boolean = set(rename_map) - set(df.columns)\n", "\n", "\n", "# Assuming 'df' is your DataFrame and 'timestamp_column' is the column with timestamps\n", " # df = df.withColumn(\"time_of_day\", \n", " # F.when((F.hour(\"timestamp_column\") >= 5) & (F.hour(\"timestamp_column\") < 12), \"morning\")\n", " # .when((F.hour(\"timestamp_column\") >= 12) & (F.hour(\"timestamp_column\") < 18), \"afternoon\")\n", " # .when((F.hour(\"timestamp_column\") >= 18) & (F.hour(\"timestamp_column\") < 22), \"evening\")\n", " # .otherwise(\"night\"))\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": null, "id": "faa6a8cb", "metadata": {}, "outputs": [], "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 }