This commit is contained in:
CC
2026-10-18 01:04:48 +01:00
parent a8fba1b143
commit ad26b35c9b

View File

@@ -490,49 +490,125 @@
}, },
{ {
"cell_type": "code", "cell_type": "code",
"execution_count": null, "execution_count": 96,
"id": "a0e9641e", "id": "a0e9641e",
"metadata": {}, "metadata": {},
"outputs": [ "outputs": [
{ {
"ename": "SyntaxError", "name": "stdout",
"evalue": "expected ':' (3030084487.py, line 14)", "output_type": "stream",
"output_type": "error", "text": [
"traceback": [ "root\n",
" \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" " |-- 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": [ "source": [
"from pyspark.sql import SparkSession\n", "from pyspark.sql import SparkSession\n",
"import os \n", "from pyspark.sql import functions as F\n",
"from pathlib import Path\n",
"import glob\n", "import glob\n",
"\n", "\n",
"BASE = Path.cwd() / \"ingest\"\n",
"\n",
"def music_source_paths():\n", "def music_source_paths():\n",
" music_path = os.getcwd() + \"/ingest/*_sesh/*.json\"\n", " return sorted(glob.glob(str(BASE / \"*_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",
"\n", "\n",
"# def music_df(spark, music_files):\n", "def video_source_paths():\n",
"# for music_file in music_files:\n", " return sorted(glob.glob(str(BASE / \"trakt_*\" / \"*.json\")))\n",
"# if music_file.endswith(\".json\")\n",
"\n", "\n",
"def video_ingest_path(file_path):\n", "def read_json(spark, paths):\n",
" video_path = os.getcwd() + \"/ingest/trakt_*\"\n", " if not paths:\n",
" video_files = []\n", " raise FileNotFoundError(\"No JSON files matched the pattern\")\n",
" for file in glob.iglob(video_path, recursive=True):\n", " return (\n",
" return(file)\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", "\n",
"spark = SparkSession.builder.appName(\"media_ml\").getOrCreate()\n", "spark = SparkSession.builder.appName(\"media_ml\").getOrCreate()\n",
"spark.sparkContext.setLogLevel(\"INFO\")\n", "spark.sparkContext.setLogLevel(\"WARN\")\n",
"\n", "\n",
"music_json = music_source_paths()\n", "music_df = read_json(spark, music_source_paths())\n",
"print(music_json)\n",
"music_df = spark.read.json(*music_json)\n",
"\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"
] ]
} }
], ],