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

pyspark sql

Browse Source
This commit is contained in:
CC
2026-10-18 17:39:19 +01:00
parent 73015e9230
commit 22930da424
4 changed files with 211 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files

214
spotify.ipynb
View File

@@ -2,7 +2,7 @@
"cells": [
{
"cell_type": "code",
"execution_count": 459,
"execution_count": null,
"id": "e9bf1d26",
"metadata": {},
"outputs": [],
@@ -10,12 +10,14 @@
"from pyspark.sql import SparkSession, DataFrame\n",
"from pyspark.sql import functions as F\n",
"from pathlib import Path\n",
"import glob"
"import glob\n",
"import os\n",
"from dotenv import load_dotenv, find_dotenv"
]
},
{
"cell_type": "code",
"execution_count": 460,
"execution_count": 2,
"id": "11bb5dc2",
"metadata": {},
"outputs": [],
@@ -25,11 +27,11 @@
"BASE = Path.cwd() / \"ingest\"\n",
"\n",
"def start_spark(app_name):\n",
" spark = SparkSession.builder.appName(app_name).getOrCreate()\n",
" spark.sparkContext.setLogLevel(\"ERROR\")\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",
" return spark\n",
"\n",
"\n",
"def music_source_paths():\n",
" return sorted(glob.glob(str(BASE / \"*_sesh\" / \"*.json\")))\n",
"\n",
@@ -54,7 +56,7 @@
},
{
"cell_type": "code",
"execution_count": 461,
"execution_count": 3,
"id": "a898326e",
"metadata": {},
"outputs": [],
@@ -121,7 +123,47 @@
},
{
"cell_type": "code",
"execution_count": 462,
"execution_count": 4,
"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": 5,
"id": "c609291c",
"metadata": {},
"outputs": [],
@@ -143,133 +185,89 @@
},
{
"cell_type": "code",
"execution_count": 463,
"execution_count": 6,
"id": "4cc14bd0",
"metadata": {},
"outputs": [
{
"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"
]
},
{
"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",
"| album_name| artist_name| track_name|ms_played|offline| offline_timestamp|platform| reason_end|reason_start|shuffle|skipped| track_uri| ts|username| time| date|offline_time|offline_date|\n",
"+------------------+-------------+---------------+---------+-------+-------------------+--------+--------------------+------------+-------+-------+--------------------+-------------------+--------+--------+----------+------------+------------+\n",
"|Last Day Of Summer|Summer Walker|Girls Need Love| 1.34| false|2023-04-06 22:57:10| ios|unexpected-exit-w...| trackdone| true| false|spotify:track:644...|2023-04-06 22:58:31| csm3-us|22:58:31|2023-04-06| 22:57:10| 2023-04-06|\n",
"|Last Day Of Summer|Summer Walker|Girls Need Love| 0.17| false|2023-04-07 12:39:00| ios| endplay| appload| true| true|spotify:track:644...|2023-04-07 12:39:14| csm3-us|12:39:14|2023-04-07| 12:39:00| 2023-04-07|\n",
"+------------------+-------------+---------------+---------+-------+-------------------+--------+--------------------+------------+-------+-------+--------------------+-------------------+--------+--------+----------+------------+------------+\n",
"only showing top 2 rows\n",
"root\n",
" |-- album_name: string (nullable = true)\n",
" |-- artist_name: string (nullable = true)\n",
" |-- track_name: string (nullable = true)\n",
" |-- ms_played: double (nullable = true)\n",
" |-- offline: boolean (nullable = true)\n",
" |-- offline_timestamp: timestamp (nullable = true)\n",
" |-- platform: string (nullable = true)\n",
" |-- reason_end: string (nullable = true)\n",
" |-- reason_start: string (nullable = true)\n",
" |-- shuffle: boolean (nullable = true)\n",
" |-- skipped: boolean (nullable = true)\n",
" |-- track_uri: string (nullable = true)\n",
" |-- ts: timestamp (nullable = true)\n",
" |-- username: string (nullable = true)\n",
" |-- time: string (nullable = true)\n",
" |-- date: string (nullable = true)\n",
" |-- offline_time: string (nullable = true)\n",
" |-- offline_date: string (nullable = true)\n",
"\n",
" Raw DF Rows: 105477\n",
" Transformed DF Rows: 102626\n",
" Row difference is 2851\n"
"/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"
]
}
],
"source": [
" ## MAIN ## \n",
"\n",
"app_name = \"ml_media\"\n",
"spark = start_spark(app_name)\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",
"# raw_df = read_json(spark, music_source_paths())\n",
"# raw_df.show(1)\n",
"\n",
"transformed_df = clean_track_df(raw_df)\n",
"transformed_df.show(2)\n",
"transformed_df.printSchema()\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",
"# 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",
" 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",
" spark.stop\n",
"if __name__ == \"__main__\":\n",
" main()\n",
"\n",
"spark.stop()\n",
" \n",
"\n",
"# transformed_df.groupBy(\"username\").count().show()"
]
},
{
"cell_type": "code",
"execution_count": 215,
"execution_count": 7,
"id": "faa6a8cb",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"7\n",
"21\n",
"-RECORD 0----------------------\n",
" conn_country | 0 \n",
" episode_name | 105093 \n",
" episode_show_name | 105093 \n",
" incognito_mode | 0 \n",
" ip_addr_decrypted | 4492 \n",
" album_name | 2851 \n",
" artist_name | 2851 \n",
" track_name | 2851 \n",
" ms_played | 0 \n",
" offline | 279 \n",
" offline_timestamp | 2 \n",
" platform | 0 \n",
" reason_end | 4490 \n",
" reason_start | 0 \n",
" shuffle | 0 \n",
" skipped | 44457 \n",
" spotify_episode_uri | 105093 \n",
" spotify_track_uri | 2851 \n",
" ts | 0 \n",
" user_agent_decrypted | 24700 \n",
" username | 0 \n",
"\n",
"+------------+------------+-----------------+--------------+-----------------+--------------------------------+---------------------------------+--------------------------+---------+-------+-----------------+--------+--------------------+------------+-------+-------+-------------------+--------------------+--------------------+--------------------+--------+\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",
"| GB| NULL| NULL| false| 31.94.22.158| Last Day Of Summer| Summer Walker| Girls Need Love| 10100| false| 1680867540| ios| endplay| appload| true| true| NULL|spotify:track:644...|2023-04-07T11:39:14Z| unknown| csm3-us|\n",
"| GB| NULL| NULL| false| 31.94.22.158| Famouz| JHAYCO| No Me Conoce - Remix| 309120| false| 1680867555| ios| trackdone| clickrow| true| false| NULL|spotify:track:2be...|2023-04-07T11:45:50Z| unknown| csm3-us|\n",
"| GB| NULL| NULL| false| 31.94.22.158| 4K| El Alfa| 4K| 201500| false| 1680867951| ios| trackdone| trackdone| true| false| NULL|spotify:track:4xI...|2023-04-07T11:49:12Z| unknown| csm3-us|\n",
"| GB| NULL| NULL| false| 31.94.22.158| Sauce Boyz 2| Eladio Carrion| No Te Deseo el Mal| 18282| false| 1680868153| ios| fwdbtn| trackdone| true| true| NULL|spotify:track:3HQ...|2023-04-07T11:49:32Z| unknown| csm3-us|\n",
"+------------+------------+-----------------+--------------+-----------------+--------------------------------+---------------------------------+--------------------------+---------+-------+-----------------+--------+--------------------+------------+-------+-------+-------------------+--------------------+--------------------+--------------------+--------+\n",
"only showing top 5 rows\n",
"Original Rows 105477\n",
"+------------+------------+-----------------+--------------+------------------+--------------+--------------------+---------+-------+-----------------+--------+--------------------+------------+-------+-------+-----------+--------------------+--------------------+--------+\n",
"|conn_country|episode_name|episode_show_name|incognito_mode| album_name| artist_name| track_name|ms_played|offline|offline_timestamp|platform| reason_end|reason_start|shuffle|skipped|episode_uri| spotify_track_uri| ts|username|\n",
"+------------+------------+-----------------+--------------+------------------+--------------+--------------------+---------+-------+-----------------+--------+--------------------+------------+-------+-------+-----------+--------------------+--------------------+--------+\n",
"| GB| NULL| NULL| false|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| csm3-us|\n",
"| GB| NULL| NULL| false|Last Day Of Summer| Summer Walker| Girls Need Love| 10100| false| 1680867540| ios| endplay| appload| true| true| NULL|spotify:track:644...|2023-04-07T11:39:14Z| csm3-us|\n",
"| GB| NULL| NULL| false| Famouz| JHAYCO|No Me Conoce - Remix| 309120| false| 1680867555| ios| trackdone| clickrow| true| false| NULL|spotify:track:2be...|2023-04-07T11:45:50Z| csm3-us|\n",
"| GB| NULL| NULL| false| 4K| El Alfa| 4K| 201500| false| 1680867951| ios| trackdone| trackdone| true| false| NULL|spotify:track:4xI...|2023-04-07T11:49:12Z| csm3-us|\n",
"| GB| NULL| NULL| false| Sauce Boyz 2|Eladio Carrion| No Te Deseo el Mal| 18282| false| 1680868153| ios| fwdbtn| trackdone| true| true| NULL|spotify:track:3HQ...|2023-04-07T11:49:32Z| csm3-us|\n",
"+------------+------------+-----------------+--------------+------------------+--------------+--------------------+---------+-------+-----------------+--------+--------------------+------------+-------+-------+-----------+--------------------+--------------------+--------+\n",
"only showing top 5 rows\n",
"Track Rows 102626\n"
"ename": "NameError",
"evalue": "name 'music_df' is not defined",
"output_type": "error",
"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[31mNameError\u001b[39m: name 'music_df' is not defined"
]
}
],