From bb36c54f961816abfb3cbcb9243dfaf68a0854c3 Mon Sep 17 00:00:00 2001 From: CC Date: Sun, 18 Oct 2026 18:03:05 +0100 Subject: [PATCH] bronze -> silver postgres --- spotify.ipynb | 69 ++++++++++++++++----------------------------------- 1 file changed, 22 insertions(+), 47 deletions(-) diff --git a/spotify.ipynb b/spotify.ipynb index 2acb52b..a53b8ed 100644 --- a/spotify.ipynb +++ b/spotify.ipynb @@ -2,7 +2,7 @@ "cells": [ { "cell_type": "code", - "execution_count": null, + "execution_count": 8, "id": "e9bf1d26", "metadata": {}, "outputs": [], @@ -17,7 +17,7 @@ }, { "cell_type": "code", - "execution_count": 2, + "execution_count": 9, "id": "11bb5dc2", "metadata": {}, "outputs": [], @@ -29,7 +29,7 @@ "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(\"WARN\")\n", + " spark.sparkContext.setLogLevel(\"ERROR\")\n", " return spark\n", "\n", "def music_source_paths():\n", @@ -56,7 +56,7 @@ }, { "cell_type": "code", - "execution_count": 3, + "execution_count": null, "id": "a898326e", "metadata": {}, "outputs": [], @@ -103,18 +103,18 @@ "def add_time_parts(df: DataFrame, ts_col) -> DataFrame:\n", " return (df\n", " .withColumn(\"ts\", F.to_timestamp(\"ts\"))\n", - " .withColumn(\"offline_timestamp\", F.to_timestamp(\"offline_timestamp\"))\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_timestamp', 'HH:mm:ss'))\n", - " .withColumn('offline_date', F.date_format('offline_timestamp', '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_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", @@ -123,7 +123,7 @@ }, { "cell_type": "code", - "execution_count": 4, + "execution_count": 11, "id": "817475cf", "metadata": {}, "outputs": [], @@ -163,7 +163,7 @@ }, { "cell_type": "code", - "execution_count": 5, + "execution_count": 12, "id": "c609291c", "metadata": {}, "outputs": [], @@ -185,7 +185,7 @@ }, { "cell_type": "code", - "execution_count": 6, + "execution_count": 13, "id": "4cc14bd0", "metadata": {}, "outputs": [ @@ -193,58 +193,33 @@ "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:36:31 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:36:31 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address\n", - "26/10/18 17:36:32 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": [ - "/Users/carlacastillo/Documents/Gitea/media_ml/postgresql.jar\n" - ] - }, - { - "name": "stderr", - "output_type": "stream", - "text": [ - "26/10/18 17:36:36 WARN SparkStringUtils: Truncated the string representation of a plan since it was too large. This behavior can be adjusted by setting 'spark.sql.debug.maxToStringFields'.\n", - " \r" + " Raw DF Rows: 105477\n", + " Transformed DF Rows: 102626\n", + " Row difference is 2851\n" ] } ], "source": [ " ## MAIN ## \n", - "\n", - "# app_name = \"ml_media\"\n", - "# spark = start_spark(app_name)\n", - "\n", - "# raw_df = read_json(spark, music_source_paths())\n", - "# raw_df.show(1)\n", - "\n", - "# transformed_df = clean_track_df(raw_df)\n", - "# # print(transformed_df.tail(5))\n", - "# # transformed_df.printSchema()\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", - "\n", - "\n", "def main():\n", " app_name = \"ml_media\"\n", " spark = start_spark(app_name)\n", - " print(spark.conf.get(\"spark.jars\", \"NOT SET\"))\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", @@ -255,7 +230,7 @@ }, { "cell_type": "code", - "execution_count": 7, + "execution_count": 14, "id": "faa6a8cb", "metadata": {}, "outputs": [ @@ -266,7 +241,7 @@ "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[36mCell\u001b[39m\u001b[36m \u001b[39m\u001b[32mIn[14]\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" ] }