analysis.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Setup"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# pyspark imports\n",
    "from pyspark.sql import Window\n",
    "from pyspark.sql import functions\n",
    "import pyspark.sql.functions as F\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.ml.feature import StringIndexer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# regular imports\n",
    "import os"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "24/04/26 08:47:40 WARN Utils: Your hostname, cie-B760M-DS3H-DDR4 resolves to a loopback address: 127.0.1.1; using 172.16.171.63 instead (on interface enxac15a2afb0c0)\n",
      "24/04/26 08:47:40 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n",
      "24/04/26 08:47:40 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - in-memory</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://172.16.171.63:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v3.4.3</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>ccbd-demo</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x7fb308827730>"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark = SparkSession.builder.appName('ccbd-demo').master('local').getOrCreate()\n",
    "spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# folder paths\n",
    "data_folder = \"data/\"\n",
    "output_folder = \"outputs/\"\n",
    "graphs_folder = \"graphs/\"\n",
    "\n",
    "# file paths\n",
    "dataset_path = data_folder + \"anime_data.csv\"\n",
    "\n",
    "if not os.path.exists(dataset_path):\n",
    "  raise Exception(f\"{dataset_path} does not exist!\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = spark.read.csv(dataset_path, header = True)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Data Preprocessing"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**DATASET DESCRIPTION**\n",
    "\n",
    "| Column     | Description                                                                         |\n",
    "|------------|-------------------------------------------------------------------------------------|\n",
    "| username   | Username of the user who interacted with the anime (might be null if not logged in) |\n",
    "| anime_id   | Unique identifier for the anime entry                                               |\n",
    "| my_score   | User's personal score for the anime (might be null if not rated)                    |\n",
    "| user_id    | Unique identifier for the user profile (might be null if not logged in)             |\n",
    "| gender     | User's reported gender (might be null if not provided) (Female/Non-Binary/Male)     |\n",
    "| title      | Official title of the anime                                                         |\n",
    "| type       | Type of anime (e.g., TV, Movie, OVA, etc.)                                          |\n",
    "| source     | Source material of the anime (e.g., Manga, Light Novel, etc.)                       |\n",
    "| score      | Average score for the anime from all users (might be null if not enough ratings)    |\n",
    "| scored_by  | Number of users who rated the anime (might be null)                                 |\n",
    "| rank       | Anime's ranking based on either score or popularity (might be null)                 |\n",
    "| popularity | Anime's popularity ranking on MyAnimeList (might be null)                           |\n",
    "| genre      | Genre(s) of the anime (might be a single string or comma-separated list)            |\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# get summary + peek dataset\n",
    "print(\"Dataset size:\", df.count())\n",
    "df.head(10)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. NULL value inspection\n",
    "- There is a problem in the 'score' column\n",
    "- Before casting there are no null values\n",
    "- After casting to float there are ~8000 null values\n",
    "- This is due to strings like \"Game\" or \"Visual Novel\" being placed in the 'score' column\n",
    "- Null values MUST be calculated after casting"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# inspect rows where 'score' is null (before casting to float)\n",
    "score_null = df.filter(df['score'].isNull())\n",
    "score_null.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.select('score').distinct().write.csv(output_folder + \"distinct_scores.csv\", header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "value_to_check = \"Visual novel\"\n",
    "problem_df = df.filter(df['score'] == value_to_check)\n",
    "count = problem_df.count()\n",
    "print(\"Number of times '{}' appears in the 'score' column: {}\".format(value_to_check, count))"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Dealing with NULL values\n",
    "\n",
    "**NULL VALUE COUNTS**\n",
    "\n",
    "| Column        | Count  |\n",
    "|---------------|--------|\n",
    "| username      | 256    |\n",
    "| anime_id      | 0      |\n",
    "| my_score      | 0      |\n",
    "| user_id       | 0      |\n",
    "| gender        | 0      |\n",
    "| title         | 0      |\n",
    "| type          | 0      |\n",
    "| score         | 8160   |\n",
    "| scored_by     | 0      |\n",
    "| rank          | 751600 |\n",
    "| popularity    | 370    |\n",
    "| genre         | 2267   |"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(username='karthiga', anime_id=21, my_score=9, user_id=2255153, gender='Female', title='One Piece', type='TV', source='Manga', score=8.539999961853027, scored_by=423868, rank=91, popularity=35, genre='Action, Adventure, Comedy, Super Power, Drama, Fantasy, Shounen'),\n",
       " Row(username='karthiga', anime_id=59, my_score=7, user_id=2255153, gender='Female', title='Chobits', type='TV', source='Manga', score=7.53000020980835, scored_by=175388, rank=1546, popularity=188, genre='Sci-Fi, Comedy, Drama, Romance, Ecchi, Seinen'),\n",
       " Row(username='karthiga', anime_id=74, my_score=7, user_id=2255153, gender='Female', title='Gakuen Alice', type='TV', source='Manga', score=7.769999980926514, scored_by=33244, rank=941, popularity=1291, genre='Comedy, School, Shoujo, Super Power'),\n",
       " Row(username='karthiga', anime_id=120, my_score=7, user_id=2255153, gender='Female', title='Fruits Basket', type='TV', source='Manga', score=7.769999980926514, scored_by=167968, rank=939, popularity=222, genre='Slice of Life, Comedy, Drama, Romance, Fantasy, Shoujo'),\n",
       " Row(username='karthiga', anime_id=178, my_score=7, user_id=2255153, gender='Female', title='Ultra Maniac', type='TV', source='Manga', score=7.260000228881836, scored_by=9663, rank=2594, popularity=2490, genre='Magic, Comedy, Romance, School, Shoujo'),\n",
       " Row(username='karthiga', anime_id=210, my_score=7, user_id=2255153, gender='Female', title='Ranma ½', type='TV', source='Manga', score=7.849999904632568, scored_by=59911, rank=802, popularity=623, genre='Slice of Life, Comedy, Martial Arts, Fantasy'),\n",
       " Row(username='karthiga', anime_id=232, my_score=6, user_id=2255153, gender='Female', title='Cardcaptor Sakura', type='TV', source='Manga', score=8.210000038146973, scored_by=121898, rank=297, popularity=292, genre='Adventure, Comedy, Drama, Magic, Romance, Fantasy, School, Shoujo'),\n",
       " Row(username='karthiga', anime_id=233, my_score=6, user_id=2255153, gender='Female', title='Daa! Daa! Daa!', type='TV', source='Manga', score=7.78000020980835, scored_by=6598, rank=919, popularity=3045, genre='Comedy, Sci-Fi, Shoujo'),\n",
       " Row(username='karthiga', anime_id=249, my_score=8, user_id=2255153, gender='Female', title='InuYasha', type='TV', source='Manga', score=7.900000095367432, scored_by=181978, rank=697, popularity=141, genre='Action, Adventure, Comedy, Historical, Demons, Supernatural, Magic, Romance, Fantasy, Shounen'),\n",
       " Row(username='karthiga', anime_id=269, my_score=10, user_id=2255153, gender='Female', title='Bleach', type='TV', source='Manga', score=7.900000095367432, scored_by=433097, rank=693, popularity=18, genre='Action, Adventure, Comedy, Super Power, Supernatural, Shounen')]"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# get the dataframe columns in their correct datatypes\n",
    "df = df.withColumn(\"anime_id\", functions.col(\"anime_id\").cast(\"int\")) \\\n",
    "\t\t.withColumn(\"my_score\", functions.col(\"my_score\").cast(\"int\")) \\\n",
    "  \t\t.withColumn(\"user_id\", functions.col(\"user_id\").cast(\"int\")) \\\n",
    "      \t.withColumn(\"score\", functions.col(\"score\").cast(\"float\")) \\\n",
    "\t\t.withColumn(\"scored_by\", functions.col(\"scored_by\").cast(\"int\")) \\\n",
    "\t\t.withColumn(\"rank\", functions.col(\"rank\").cast(\"int\")) \\\n",
    "\t\t.withColumn(\"popularity\", functions.col(\"popularity\").cast(\"int\"))\n",
    "\n",
    "# round the 'score' column to 2 decimal places\n",
    "df = df.withColumn(\"score\", functions.round(functions.col(\"score\"), 2))\n",
    "\n",
    "df.head(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# count the number of null values in each column\n",
    "null_counts = {}\n",
    "\n",
    "for col in df.columns:\n",
    "\tnull_counts[col] = df.filter(functions.col(col).isNull()).count()\n",
    "\n",
    "print(null_counts)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- Despite the 'score' column having ~8000 null values\n",
    "- We will not drop records due to that\n",
    "- Because no EDA task requires us to use the average score"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# inspect rows where 'score' is null\n",
    "score_null = df.filter(df['score'].isNull())\n",
    "score_null.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# drop rows with null values in any column (other than 'rank')\n",
    "clean_df = df.na.drop(subset = [\"username\", \"popularity\", \"genre\"])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# check if and how many titles were lost due to above\n",
    "unique_animes_before = df.select(\"anime_id\").distinct().count()\n",
    "unique_animes_after = clean_df.select(\"anime_id\").distinct().count()\n",
    "unique_percentage_change = ((unique_animes_before - unique_animes_after) / unique_animes_before) * 100\n",
    "\n",
    "print(f\"Total unique anime titles: {unique_animes_before}\")\n",
    "print(f\"Number of unique animes lost after NA drop: {unique_animes_before - unique_animes_after}\")\n",
    "print(f\"Percentage change in unique titles after NA drop: {unique_percentage_change}\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Exploratory Data Analysis"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. The most popular animes according to users"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "task_1_path = output_folder + \"task_1.csv\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "# get the top 50 popular anime titles based on their 'popularity'\n",
    "# explanation:\n",
    "#   - select rows where 'popularity' > 0\n",
    "#   - remove duplicate rows based on 'title'\n",
    "#   - sort in ascending order\n",
    "\n",
    "top_50_popular = clean_df.filter(functions.col(\"popularity\") > 0).dropDuplicates([\"title\"]).orderBy(\"popularity\").limit(50)\n",
    "top_50_popular = top_50_popular.select(\"anime_id\", \"title\", \"score\", \"popularity\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "top_50_popular = top_50_popular.coalesce(1)\n",
    "top_50_popular.write.csv(task_1_path, header=True)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. The most popular anime for every genre"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "task_2_path = output_folder + \"task_2.csv\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 5:=======================================================> (33 + 1) / 34]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "No. of distinct genres: 44\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "exploded_genres_df = clean_df.withColumn(\"genre\", F.explode(F.split(\"genre\", \", \")))\n",
    "no_distinct_genres = exploded_genres_df.select(\"genre\").distinct().count()\n",
    "\n",
    "print(\"No. of distinct genres:\", no_distinct_genres)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 8:=====================================================>   (14 + 1) / 15]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+----------+--------------------+\n",
      "|       genre|popularity|               title|\n",
      "+------------+----------+--------------------+\n",
      "|      Seinen|         5|       One Punch Man|\n",
      "|    Dementia|        48|Neon Genesis Evan...|\n",
      "|     Romance|         3|    Sword Art Online|\n",
      "|       Magic|         4|Fullmetal Alchemi...|\n",
      "|       Josei|       256|          Usagi Drop|\n",
      "|    Thriller|         1|          Death Note|\n",
      "|   Adventure|         3|    Sword Art Online|\n",
      "| Super Power|         2|  Shingeki no Kyojin|\n",
      "|Martial Arts|        10|              Naruto|\n",
      "|      Sports|        85|    Kuroko no Basket|\n",
      "|      Shoujo|        69|Ouran Koukou Host...|\n",
      "|       Drama|         2|  Shingeki no Kyojin|\n",
      "|   Shoujo Ai|       380|           Yuru Yuri|\n",
      "|      School|         7|        Angel Beats!|\n",
      "|      Hentai|      1057|        Boku no Pico|\n",
      "|    Military|         2|  Shingeki no Kyojin|\n",
      "|     Samurai|        67|    Samurai Champloo|\n",
      "|        Yaoi|      1057|        Boku no Pico|\n",
      "|     Fantasy|         2|  Shingeki no Kyojin|\n",
      "|      Demons|        14|      Ao no Exorcist|\n",
      "+------------+----------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# find the most popular anime for each genre\n",
    "# explanation:\n",
    "#     - split the 'genre' column into multiple rows (using explode)\n",
    "#     - partition the df by genre and order by popularity\n",
    "#     - assign rank to each partition (rank = 1 => highest popularity for that genre)\n",
    "\n",
    "genre_exploded = clean_df.withColumn(\"genre\", F.explode(F.split(\"genre\", \", \"))).select(\"genre\", \"popularity\", \"title\")\n",
    "\n",
    "window = Window.partitionBy(\"genre\").orderBy(F.col(\"popularity\"))\n",
    "ranked = genre_exploded.withColumn(\"rank\", F.dense_rank().over(window))\n",
    "max_popularity_by_genre = ranked.filter(F.col(\"rank\") == 1).select(\"genre\", \"popularity\", \"title\")\n",
    "\n",
    "most_popular_by_genre = max_popularity_by_genre.dropDuplicates()\n",
    "\n",
    "most_popular_by_genre.show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "most_popular_by_genre = most_popular_by_genre.coalesce(1)\n",
    "most_popular_by_genre.write.csv(task_2_path, header=True)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. The most popular anime genres for every gender"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "task_3_path = output_folder + \"task_3.csv\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "# split genre column and put each genre in its own row\n",
    "genre_gender_exploded = clean_df.withColumn(\"genre\", F.explode(F.split(\"genre\", \", \"))).select(\"genre\", \"gender\")\n",
    "\n",
    "# count instances of each (genre, gender) combination\n",
    "genre_gender_counts = genre_gender_exploded.groupBy(\"genre\", \"gender\").agg(F.count(\"*\").alias(\"count\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 24:======================================================> (33 + 1) / 34]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-------+--------+\n",
      "|gender    |genre  |count   |\n",
      "+----------+-------+--------+\n",
      "|Female    |Comedy |4671909 |\n",
      "|Female    |Action |3482950 |\n",
      "|Female    |Drama  |3102596 |\n",
      "|Female    |Romance|3058903 |\n",
      "|Female    |Fantasy|2477082 |\n",
      "|Male      |Comedy |12438080|\n",
      "|Male      |Action |10506368|\n",
      "|Male      |Romance|7422895 |\n",
      "|Male      |Drama  |6713172 |\n",
      "|Male      |Fantasy|6232376 |\n",
      "|Non-Binary|Comedy |112258  |\n",
      "|Non-Binary|Action |86008   |\n",
      "|Non-Binary|Drama  |65602   |\n",
      "|Non-Binary|Romance|62128   |\n",
      "|Non-Binary|School |57705   |\n",
      "+----------+-------+--------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# for each gender, find top 5 genres\n",
    "window = Window.partitionBy(\"gender\").orderBy(F.desc(\"count\"))\n",
    "ranked = genre_gender_counts.withColumn(\"rank\", F.dense_rank().over(window))\n",
    "top_genres_by_gender = ranked.filter(F.col(\"rank\") <= 5) \\\n",
    "                              .select(\"gender\", \"genre\", \"count\") \\\n",
    "                              .orderBy(\"gender\", \"rank\")\n",
    "\n",
    "# display top 5 genres for Female, Male & Non-Binary\n",
    "top_genres_by_gender.show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "top_genres_by_gender = top_genres_by_gender.coalesce(1)\n",
    "top_genres_by_gender.write.csv(task_3_path, header=True)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4. Power users (users with a high number of ratings)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "task_4_path = output_folder + \"task_4.csv\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 43:======================================================> (33 + 1) / 34]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----------------+------------+\n",
      "|user_id|        username|rating_count|\n",
      "+-------+----------------+------------+\n",
      "|   8669|     spacecowboy|        8739|\n",
      "|1245229|      TsukasaKei|        8739|\n",
      "|4561255|         uemmega|        8588|\n",
      "|3979333|          Exxorn|        8558|\n",
      "|1237755|    DeadlyKizuna|        8153|\n",
      "|1381655|          xbhrjd|        8078|\n",
      "|1636745|      JakCooper2|        7894|\n",
      "|2063865|         De_Baer|        7412|\n",
      "| 291713|      Dedzapadlo|        7235|\n",
      "| 805623|   DesireDestiny|        7010|\n",
      "|4042345|       ComfyLoli|        6705|\n",
      "| 132251|            canc|        6605|\n",
      "|1283539|      Dragonflyk|        6535|\n",
      "| 337383|VincentHarkonnen|        6511|\n",
      "|4328447|     KanaenuYume|        6423|\n",
      "| 186731|     NightTerror|        6407|\n",
      "|  98846|        coty9090|        6292|\n",
      "| 216713|           Cafer|        6285|\n",
      "|2476641|         Tsutaee|        6281|\n",
      "|1287643|      AngelShiva|        6049|\n",
      "+-------+----------------+------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# no. of ratings for each user\n",
    "# get top 50 power users (using limit)\n",
    "user_rating_counts = clean_df.select(\"user_id\", \"username\") \\\n",
    "                       .groupBy(\"user_id\", \"username\") \\\n",
    "                       .agg(F.count(\"*\").alias(\"rating_count\")) \\\n",
    "                       .orderBy(F.desc(\"rating_count\")).limit(50)\n",
    "\n",
    "# show results (top 50)\n",
    "user_rating_counts.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "user_rating_counts = user_rating_counts.coalesce(1)\n",
    "user_rating_counts.write.csv(task_4_path, header=True)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 5. Anime titles with the highest number of user ratings"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "task_5_path = output_folder + \"task_5.csv\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 49:======================================================> (33 + 1) / 34]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+--------------------+---------+\n",
      "|anime_id|               title|scored_by|\n",
      "+--------+--------------------+---------+\n",
      "|    1535|          Death Note|  1009477|\n",
      "|   16498|  Shingeki no Kyojin|   940211|\n",
      "|   11757|    Sword Art Online|   915986|\n",
      "|    5114|Fullmetal Alchemi...|   733592|\n",
      "|   30276|       One Punch Man|   691845|\n",
      "|   22319|         Tokyo Ghoul|   659308|\n",
      "|      20|              Naruto|   648605|\n",
      "|    6547|        Angel Beats!|   641851|\n",
      "|    1575|Code Geass: Hangy...|   627740|\n",
      "|   19815|     No Game No Life|   623227|\n",
      "|   10620|    Mirai Nikki (TV)|   592994|\n",
      "|    9253|         Steins;Gate|   563857|\n",
      "|    4224|           Toradora!|   557898|\n",
      "|    2904|Code Geass: Hangy...|   543904|\n",
      "|   21881| Sword Art Online II|   531486|\n",
      "|    9919|      Ao no Exorcist|   521881|\n",
      "|     226|          Elfen Lied|   514656|\n",
      "|   20507|            Noragami|   502213|\n",
      "|     199|Sen to Chihiro no...|   498602|\n",
      "|   31964|Boku no Hero Acad...|   494037|\n",
      "+--------+--------------------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "top_50_by_user_ratings = clean_df.filter(functions.col(\"scored_by\") > 0).dropDuplicates([\"title\"]).orderBy(functions.col(\"scored_by\").desc()).limit(50)\n",
    "top_50_by_user_ratings = top_50_by_user_ratings.select(\"anime_id\", \"title\", \"scored_by\")\n",
    "top_50_by_user_ratings.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "top_50_by_user_ratings = top_50_by_user_ratings.coalesce(1)\n",
    "top_50_by_user_ratings.write.csv(task_5_path, header=True)"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 6. Checking for a relationship between an anime's source (e.g., manga, light novel original) and its score"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Why use eta value here?\n",
    "- Our objective is to investigate if an anime's source is related to its overall score, so apart from a scatterplot, it would be nice to have some sort of a statistical measure for the strength of the relationship between these variables.\n",
    "- Since we want to find the correlation between the 'source' column (independent nominal variable) and the 'score' column (dependent scale variable), we can not use a traditional correlation metric like Pearson's coefficient.\n",
    "- A suitable candidate to measure this nominal-by-interval association would be eta correlation. Eta is a coefficient of nonlinear association, and requires that the dependent variable be interval in level, and the independent variable be categorical (nominal, ordinal, or grouped interval)."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "| Eta value       | Association       |\n",
    "|-----------------|-------------------|\n",
    "| < 0.20          | No association    |\n",
    "| 0.21 - 0.40     | Weak association  |\n",
    "| 0.41 - 0.70     | Medium association|\n",
    "| > 0.70          | Strong association|\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 79:>                                                         (0 + 1) / 1]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Eta correlation between anime source and average user score: 0.31289410349397284\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# convert anime source -> index (number)\n",
    "string_indexer = StringIndexer(inputCol=\"source\", outputCol=\"source_indexed\")\n",
    "indexed_df = string_indexer.fit(clean_df).transform(clean_df)\n",
    "\n",
    "# group means and grand mean\n",
    "group_means = indexed_df.groupBy(\"source_indexed\").agg(F.mean(\"score\").alias(\"group_mean\"))\n",
    "grand_mean = indexed_df.agg(F.mean(\"score\")).collect()[0][0]\n",
    "\n",
    "# sum of suqared differences (ssw)\n",
    "ssw_per_group = indexed_df.join(group_means, \"source_indexed\") \\\n",
    "                          .withColumn(\"deviation\", F.col(\"score\") - F.col(\"group_mean\")) \\\n",
    "                          .groupBy(\"source_indexed\") \\\n",
    "                          .agg(F.sum(F.pow(\"deviation\", 2)).alias(\"ssw\"))\n",
    "ssw = ssw_per_group.agg(F.sum(\"ssw\")).collect()[0][0]\n",
    "\n",
    "# sum of squared differences bw groups (ssb)\n",
    "ssb = group_means.join(indexed_df, \"source_indexed\") \\\n",
    "                  .groupBy(\"source_indexed\") \\\n",
    "                  .agg((F.pow(F.first(\"group_mean\") - grand_mean, 2) * F.count(\"*\")).alias(\"ssb\")) \\\n",
    "                  .agg(F.sum(\"ssb\").alias(\"ssb_sum\")).collect()[0][\"ssb_sum\"]\n",
    "\n",
    "# eta correlation = sqrt(ssb / (ssb + ssw))\n",
    "eta_correlation = (ssb / (ssb + ssw)) ** 0.5\n",
    "print(f\"Eta correlation between anime source and average user score: {eta_correlation}\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "According to the table above, we can conclude that an anime's source and its average score are **WEAKLY CORRELATED**."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.14"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}