CC/media_ml
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

databricks

Browse Source
This commit is contained in:
CC
2026-10-18 22:55:40 +01:00
parent 7b6803b1ee
commit 7fb281f9db
2 changed files with 85 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
spotify_bronze_silver.ipynb
View File

@@ -17,7 +17,7 @@
}, },
{ {
"cell_type": "code", "cell_type": "code",
"execution_count": 69, "execution_count": null,
"id": "11bb5dc2", "id": "11bb5dc2",
"metadata": {}, "metadata": {},
"outputs": [], "outputs": [],
@@ -29,7 +29,7 @@
"def start_spark(app_name):\n", "def start_spark(app_name):\n",
" jar_path = str(Path.cwd() / \"postgresql.jar\")\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 = 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", " return spark\n",
"\n", "\n",
"def music_source_paths():\n", "def music_source_paths():\n",
@@ -56,7 +56,7 @@
}, },
{ {
"cell_type": "code", "cell_type": "code",
"execution_count": 70, "execution_count": null,
"id": "a898326e", "id": "a898326e",
"metadata": {}, "metadata": {},
"outputs": [], "outputs": [],
@@ -106,7 +106,7 @@
" .withColumn(\"offline_timestamped\", 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('time', F.date_format('ts', 'HH:mm:ss'))\n",
" .withColumn('date', F.date_format('ts', 'yyyy-MM-dd'))\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('hour', F.hour('ts'))\n",
" .withColumn('offline_time', F.date_format('offline_timestamped', 'HH:mm:ss'))\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", " .withColumn('offline_date', F.date_format('offline_timestamped', 'yyyy-MM-dd')) \n",
@@ -125,7 +125,7 @@
}, },
{ {
"cell_type": "code", "cell_type": "code",
"execution_count": 71, "execution_count": null,
"id": "817475cf", "id": "817475cf",
"metadata": {}, "metadata": {},
"outputs": [], "outputs": [],
@@ -157,10 +157,16 @@
" }\n", " }\n",
" return url, properties\n", " return url, properties\n",
"\n", "\n",
"def load_data(df, table_name):\n", "def jdbc_write(df, table_name):\n",
" \"\"\"Load the data into the database.\"\"\"\n", " \"\"\"Load the data into the database.\"\"\"\n",
" url, properties = get_jdbc_config()\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", "cell_type": "code",
"execution_count": 91, "execution_count": null,
"id": "4cc14bd0", "id": "4cc14bd0",
"metadata": {}, "metadata": {},
"outputs": [ "outputs": [
@@ -229,11 +235,16 @@
" app_name = \"ml_media\"\n", " app_name = \"ml_media\"\n",
" spark = start_spark(app_name)\n", " spark = start_spark(app_name)\n",
"\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 = 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", " 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", " # transformed_df.show(5)\n",
" # skip_reason_df = transformed_df.groupBy(\"skipped\", \"reason_end\").count().orderBy(F.desc(\"reason_end\")).show()\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", " # reason_df = transformed_df.groupBy(\"reason_end\").orderBy(F.desc(\"count\")).show(5)\n",
@@ -266,6 +277,35 @@
"# transformed_df.groupBy(\"username\").count().show()" "# 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", "cell_type": "code",
"execution_count": null, "execution_count": null,

41
spotify_silver_gold.ipynb
View File

@@ -17,7 +17,7 @@
}, },
{ {
"cell_type": "code", "cell_type": "code",
"execution_count": 14, "execution_count": null,
"id": "2c1aeb30", "id": "2c1aeb30",
"metadata": {}, "metadata": {},
"outputs": [], "outputs": [],
@@ -26,7 +26,7 @@
"def start_spark(app_name):\n", "def start_spark(app_name):\n",
" jar_path = str(Path.cwd() / \"postgresql.jar\")\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 = 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", " return spark\n",
"\n", "\n",
"def get_jdbc_config():\n", "def get_jdbc_config():\n",
@@ -57,7 +57,14 @@
"\n", "\n",
"def read_jdbc(spark, table_name):\n", "def read_jdbc(spark, table_name):\n",
" url, properties = get_jdbc_config()\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", "cell_type": "code",
"execution_count": 19, "execution_count": null,
"id": "f7673c72", "id": "f7673c72",
"metadata": {}, "metadata": {},
"outputs": [ "outputs": [
@@ -128,14 +135,34 @@
" app_name = \"ml_media_silver\"\n", " app_name = \"ml_media_silver\"\n",
" spark = start_spark(app_name)\n", " spark = start_spark(app_name)\n",
"\n", "\n",
" table_name = \"spotify_data\"\n", " bronze_table_name = \"bronze_spotify_data\"\n",
" silver_df = read_jdbc(spark, table_name)\n", " silver_table_name = \"silver_spotify_data\"\n",
" silver_df = read_jdbc(spark, bronze_table_name)\n",
"\n", "\n",
" silver_df.groupBy(\"track_name\", \"username\").count().show()\n", " silver_df.groupBy(\"track_name\", \"username\").count().show()\n",
"\n", "\n",
" #spark.stop()\n", " #spark.stop()\n",
"if __name__ == \"__main__\":\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"
] ]
} }
], ],