372 lines
15 KiB
Plaintext
372 lines
15 KiB
Plaintext
{
|
|
"cells": [
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": 14,
|
|
"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": null,
|
|
"id": "11bb5dc2",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
" ## Extract ##\n",
|
|
"\n",
|
|
"BASE = Path.cwd() / \"ingest\"\n",
|
|
"\n",
|
|
"def start_spark(app_name):\n",
|
|
" \"\"\"Initialize and return a SparkSession with the PostgreSQL JDBC driver.\"\"\"\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",
|
|
" return spark\n",
|
|
"\n",
|
|
"def music_source_paths():\n",
|
|
" \"\"\"Return sorted list of Spotify listening history JSON file paths.\"\"\"\n",
|
|
" return sorted(glob.glob(str(BASE / \"*_sesh\" / \"*.json\")))\n",
|
|
"\n",
|
|
"def video_source_paths():\n",
|
|
" \"\"\"Return sorted list of Trakt video watching history JSON file paths.\"\"\"\n",
|
|
" return sorted(glob.glob(str(BASE / \"trakt_*\" / \"*.json\")))\n",
|
|
"\n",
|
|
"def read_json(spark, paths):\n",
|
|
" \"\"\"Read a list of JSON files into a multi-line Spark DataFrame.\"\"\"\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": null,
|
|
"id": "a898326e",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
" ## Transform ##\n",
|
|
"def null_counts_by_column(df):\n",
|
|
" \"\"\"Count nulls in each column of the DataFrame.\"\"\"\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",
|
|
" \"\"\"Rename verbose album/artist/track metadata columns to short names.\"\"\"\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",
|
|
" \"\"\"Drop specified columns and rename remaining ones per the rename map.\"\"\"\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",
|
|
" \"\"\"Filter to rows where track_name is not null.\"\"\"\n",
|
|
" return df.where(df[\"track_name\"].isNotNull())\n",
|
|
"\n",
|
|
"def parse_timestamps(df: DataFrame, columns) -> DataFrame:\n",
|
|
" \"\"\"Cast each column in the list to timestamp type.\"\"\"\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",
|
|
" \"\"\"Cast each column in the list to string type.\"\"\"\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",
|
|
" \"\"\"Convert milliseconds played to minutes and round to 2 decimals.\"\"\"\n",
|
|
" return (df.withColumn(dest, F.round(F.col(src) / 60000, 2)))\n",
|
|
"\n",
|
|
"def add_time_parts(df: DataFrame, ts_col) -> DataFrame:\n",
|
|
" \"\"\"Add time/date/day-of-week/hour parts derived from the timestamp columns.\"\"\"\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",
|
|
" \"\"\"Apply the full transform pipeline: rename, filter tracks, add minutes and time parts.\"\"\"\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": 17,
|
|
"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 jdbc_write(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)\n",
|
|
"\n",
|
|
"def databricks_write(df, table_name):\n",
|
|
" \"\"\"Load the data into bronze catalog.\"\"\"\n",
|
|
" catalog = \"workspace\"\n",
|
|
" schema = \"default\"\n",
|
|
" return df.write.format(\"delta\").mode(\"overwrite\").saveAsTable(f\"{catalog}.{schema}.{table_name}\")"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": 18,
|
|
"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": 19,
|
|
"id": "4cc14bd0",
|
|
"metadata": {},
|
|
"outputs": [
|
|
{
|
|
"name": "stderr",
|
|
"output_type": "stream",
|
|
"text": [
|
|
" \r"
|
|
]
|
|
},
|
|
{
|
|
"name": "stdout",
|
|
"output_type": "stream",
|
|
"text": [
|
|
"+----------------------------+--------------+\n",
|
|
"|reason_end |skipped_values|\n",
|
|
"+----------------------------+--------------+\n",
|
|
"|NULL |[] |\n",
|
|
"|backbtn |[true] |\n",
|
|
"|endplay |[true] |\n",
|
|
"|fwdbtn |[true] |\n",
|
|
"|logout |[false] |\n",
|
|
"|remote |[false] |\n",
|
|
"|trackdone |[false] |\n",
|
|
"|trackerror |[false] |\n",
|
|
"|unexpected-exit |[false] |\n",
|
|
"|unexpected-exit-while-paused|[false] |\n",
|
|
"|unknown |[false] |\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",
|
|
" spark.sparkContext.setLogLevel(\"ERROR\")\n",
|
|
"\n",
|
|
" raw_table_name = \"raw_spotify_data\"\n",
|
|
" bronze_table_name = \"bronze_spotify_data\"\n",
|
|
" silver_table_name = \"silver_spotify_data\"\n",
|
|
" \n",
|
|
" raw_df = read_json(spark, music_source_paths())\n",
|
|
"\n",
|
|
" jdbc_write(raw_df, raw_table_name)\n",
|
|
"\n",
|
|
" transformed_df = clean_track_df(raw_df)\n",
|
|
" \n",
|
|
" jdbc_write(transformed_df, bronze_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\", \"skipped\").distinct().orderBy(\"reason_end\").show(100)\n",
|
|
" skip_reason_bool = (transformed_df.groupBy(\"reason_end\")\n",
|
|
" .agg(F.collect_set(\"skipped\").alias(\"skipped_values\"))\n",
|
|
" .orderBy(\"reason_end\")\n",
|
|
" .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": 20,
|
|
"id": "35878e16",
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"## Databricks\n",
|
|
"\n",
|
|
" ## MAIN ## \n",
|
|
"\n",
|
|
"# app_name = \"ml_media\"\n",
|
|
"# spark.log.level(\"ERROR\")\n",
|
|
"\n",
|
|
"# raw_table_name = \"raw_spotify_data\"\n",
|
|
"# bronze_table_name = \"bronze_spotify_data\"\n",
|
|
"# silver_table_name = \"silver_spotify_data\"\n",
|
|
"# raw_df = read_json(spark, music_source_paths())\n",
|
|
"# databricks_write(raw_df, raw_table_name)\n",
|
|
"# transformed_df = clean_track_df(raw_df)\n",
|
|
"\n",
|
|
"# databricks_write(transformed_df, bronze_table_name)\n",
|
|
"\n",
|
|
"# skip_reason_bool = (transformed_df.groupBy(\"reason_end\")\n",
|
|
"# .agg(F.collect_set(\"skipped\").alias(\"skipped_values\"))\n",
|
|
"# .orderBy(\"reason_end\")\n",
|
|
"# .show(truncate=False))\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": 21,
|
|
"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
|
|
}
|