diff --git a/spotify.ipynb b/spotify.ipynb index 4f5c3e3..21aa187 100644 --- a/spotify.ipynb +++ b/spotify.ipynb @@ -490,49 +490,125 @@ }, { "cell_type": "code", - "execution_count": null, + "execution_count": 96, "id": "a0e9641e", "metadata": {}, "outputs": [ { - "ename": "SyntaxError", - "evalue": "expected ':' (3030084487.py, line 14)", - "output_type": "error", - "traceback": [ - " \u001b[36mCell\u001b[39m\u001b[36m \u001b[39m\u001b[32mIn[29]\u001b[39m\u001b[32m, line 14\u001b[39m\n\u001b[31m \u001b[39m\u001b[31mif music_file.endswith(\".json\")\u001b[39m\n ^\n\u001b[31mSyntaxError\u001b[39m\u001b[31m:\u001b[39m expected ':'\n" + "name": "stdout", + "output_type": "stream", + "text": [ + "root\n", + " |-- conn_country: string (nullable = true)\n", + " |-- episode_name: string (nullable = true)\n", + " |-- episode_show_name: string (nullable = true)\n", + " |-- incognito_mode: boolean (nullable = true)\n", + " |-- ip_addr_decrypted: string (nullable = true)\n", + " |-- master_metadata_album_album_name: string (nullable = true)\n", + " |-- master_metadata_album_artist_name: string (nullable = true)\n", + " |-- master_metadata_track_name: string (nullable = true)\n", + " |-- ms_played: long (nullable = true)\n", + " |-- offline: boolean (nullable = true)\n", + " |-- offline_timestamp: long (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", + " |-- spotify_episode_uri: string (nullable = true)\n", + " |-- spotify_track_uri: string (nullable = true)\n", + " |-- ts: string (nullable = true)\n", + " |-- user_agent_decrypted: string (nullable = true)\n", + " |-- username: string (nullable = true)\n", + "\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" ] } ], "source": [ "from pyspark.sql import SparkSession\n", - "import os \n", - "import glob \n", + "from pyspark.sql import functions as F\n", + "from pathlib import Path\n", + "import glob\n", + "\n", + "BASE = Path.cwd() / \"ingest\"\n", "\n", "def music_source_paths():\n", - " music_path = os.getcwd() + \"/ingest/*_sesh/*.json\"\n", - " music_files = []\n", - " for file in glob.iglob(music_path, recursive=True):\n", - " music_files.append(file)\n", - " return(music_files)\n", + " return sorted(glob.glob(str(BASE / \"*_sesh\" / \"*.json\")))\n", "\n", - "# def music_df(spark, music_files):\n", - "# for music_file in music_files:\n", - "# if music_file.endswith(\".json\")\n", + "def video_source_paths():\n", + " return sorted(glob.glob(str(BASE / \"trakt_*\" / \"*.json\")))\n", "\n", - "def video_ingest_path(file_path):\n", - " video_path = os.getcwd() + \"/ingest/trakt_*\"\n", - " video_files = []\n", - " for file in glob.iglob(video_path, recursive=True):\n", - " return(file)\n", + "def read_json(spark, paths):\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", + " )\n", + "\n", + "\n", + "def null_counts_by_column(df):\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", + " 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", + "## MAIN ## \n", "\n", "spark = SparkSession.builder.appName(\"media_ml\").getOrCreate()\n", - "spark.sparkContext.setLogLevel(\"INFO\")\n", + "spark.sparkContext.setLogLevel(\"WARN\")\n", "\n", - "music_json = music_source_paths()\n", - "print(music_json)\n", - "music_df = spark.read.json(*music_json)\n", + "music_df = read_json(spark, music_source_paths())\n", "\n", - "music_df.show(5)" + "music_df.printSchema()\n", + "# music_df.show(10)\n", + "\n", + "\n", + "null_counts = null_counts_by_column(metadata_df(music_df))\n", + "null_counts.show(vertical=True, truncate=False)\n", + "\n", + "\n", + "# total_rows = music_df.count\n", + "# print(music_df.count())\n", + "\n", + "\n", + "# print(group)\n", + "# display(group)\n", + "\n" ] } ],