From 7fb281f9db70da06db193d6ea2c43c1fa8500eca Mon Sep 17 00:00:00 2001 From: CC Date: Sun, 18 Oct 2026 22:55:40 +0100 Subject: [PATCH] databricks --- spotify_bronze_silver.ipynb | 62 ++++++++++++++++++++++++++++++------- spotify_silver_gold.ipynb | 41 +++++++++++++++++++----- 2 files changed, 85 insertions(+), 18 deletions(-) diff --git a/spotify_bronze_silver.ipynb b/spotify_bronze_silver.ipynb index 45610f2..6a2bfe2 100644 --- a/spotify_bronze_silver.ipynb +++ b/spotify_bronze_silver.ipynb @@ -17,7 +17,7 @@ }, { "cell_type": "code", - "execution_count": 69, + "execution_count": null, "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(\"ERROR\")\n", + " spark.log.level(\"ERROR\")\n", " return spark\n", "\n", "def music_source_paths():\n", @@ -56,7 +56,7 @@ }, { "cell_type": "code", - "execution_count": 70, + "execution_count": null, "id": "a898326e", "metadata": {}, "outputs": [], @@ -106,7 +106,7 @@ " .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('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", @@ -125,7 +125,7 @@ }, { "cell_type": "code", - "execution_count": 71, + "execution_count": null, "id": "817475cf", "metadata": {}, "outputs": [], @@ -157,10 +157,16 @@ " }\n", " return url, properties\n", "\n", - "def load_data(df, table_name):\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)" + " 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}\")" ] }, { @@ -187,7 +193,7 @@ }, { "cell_type": "code", - "execution_count": 91, + "execution_count": null, "id": "4cc14bd0", "metadata": {}, "outputs": [ @@ -229,11 +235,16 @@ " app_name = \"ml_media\"\n", " spark = start_spark(app_name)\n", "\n", - " table_name = \"spotify_data\"\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", - " raw_df.show(1)\n", + "\n", + " load_data(raw_df, raw_table_name)\n", + "\n", " transformed_df = clean_track_df(raw_df)\n", - " #load_data(transformed_df, table_name)\n", + " \n", + " load_data(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", @@ -266,6 +277,35 @@ "# transformed_df.groupBy(\"username\").count().show()" ] }, + { + "cell_type": "code", + "execution_count": null, + "id": "35878e16", + "metadata": {}, + "outputs": [], + "source": [ + "## Databricks\n", + "\n", + " ## MAIN ## \n", + "\n", + "# app_name = \"ml_media\"\n", + "\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": null, diff --git a/spotify_silver_gold.ipynb b/spotify_silver_gold.ipynb index d4086f2..2c7e1c3 100644 --- a/spotify_silver_gold.ipynb +++ b/spotify_silver_gold.ipynb @@ -17,7 +17,7 @@ }, { "cell_type": "code", - "execution_count": 14, + "execution_count": null, "id": "2c1aeb30", "metadata": {}, "outputs": [], @@ -26,7 +26,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(\"ERROR\")\n", + " spark.log.level(\"ERROR\")\n", " return spark\n", "\n", "def get_jdbc_config():\n", @@ -57,7 +57,14 @@ "\n", "def read_jdbc(spark, table_name):\n", " url, properties = get_jdbc_config()\n", - " return spark.read.jdbc(url=url, table=table_name, properties=properties)" + " return spark.read.jdbc(url=url, table=table_name, properties=properties)\n", + "\n", + "def databricks_read(df, table_name):\n", + " \"\"\"Read the data from catalog.\"\"\"\n", + " catalog = \"workspace\"\n", + " schema = \"default\"\n", + " path = f\"{catalog}.{schema}.{table_name}\"\n", + " return df.read.table(path)" ] }, { @@ -72,7 +79,7 @@ }, { "cell_type": "code", - "execution_count": 19, + "execution_count": null, "id": "f7673c72", "metadata": {}, "outputs": [ @@ -128,14 +135,34 @@ " app_name = \"ml_media_silver\"\n", " spark = start_spark(app_name)\n", "\n", - " table_name = \"spotify_data\"\n", - " silver_df = read_jdbc(spark, table_name)\n", + " bronze_table_name = \"bronze_spotify_data\"\n", + " silver_table_name = \"silver_spotify_data\"\n", + " silver_df = read_jdbc(spark, bronze_table_name)\n", "\n", " silver_df.groupBy(\"track_name\", \"username\").count().show()\n", "\n", " #spark.stop()\n", "if __name__ == \"__main__\":\n", - " main()" + " main()\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "c9a695ae", + "metadata": {}, + "outputs": [], + "source": [ + "## Databricks ##\n", + " ## MAIN ## \n", + "\n", + "# app_name = \"ml_media_silver\"\n", + "# bronze_table_name = \"bronze_spotify_data\"\n", + "# silver_table_name = \"silver_spotify_data\"\n", + "# silver_df = databricks_read(spark, bronze_table_name)\n", + "# silver_df.show(1)\n", + "\n", + "\n" ] } ],