recommendation_steam.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Recommender Systems for Steam Video Games"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This project focuses on building a recommendation system using Steam video game data. The dataset includes **user_id**, **name** (game title), **hours** (time spent playing), and **action** (whether the game was purchased or played). Since the data does not contain explicit feedback, I rely on implicit feedback using the hours feature. The more hours a user has spent playing a game, the more likely they are to prefer similar games, making hours a strong indicator of user preferences.\n",
    "\n",
    "You can access the dataset here: [Steam Video Games Data](https://www.kaggle.com/datasets/tamber/steam-video-games/data)\n",
    "\n",
    "Two models were chosen to implement this recommendation system:\n",
    "* **Alternating Least Squares (ALS):** A collaborative filtering model implemented with PySpark, which is well-suited for implicit feedback data and scales efficiently with larger datasets. ALS leverages matrix factorization to learn latent factors for users and games.\n",
    "* **Neural Collaborative Filtering (NCF):** A deep learning-based recommendation approach that models user-item interactions using neural networks. \n",
    "\n",
    "Although the dataset contains 200,000 records, which could be handled using **pandas**, I opted to use **PySpark** for data preprocessing to gain experience with distributed data processing tools. This choice helps develop proficiency in handling larger datasets efficiently and prepares for future projects where scalability might be critical.\n",
    "\n",
    "Some sources:\n",
    "* An Overview of Collaborative Filtering Algorithms for Implicit Feedback Data: https://andbloch.github.io/An-Overview-of-Collaborative-Filtering-Algorithms/\n",
    "* Neural Collaborative Filtering: https://liqiangnie.github.io/paper/p173-he.pdf"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 122,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import col, count, when, isnan, countDistinct, sum, round, max, min, explode, udf\n",
    "from pyspark.sql.types import IntegerType\n",
    "from pyspark.ml.feature import StringIndexer\n",
    "from pyspark.ml.recommendation import ALS\n",
    "from pyspark.ml.evaluation import RegressionEvaluator\n",
    "from pyspark.ml.tuning import CrossValidator, ParamGridBuilder\n",
    "import tensorflow as tf\n",
    "import tensorflow_recommenders as tfrs\n",
    "from tensorflow.keras import layers, Model\n",
    "import numpy as np\n",
    "from sklearn.model_selection import train_test_split\n",
    "from sklearn.metrics import mean_squared_error\n",
    "import pandas as pd"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create a Spark session"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {},
   "outputs": [
    {
     "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://192.168.1.101:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v3.5.3</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>Steam Recommender System</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x17dfb0bf0>"
      ]
     },
     "execution_count": 46,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark = SparkSession.builder.appName('Steam Recommender System').getOrCreate()\n",
    "spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Load the Steam video game data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[_c0: int, _c1: string, _c2: string, _c3: double, _c4: int]"
      ]
     },
     "execution_count": 47,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df = spark.read.csv('steam-200k.csv', inferSchema = True)\n",
    "df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------------------+--------+-----+---+\n",
      "|      _c0|                 _c1|     _c2|  _c3|_c4|\n",
      "+---------+--------------------+--------+-----+---+\n",
      "|151603712|The Elder Scrolls...|purchase|  1.0|  0|\n",
      "|151603712|The Elder Scrolls...|    play|273.0|  0|\n",
      "|151603712|           Fallout 4|purchase|  1.0|  0|\n",
      "|151603712|           Fallout 4|    play| 87.0|  0|\n",
      "|151603712|               Spore|purchase|  1.0|  0|\n",
      "+---------+--------------------+--------+-----+---+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Display the schema of the DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- _c0: integer (nullable = true)\n",
      " |-- _c1: string (nullable = true)\n",
      " |-- _c2: string (nullable = true)\n",
      " |-- _c3: double (nullable = true)\n",
      " |-- _c4: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['_c0', '_c1', '_c2', '_c3', '_c4']"
      ]
     },
     "execution_count": 50,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Rename the columns for better readability:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = df.withColumnRenamed('_c0', 'user_id') \\\n",
    "       .withColumnRenamed('_c1', 'name') \\\n",
    "       .withColumnRenamed('_c2', 'action') \\\n",
    "       .withColumnRenamed('_c3', 'hours') \\\n",
    "       .withColumnRenamed('_c4', 'zero')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- user_id: integer (nullable = true)\n",
      " |-- name: string (nullable = true)\n",
      " |-- action: string (nullable = true)\n",
      " |-- hours: double (nullable = true)\n",
      " |-- zero: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Get the shape of the DataFrame by counting the number of rows and columns."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(200000, 5)"
      ]
     },
     "execution_count": 53,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.count(), len(df.columns) ## shape"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Count the number of null values in each column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----+------+-----+----+\n",
      "|user_id|name|action|hours|zero|\n",
      "+-------+----+------+-----+----+\n",
      "|      0|   0|     0|    0|   0|\n",
      "+-------+----+------+-----+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.select([count(when(col(c).isNull(), c)).alias(c) for c in df.columns]).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Count the number of NaN values in each column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----+------+-----+----+\n",
      "|user_id|name|action|hours|zero|\n",
      "+-------+----+------+-----+----+\n",
      "|      0|   0|     0|    0|   0|\n",
      "+-------+----+------+-----+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "columns = ['user_id', 'name', 'action', 'hours', 'zero']\n",
    "\n",
    "df.select([count(when(isnan(col(c)), c)).alias(c) for c in columns]).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*Duplicate records*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 229:===================>                                     (1 + 2) / 3]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+--------------------+--------+-----+----+-----+\n",
      "|  user_id|                name|  action|hours|zero|count|\n",
      "+---------+--------------------+--------+-----+----+-----+\n",
      "| 86338111|Grand Theft Auto ...|purchase|  1.0|   0|    2|\n",
      "|189858084|Grand Theft Auto ...|purchase|  1.0|   0|    2|\n",
      "|150882304|Sid Meier's Civil...|purchase|  1.0|   0|    2|\n",
      "|189858084|Grand Theft Auto ...|purchase|  1.0|   0|    2|\n",
      "|116617462|Grand Theft Auto ...|purchase|  1.0|   0|    2|\n",
      "| 37422528|Sid Meier's Civil...|purchase|  1.0|   0|    2|\n",
      "|147859903|Sid Meier's Civil...|purchase|  1.0|   0|    2|\n",
      "|138941587|Sid Meier's Civil...|purchase|  1.0|   0|    2|\n",
      "|145825155|Grand Theft Auto ...|purchase|  1.0|   0|    2|\n",
      "|105782521|Sid Meier's Civil...|purchase|  1.0|   0|    2|\n",
      "| 46301758|Sid Meier's Civil...|purchase|  1.0|   0|    2|\n",
      "|179936723|Grand Theft Auto ...|purchase|  1.0|   0|    2|\n",
      "| 64455019|Sid Meier's Civil...|purchase|  1.0|   0|    2|\n",
      "|142650116|Grand Theft Auto ...|purchase|  1.0|   0|    2|\n",
      "| 33013552|Sid Meier's Civil...|purchase|  1.0|   0|    2|\n",
      "| 93275329|Grand Theft Auto III|purchase|  1.0|   0|    2|\n",
      "| 86055705|Grand Theft Auto ...|purchase|  1.0|   0|    2|\n",
      "| 24721232|Grand Theft Auto ...|purchase|  1.0|   0|    2|\n",
      "| 24721232|Sid Meier's Civil...|purchase|  1.0|   0|    2|\n",
      "|105877396|Sid Meier's Civil...|purchase|  1.0|   0|    2|\n",
      "+---------+--------------------+--------+-----+----+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "df.groupBy(columns).count().filter('count > 1').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_new = df.dropDuplicates()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "199293"
      ]
     },
     "execution_count": 58,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_new.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*Unique Values* -- Calculates the number of unique values for each column "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----+------+-----+----+\n",
      "|user_id|name|action|hours|zero|\n",
      "+-------+----+------+-----+----+\n",
      "|  12393|5155|     2| 1593|   1|\n",
      "+-------+----+------+-----+----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_new.select([countDistinct(c).alias(c) for c in columns]).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Remove the 'zero' column from the DataFrame as it is not needed for analysis"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_new = df_new.drop('zero')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### User Interactions Analysis"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Count of Users per Game"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-----------+\n",
      "|                name|users_count|\n",
      "+--------------------+-----------+\n",
      "|              Dota 2|       9682|\n",
      "|     Team Fortress 2|       4646|\n",
      "|Counter-Strike Gl...|       2789|\n",
      "|            Unturned|       2632|\n",
      "|       Left 4 Dead 2|       1752|\n",
      "|Counter-Strike So...|       1693|\n",
      "|      Counter-Strike|       1424|\n",
      "|         Garry's Mod|       1397|\n",
      "|The Elder Scrolls...|       1394|\n",
      "|            Warframe|       1271|\n",
      "|Half-Life 2 Lost ...|       1158|\n",
      "|Sid Meier's Civil...|       1150|\n",
      "|           Robocraft|       1096|\n",
      "|Half-Life 2 Death...|       1021|\n",
      "|              Portal|       1005|\n",
      "|            Portal 2|        997|\n",
      "|         Half-Life 2|        995|\n",
      "|   Heroes & Generals|        993|\n",
      "|            Terraria|        956|\n",
      "|Counter-Strike Co...|        904|\n",
      "+--------------------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.groupBy('name')\\\n",
    "    .agg(count('user_id').alias('users_count'))\\\n",
    "    .orderBy('users_count', ascending = False)\\\n",
    "    .limit(20)\\\n",
    "    .show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Count of Purchases per Game"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+-------------------+\n",
      "|                name|count_user_purchase|\n",
      "+--------------------+-------------------+\n",
      "|              Dota 2|               4841|\n",
      "|     Team Fortress 2|               2323|\n",
      "|            Unturned|               1563|\n",
      "|Counter-Strike Gl...|               1412|\n",
      "|Half-Life 2 Lost ...|                981|\n",
      "|Counter-Strike So...|                978|\n",
      "|       Left 4 Dead 2|                951|\n",
      "|      Counter-Strike|                856|\n",
      "|            Warframe|                847|\n",
      "|Half-Life 2 Death...|                823|\n",
      "|         Garry's Mod|                731|\n",
      "|The Elder Scrolls...|                717|\n",
      "|           Robocraft|                689|\n",
      "|Counter-Strike Co...|                679|\n",
      "|Counter-Strike Co...|                679|\n",
      "|   Heroes & Generals|                658|\n",
      "|         Half-Life 2|                639|\n",
      "|Sid Meier's Civil...|                596|\n",
      "|         War Thunder|                590|\n",
      "|              Portal|                588|\n",
      "+--------------------+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter(col('action') == 'purchase') \\\n",
    "  .groupBy('name') \\\n",
    "  .agg(count('user_id').alias('count_user_purchase')) \\\n",
    "  .orderBy('count_user_purchase', ascending=False) \\\n",
    "  .limit(20) \\\n",
    "  .show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Count of Plays per Game"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+---------------+\n",
      "|                name|count_user_play|\n",
      "+--------------------+---------------+\n",
      "|              Dota 2|           4841|\n",
      "|     Team Fortress 2|           2323|\n",
      "|Counter-Strike Gl...|           1377|\n",
      "|            Unturned|           1069|\n",
      "|       Left 4 Dead 2|            801|\n",
      "|Counter-Strike So...|            715|\n",
      "|The Elder Scrolls...|            677|\n",
      "|         Garry's Mod|            666|\n",
      "|      Counter-Strike|            568|\n",
      "|Sid Meier's Civil...|            554|\n",
      "|            Terraria|            460|\n",
      "|            Portal 2|            453|\n",
      "|            Warframe|            424|\n",
      "|              Portal|            417|\n",
      "|           Robocraft|            407|\n",
      "|            PAYDAY 2|            390|\n",
      "|       Borderlands 2|            386|\n",
      "|         Half-Life 2|            356|\n",
      "|   Heroes & Generals|            335|\n",
      "|         War Thunder|            303|\n",
      "+--------------------+---------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter(col('action') == 'play') \\\n",
    "  .groupBy('name') \\\n",
    "  .agg(count('user_id').alias('count_user_play')) \\\n",
    "  .orderBy('count_user_play', ascending=False) \\\n",
    "  .limit(20) \\\n",
    "  .show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "####  Total Play Hours per Game"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+----------+\n",
      "|                name|total_play|\n",
      "+--------------------+----------+\n",
      "|              Dota 2|  981684.6|\n",
      "|Counter-Strike Gl...|  322771.6|\n",
      "|     Team Fortress 2|  173673.3|\n",
      "|      Counter-Strike|  134261.1|\n",
      "|Sid Meier's Civil...|   99821.3|\n",
      "|Counter-Strike So...|   96075.5|\n",
      "|The Elder Scrolls...|   70889.3|\n",
      "|         Garry's Mod|   49725.3|\n",
      "|Call of Duty Mode...|   42009.9|\n",
      "|       Left 4 Dead 2|   33596.7|\n",
      "|Football Manager ...|   32308.6|\n",
      "|Football Manager ...|   30845.8|\n",
      "|Football Manager ...|   30574.8|\n",
      "|            Terraria|   29951.8|\n",
      "|            Warframe|   27074.6|\n",
      "|Football Manager ...|   24283.1|\n",
      "|              Arma 3|   24055.7|\n",
      "|  Grand Theft Auto V|   22956.7|\n",
      "|       Borderlands 2|   22667.9|\n",
      "|    Empire Total War|   21030.3|\n",
      "+--------------------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.filter(col('action') == 'play') \\\n",
    "  .groupBy('name') \\\n",
    "  .agg(round(sum('hours'), 2).alias('total_play')) \\\n",
    "  .orderBy('total_play', ascending = False) \\\n",
    "  .limit(50) \\\n",
    "  .show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Data prep for recommedation system models"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This part processes the dataset to prepare it for recommendation models. It begins by filtering out entries where the **action** column indicates a purchase or play, resulting in two separate DataFrames: one for purchases and one for plays. These two DataFrames are then merged based on user_id and name. Afterward, null values in the **hours** column are replaced with 0 to ensure accurate calculations and unnecessary columns are removed. Finally, the **hours** column is *normalized*."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Filter the DataFrame to create a new DataFrame 'df_purchase' that includes only the records where the action is 'purchase'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "128804"
      ]
     },
     "execution_count": 65,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_purchase = df_new.filter(col('action') == 'purchase')\n",
    "df_purchase.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Count distinct users made purchases and rename the columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------------------------+\n",
      "|number of unique users made purchases|\n",
      "+-------------------------------------+\n",
      "|                                12393|\n",
      "+-------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_purchase.select(countDistinct('user_id').alias('number of unique users made purchases')).show()\n",
    "df_purchase = df_purchase.withColumnRenamed('action', 'action_pur') \\\n",
    "                         .withColumnRenamed('hours', 'purchase')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create a new DataFrame 'df_play' that contains only the records where the action is 'play'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "70489"
      ]
     },
     "execution_count": 67,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_play = df_new.filter(col('action') == 'play')\n",
    "df_play.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Count the number of unique users"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------------+\n",
      "|number of unique users|\n",
      "+----------------------+\n",
      "|                 11350|\n",
      "+----------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_play.select(countDistinct('user_id').alias('number of unique users')).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The two DataFrames (df_purchase and df_play) are joined based on the **user_id** and **name** (game title). "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {},
   "outputs": [],
   "source": [
    "full_df = df_purchase.join(df_play, [\"user_id\", \"name\"], \"outer\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "128816"
      ]
     },
     "execution_count": 70,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "full_df.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Count the distinct values for each column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {},
   "outputs": [],
   "source": [
    "new_columns = full_df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 307:=========>                                               (1 + 5) / 6]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----+----------+--------+------+-----+\n",
      "|user_id|name|action_pur|purchase|action|hours|\n",
      "+-------+----+----------+--------+------+-----+\n",
      "|  12393|5155|         1|       1|     1| 1593|\n",
      "+-------+----+----------+--------+------+-----+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "full_df_unique = full_df.select([countDistinct(c).alias(c) for c in new_columns]).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Count the number of null (missing) values in each column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----+----------+--------+------+-----+\n",
      "|user_id|name|action_pur|purchase|action|hours|\n",
      "+-------+----+----------+--------+------+-----+\n",
      "|      0|   0|         0|       0| 58327|58327|\n",
      "+-------+----+----------+--------+------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "full_df.select([count(when(col(c).isNull(), c)).alias(c) for c in new_columns]).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Fill any null values in the **hours** column with zero"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = full_df.fillna(0, subset=['hours'])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Drop unnecessary columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {},
   "outputs": [],
   "source": [
    "data = data.drop(\"action_pur\", \"purchase\", \"action\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------------------+\n",
      "|summary|             hours|\n",
      "+-------+------------------+\n",
      "|  count|            128816|\n",
      "|   mean|26.746411936405334|\n",
      "| stddev|171.38236775962963|\n",
      "|    min|               0.0|\n",
      "|    max|           11754.0|\n",
      "+-------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "data.describe(\"hours\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Normalize the **hours** column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {},
   "outputs": [],
   "source": [
    "min_hours = data.select(min(\"hours\")).collect()[0][0]\n",
    "max_hours = data.select(max(\"hours\")).collect()[0][0]\n",
    "\n",
    "data = data.withColumn(\"normalized_hours\", (col(\"hours\") - min_hours) / (max_hours - min_hours))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Recommender Systems"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1) Alternating Least Squares (ALS) - Pyspark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this code, *StringIndexer* is used to convert the categorical game names in the **name** column into unique numerical identifiers stored in the **game_id** column. This transformation is necessary for preparing the dataset for the ALS model, which requires numerical input for training."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "indexer = StringIndexer(inputCol=\"name\", outputCol=\"game_id\")\n",
    "data_indexed = indexer.fit(data).transform(data)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*Train - Test Split*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {},
   "outputs": [],
   "source": [
    "(train_set, test_set) = data_indexed.randomSplit([0.8, 0.2])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*Model Selection - Hyperparameter Tunning*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "24/10/21 14:37:11 WARN InstanceBuilder: Failed to load implementation from:dev.ludovic.netlib.blas.JNIBLAS\n",
      "24/10/21 14:37:11 WARN InstanceBuilder: Failed to load implementation from:dev.ludovic.netlib.lapack.JNILAPACK\n",
      "                                                                                ]\r"
     ]
    }
   ],
   "source": [
    "# the als model \n",
    "als = ALS(userCol=\"user_id\", itemCol=\"game_id\", ratingCol=\"normalized_hours\", implicitPrefs=True, coldStartStrategy='drop') #Note we set cold start strategy to 'drop' \n",
    "#to ensure we don't get NaN evaluation metrics\n",
    "\n",
    "# the parameter grid\n",
    "paramGrid = ParamGridBuilder() \\\n",
    "  .addGrid(als.maxIter, [5, 10, 15]) \\\n",
    "  .addGrid(als.regParam, [0.01, 0.05, 0.1]) \\\n",
    "  .build()\n",
    "\n",
    "# cross-validator\n",
    "crossval = CrossValidator(estimator=als, estimatorParamMaps=paramGrid, evaluator=RegressionEvaluator(metricName=\"rmse\", labelCol=\"normalized_hours\", predictionCol=\"prediction\"), numFolds=5, seed=42)\n",
    "\n",
    "# Fit the cross-validator to the training data\n",
    "cvModel = crossval.fit(train_set)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 81,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Best maxIter: 5\n",
      "Best regParam: 0.1\n"
     ]
    }
   ],
   "source": [
    "# best model\n",
    "bestModel = cvModel.bestModel\n",
    "\n",
    "# Parameter map\n",
    "paramMap = bestModel.extractParamMap()\n",
    "\n",
    "# best params\n",
    "best_maxIter = bestModel._java_obj.parent().getMaxIter()\n",
    "best_regParam = bestModel._java_obj.parent().getRegParam()\n",
    "\n",
    "print(\"Best maxIter:\", best_maxIter)\n",
    "print(\"Best regParam:\", best_regParam)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*Model Training*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 82,
   "metadata": {},
   "outputs": [],
   "source": [
    "als_best = ALS(maxIter=best_maxIter, regParam=best_regParam, userCol=\"user_id\", itemCol=\"game_id\",\n",
    "                ratingCol=\"normalized_hours\", implicitPrefs=True, coldStartStrategy=\"drop\")\n",
    "als_best = als_best.fit(train_set)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*Making predictions*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 83,
   "metadata": {},
   "outputs": [],
   "source": [
    "predictions_als = als_best.transform(test_set)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*Model Evaluation*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 84,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "RMSE of ALS: 0.09944930743217605\n"
     ]
    }
   ],
   "source": [
    "evaluator_als = RegressionEvaluator(metricName=\"rmse\", labelCol=\"normalized_hours\", predictionCol=\"prediction\")\n",
    "rmse_als = evaluator_als.evaluate(predictions_als)\n",
    "print('RMSE of ALS:', rmse_als)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*Model Training by using all data to get recommendations*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 85,
   "metadata": {},
   "outputs": [],
   "source": [
    "als_best = ALS(maxIter=best_maxIter, regParam=best_regParam, userCol=\"user_id\", itemCol=\"game_id\",\n",
    "                ratingCol=\"normalized_hours\", implicitPrefs=True, coldStartStrategy=\"drop\")\n",
    "als_full_model = als_best.fit(data_indexed)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*Recommender System*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 86,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 24390:===============================================>    (91 + 8) / 100]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+--------------------+\n",
      "| user_id|     recommendations|\n",
      "+--------+--------------------+\n",
      "|   76767|[{44, 0.8258528},...|\n",
      "|  144736|[{7, 0.60747874},...|\n",
      "|  229911|[{7, 0.8815583}, ...|\n",
      "|  835015|[{49, 0.0}, {48, ...|\n",
      "|  948368|[{7, 0.6061667}, ...|\n",
      "|  975449|[{20, 0.7061146},...|\n",
      "| 1268792|[{7, 0.60748684},...|\n",
      "| 2531540|[{5, 0.42228198},...|\n",
      "| 2753525|[{5, 0.6406964}, ...|\n",
      "| 3450426|[{7, 0.730884}, {...|\n",
      "| 7923954|[{7, 0.6074828}, ...|\n",
      "| 7987640|[{49, 0.0}, {48, ...|\n",
      "| 8259307|[{11, 0.4106043},...|\n",
      "| 8567888|[{5, 0.3509526}, ...|\n",
      "| 8585433|[{11, 0.598985}, ...|\n",
      "| 8784496|[{1, 0.4596405}, ...|\n",
      "| 8795607|[{5, 0.66845745},...|\n",
      "|10144413|[{49, 0.0}, {48, ...|\n",
      "|10595342|[{5, 0.6881972}, ...|\n",
      "|10599862|[{11, 0.82561207}...|\n",
      "+--------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "user_recommendations = als_full_model.recommendForAllUsers(5)\n",
    "user_recommendations.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The DataFrame presented contains two columns: **user_id** and **recommendations**. In the **recommendations** column, five suggested games are listed alongside their predicted normalized hours (ratings), identified by specific game IDs. To enhance the interpretability of the recommendations, I will implement code to map these IDs to their corresponding game names. Initially, I will display the schema of the DataFrame for a clearer understanding of its structure."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 87,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- user_id: integer (nullable = false)\n",
      " |-- recommendations: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- game_id: integer (nullable = true)\n",
      " |    |    |-- rating: float (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "user_recommendations.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "According to the schema of the data, the recommendations column contains an array, and each element is a struct. Each struct includes two fields: game_id (an integer) and rating (a float). The I explode the recommendations column, which contains arrays of recommended games. This results in a new row for each recommended game, making it easier to process them individually."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 88,
   "metadata": {},
   "outputs": [],
   "source": [
    "recommendations_exploded = user_recommendations.withColumn(\"recommendation\", explode(\"recommendations\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Select relevant columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 89,
   "metadata": {},
   "outputs": [],
   "source": [
    "recommendations_als = recommendations_exploded.select(\"user_id\", \"recommendation.game_id\", \"recommendation.rating\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Mapp game IDs to names"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 90,
   "metadata": {},
   "outputs": [],
   "source": [
    "game_id_name = data_indexed[['name', 'game_id']]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 91,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "128816"
      ]
     },
     "execution_count": 91,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "game_id_name.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 92,
   "metadata": {},
   "outputs": [],
   "source": [
    "game_id_name = game_id_name.drop_duplicates()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 93,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "5155"
      ]
     },
     "execution_count": 93,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "game_id_name.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Add the game names to the recommendations DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 94,
   "metadata": {},
   "outputs": [],
   "source": [
    "recommendations_als = recommendations_als.join(game_id_name, on='game_id', how='left')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Sort results"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 95,
   "metadata": {},
   "outputs": [],
   "source": [
    "recommendations_als = recommendations_als.orderBy(['user_id', 'rating'], ascending =[True, False])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 96,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+----------+--------------------+\n",
      "|game_id|user_id|    rating|                name|\n",
      "+-------+-------+----------+--------------------+\n",
      "|      1|   5250|0.72689056|     Team Fortress 2|\n",
      "|      0|   5250|  0.701716|              Dota 2|\n",
      "|      6|   5250|0.13185938|       Left 4 Dead 2|\n",
      "|     19|   5250|0.12507021|              Portal|\n",
      "|      5|   5250|0.12311798|Counter-Strike So...|\n",
      "|     44|  76767| 0.8258528|Call of Duty Mode...|\n",
      "|     45|  76767| 0.8088316|Call of Duty Mode...|\n",
      "|     69|  76767| 0.5850272|Call of Duty Blac...|\n",
      "|     70|  76767| 0.5532179|Call of Duty Blac...|\n",
      "|      3|  76767|0.53492707|Counter-Strike Gl...|\n",
      "+-------+-------+----------+--------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "recommendations_als.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A function to find recommended games for a specific user"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 97,
   "metadata": {},
   "outputs": [],
   "source": [
    "def print_recommendation_als (user_id):\n",
    "    \"\"\"\n",
    "    A function to find and print the recommended games for a specific user.\n",
    "\n",
    "    Parameters:\n",
    "    - user_id (int): The ID of the user for whom the game recommendations will be retrieved.\n",
    "\n",
    "    The function filters the DataFrame based on the provided user_id and retrieves the game names associated with the recommendations for that user. It then prints the names of the recommended games in a numbered list.\n",
    "    \"\"\"\n",
    "    recommend_filtered = recommendations_als.filter(col('user_id') == user_id).select('name')\n",
    "    print(f\"User ID: {user_id}\")\n",
    "    recommended_games = recommend_filtered.collect()\n",
    "    print('The list of all recommended games:')\n",
    "    for i, row in enumerate(recommended_games, 1):\n",
    "        print(f\"{i}. {row['name']}\")\n",
    "\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 98,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "User ID: 1423371\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The list of all recommended games:\n",
      "1. Counter-Strike Global Offensive\n",
      "2. Counter-Strike Source\n",
      "3. PAYDAY 2\n",
      "4. Garry's Mod\n",
      "5. Left 4 Dead 2\n"
     ]
    }
   ],
   "source": [
    "print_recommendation_als(user_id = 1423371) ## example"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2) Neural Collaborative Filtering (NCF) - Tensorflow"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*Data Prep*"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Retrieve unique user IDs and game names and then converts them into Python lists "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 99,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/Users/gulsumasenacakir/recommenderSystems-SteamVideoGames/recom_tfrs/lib/python3.12/site-packages/pyspark/python/lib/pyspark.zip/pyspark/daemon.py:154: DeprecationWarning: This process (pid=12682) is multi-threaded, use of fork() may lead to deadlocks in the child.\n",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "user_ids = data_indexed.select('user_id').distinct().rdd.flatMap(lambda x:x).collect()\n",
    "game_names = data_indexed.select('name').distinct().rdd.flatMap(lambda x:x).collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Map each user ID and game name to a unique index"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 100,
   "metadata": {},
   "outputs": [],
   "source": [
    "userid_to_index = {user_id: index for index, user_id in enumerate(user_ids)}\n",
    "game_to_index = {name: index for index, name in enumerate(game_names)}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 101,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/Users/gulsumasenacakir/recommenderSystems-SteamVideoGames/recom_tfrs/lib/python3.12/site-packages/pyspark/python/lib/pyspark.zip/pyspark/daemon.py:154: DeprecationWarning: This process (pid=12689) is multi-threaded, use of fork() may lead to deadlocks in the child.\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+--------------------+-----+--------------------+-------+----------+----------+\n",
      "|user_id|                name|hours|    normalized_hours|game_id|user_index|game_index|\n",
      "+-------+--------------------+-----+--------------------+-------+----------+----------+\n",
      "|   5250|     Cities Skylines|144.0| 0.01225114854517611|  198.0|      5039|      2500|\n",
      "|   5250|      Counter-Strike|  0.0|                 0.0|    7.0|      5039|      2653|\n",
      "|   5250|Counter-Strike So...|  0.0|                 0.0|    5.0|      5039|        30|\n",
      "|   5250|       Day of Defeat|  0.0|                 0.0|   21.0|      5039|      1449|\n",
      "|   5250|              Dota 2|  0.2|1.701548409052237...|    0.0|      5039|         0|\n",
      "|   5250|Half-Life 2 Episo...|  0.0|                 0.0|   30.0|      5039|      2137|\n",
      "|   5250|Half-Life Blue Shift|  0.0|                 0.0|   40.0|      5039|       321|\n",
      "|   5250|Half-Life Opposin...|  0.0|                 0.0|   36.0|      5039|      1094|\n",
      "|  76767|Age of Empires II...| 13.1|0.001114514207929...|   83.0|       594|      2930|\n",
      "|  76767|         Alien Swarm|  0.8|6.806193636208951E-5|   54.0|       594|      2424|\n",
      "|  76767|Call of Duty Blac...| 12.5|0.001063467755657...|   70.0|       594|      1669|\n",
      "|  76767|Call of Duty Mode...| 65.0|0.005530032329419772|   44.0|       594|      2768|\n",
      "|  76767|Counter-Strike Gl...|  3.5|2.977709715841416E-4|    3.0|       594|      1702|\n",
      "|  76767|Counter-Strike So...| 25.0|0.002126935511315297|    5.0|       594|        30|\n",
      "|  76767|           Half-Life|  1.2|1.020929045431342...|   32.0|       594|      1716|\n",
      "|  76767|               Thief|  0.0|                 0.0|  254.0|       594|      2057|\n",
      "|  76767|             Thief 2|  0.0|                 0.0|  870.0|       594|      1095|\n",
      "|  86540|              Arma 2|  0.0|                 0.0|   60.0|      1163|       432|\n",
      "|  86540|Arma 2 Operation ...|  0.0|                 0.0|   71.0|      1163|       872|\n",
      "|  86540|           Audiosurf| 57.0|0.004849412965798877|  282.0|      1163|      2511|\n",
      "+-------+--------------------+-----+--------------------+-------+----------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# Functions to Map Users and Games to Indexes\n",
    "def get_user_index(user_id):\n",
    "    return userid_to_index.get(user_id, -1)\n",
    "\n",
    "def get_game_index(game_name):\n",
    "    return game_to_index.get(game_name, -1)\n",
    "  \n",
    "# Creating UDFs (User Defined Functions) to Use in PySpark\n",
    "user_index_udf = udf(get_user_index, IntegerType())\n",
    "game_index_udf = udf(get_game_index, IntegerType())\n",
    "\n",
    "# Applying the UDFs to the DataFrame\n",
    "indexed_user_data = data_indexed.withColumn(\"user_index\", user_index_udf(data_indexed[\"user_id\"]))\n",
    "indexed_user_data = indexed_user_data.withColumn(\"game_index\", game_index_udf(indexed_user_data[\"name\"]))\n",
    "\n",
    "indexed_user_data.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Convert the data to NumPy arrays"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 102,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "X = np.array(indexed_user_data.select(\"user_index\", \"game_index\").collect())\n",
    "y = np.array(indexed_user_data.select(\"normalized_hours\").collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*Train - test split*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 107,
   "metadata": {},
   "outputs": [],
   "source": [
    "X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*Model*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 105,
   "metadata": {},
   "outputs": [],
   "source": [
    "class NCFModel(Model):\n",
    "    def __init__(self, num_users, num_games, embedding_dim=8):\n",
    "        super(NCFModel, self).__init__()\n",
    "        \n",
    "        # embedding layers for users and games\n",
    "        self.user_embedding = layers.Embedding(num_users, embedding_dim)\n",
    "        self.game_embedding = layers.Embedding(num_games, embedding_dim)\n",
    "\n",
    "        # MLP (Multi-Layer Perceptron) layers\n",
    "        self.fc1 = layers.Dense(64, activation='relu')\n",
    "        self.fc2 = layers.Dense(32, activation='relu')\n",
    "        self.fc3 = layers.Dense(1, activation='sigmoid')\n",
    "\n",
    "    def call(self, inputs):\n",
    "        user_vector = self.user_embedding(inputs[:, 0])\n",
    "        game_vector = self.game_embedding(inputs[:, 1])\n",
    "\n",
    "        # merge embeddings\n",
    "        interaction = layers.concatenate([user_vector, game_vector])\n",
    "\n",
    "        # Fully Connected layers\n",
    "        x = self.fc1(interaction)\n",
    "        x = self.fc2(x)\n",
    "        return self.fc3(x)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*Model training and evaluation function*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 108,
   "metadata": {},
   "outputs": [],
   "source": [
    "def train_evaluate(num_users, num_games, embedding_dim):\n",
    "    model_ncf = NCFModel(num_users, num_games, embedding_dim)\n",
    "    model_ncf.compile(optimizer='adam', loss='mean_squared_error', metrics=['mae'])\n",
    "    \n",
    "    # Train the model\n",
    "    model_ncf.fit(X_train, y_train, epochs=10, batch_size=32)\n",
    "\n",
    "    # Predictions\n",
    "    y_pred_ncf = model_ncf.predict(X_test)\n",
    "\n",
    "    # Calculate RMSE\n",
    "    rmse = np.sqrt(mean_squared_error(y_test, y_pred_ncf))\n",
    "    return rmse\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Model Selection - Hyperparameter Tuning"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 110,
   "metadata": {},
   "outputs": [],
   "source": [
    "embedding_dims = [8, 16, 32]\n",
    "best_rmse = float('inf')\n",
    "best_embedding_dim = None\n",
    "num_users = len(user_ids)\n",
    "num_games = len(game_names)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 115,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Epoch 1/10\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m3s\u001b[0m 721us/step - loss: 0.0161 - mae: 0.0497\n",
      "Epoch 2/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m2s\u001b[0m 640us/step - loss: 2.2737e-04 - mae: 0.0036\n",
      "Epoch 3/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m2s\u001b[0m 654us/step - loss: 1.9624e-04 - mae: 0.0035\n",
      "Epoch 4/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m2s\u001b[0m 631us/step - loss: 1.9850e-04 - mae: 0.0032\n",
      "Epoch 5/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m2s\u001b[0m 639us/step - loss: 1.9515e-04 - mae: 0.0030\n",
      "Epoch 6/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m2s\u001b[0m 655us/step - loss: 1.8591e-04 - mae: 0.0031\n",
      "Epoch 7/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m2s\u001b[0m 627us/step - loss: 1.6705e-04 - mae: 0.0030\n",
      "Epoch 8/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m2s\u001b[0m 641us/step - loss: 1.8021e-04 - mae: 0.0028\n",
      "Epoch 9/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m2s\u001b[0m 685us/step - loss: 1.6814e-04 - mae: 0.0030\n",
      "Epoch 10/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m2s\u001b[0m 668us/step - loss: 1.8600e-04 - mae: 0.0031\n",
      "\u001b[1m806/806\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m0s\u001b[0m 263us/step\n",
      "Embedding Dimension: 8, RMSE: 0.01636454694972074\n",
      "Epoch 1/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m4s\u001b[0m 1ms/step - loss: 0.0156 - mae: 0.0485\n",
      "Epoch 2/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m3s\u001b[0m 1ms/step - loss: 2.1405e-04 - mae: 0.0034\n",
      "Epoch 3/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m3s\u001b[0m 1ms/step - loss: 1.9987e-04 - mae: 0.0034\n",
      "Epoch 4/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m3s\u001b[0m 998us/step - loss: 1.8371e-04 - mae: 0.0031\n",
      "Epoch 5/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m3s\u001b[0m 1ms/step - loss: 1.9013e-04 - mae: 0.0031\n",
      "Epoch 6/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m3s\u001b[0m 1ms/step - loss: 2.0021e-04 - mae: 0.0030\n",
      "Epoch 7/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m3s\u001b[0m 975us/step - loss: 1.6911e-04 - mae: 0.0029\n",
      "Epoch 8/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m3s\u001b[0m 1ms/step - loss: 1.6252e-04 - mae: 0.0030\n",
      "Epoch 9/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m3s\u001b[0m 990us/step - loss: 1.6178e-04 - mae: 0.0029\n",
      "Epoch 10/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m3s\u001b[0m 950us/step - loss: 1.5970e-04 - mae: 0.0031\n",
      "\u001b[1m806/806\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m0s\u001b[0m 240us/step\n",
      "Embedding Dimension: 16, RMSE: 0.01632310704850622\n",
      "Epoch 1/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m6s\u001b[0m 2ms/step - loss: 0.0164 - mae: 0.0497\n",
      "Epoch 2/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m6s\u001b[0m 2ms/step - loss: 2.0783e-04 - mae: 0.0033\n",
      "Epoch 3/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m6s\u001b[0m 2ms/step - loss: 1.9264e-04 - mae: 0.0034\n",
      "Epoch 4/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m6s\u001b[0m 2ms/step - loss: 1.8517e-04 - mae: 0.0033\n",
      "Epoch 5/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m6s\u001b[0m 2ms/step - loss: 2.1418e-04 - mae: 0.0032\n",
      "Epoch 6/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m6s\u001b[0m 2ms/step - loss: 1.8615e-04 - mae: 0.0031\n",
      "Epoch 7/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m6s\u001b[0m 2ms/step - loss: 1.6263e-04 - mae: 0.0032\n",
      "Epoch 8/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m6s\u001b[0m 2ms/step - loss: 1.8082e-04 - mae: 0.0031\n",
      "Epoch 9/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m6s\u001b[0m 2ms/step - loss: 1.6715e-04 - mae: 0.0031\n",
      "Epoch 10/10\n",
      "\u001b[1m3221/3221\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m5s\u001b[0m 2ms/step - loss: 1.5134e-04 - mae: 0.0029\n",
      "\u001b[1m806/806\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m0s\u001b[0m 286us/step\n",
      "Embedding Dimension: 32, RMSE: 0.016287419421987765\n",
      "Best RMSE: 0.016287419421987765 with Embedding Dimension: 32\n"
     ]
    }
   ],
   "source": [
    "for embedding_dim in embedding_dims:\n",
    "    rmse = train_evaluate(num_users, num_games, embedding_dim)\n",
    "    print(f\"Embedding Dimension: {embedding_dim}, RMSE: {rmse}\")\n",
    "\n",
    "    if rmse < best_rmse:\n",
    "        best_rmse_ncf = rmse\n",
    "        best_embedding_dim = embedding_dim\n",
    "\n",
    "print(f\"Best RMSE: {best_rmse_ncf} with Embedding Dimension: {best_embedding_dim}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 119,
   "metadata": {},
   "outputs": [],
   "source": [
    "#model\n",
    "embedding_dim = best_embedding_dim\n",
    "\n",
    "best_ncf = NCFModel(num_users, num_games, embedding_dim)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 120,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Epoch 1/10\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\u001b[1m4026/4026\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m8s\u001b[0m 2ms/step - loss: 0.0125 - mae: 0.0391\n",
      "Epoch 2/10\n",
      "\u001b[1m4026/4026\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m7s\u001b[0m 2ms/step - loss: 2.3770e-04 - mae: 0.0035\n",
      "Epoch 3/10\n",
      "\u001b[1m4026/4026\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m8s\u001b[0m 2ms/step - loss: 2.2065e-04 - mae: 0.0035\n",
      "Epoch 4/10\n",
      "\u001b[1m4026/4026\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m7s\u001b[0m 2ms/step - loss: 2.2568e-04 - mae: 0.0032\n",
      "Epoch 5/10\n",
      "\u001b[1m4026/4026\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m7s\u001b[0m 2ms/step - loss: 1.8430e-04 - mae: 0.0031\n",
      "Epoch 6/10\n",
      "\u001b[1m4026/4026\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m8s\u001b[0m 2ms/step - loss: 1.9718e-04 - mae: 0.0032\n",
      "Epoch 7/10\n",
      "\u001b[1m4026/4026\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m8s\u001b[0m 2ms/step - loss: 2.0062e-04 - mae: 0.0033\n",
      "Epoch 8/10\n",
      "\u001b[1m4026/4026\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m8s\u001b[0m 2ms/step - loss: 1.8482e-04 - mae: 0.0032\n",
      "Epoch 9/10\n",
      "\u001b[1m4026/4026\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m8s\u001b[0m 2ms/step - loss: 1.9254e-04 - mae: 0.0032\n",
      "Epoch 10/10\n",
      "\u001b[1m4026/4026\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m7s\u001b[0m 2ms/step - loss: 1.6496e-04 - mae: 0.0031\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "<keras.src.callbacks.history.History at 0x313a49220>"
      ]
     },
     "execution_count": 120,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "best_ncf.compile(optimizer='adam', loss='mean_squared_error', metrics=['mae'])\n",
    "best_ncf.fit(X, y, epochs=10, batch_size=32)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "*Recommender System*"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 121,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\u001b[1m162/162\u001b[0m \u001b[32m━━━━━━━━━━━━━━━━━━━━\u001b[0m\u001b[37m\u001b[0m \u001b[1m0s\u001b[0m 400us/step\n",
      "Recommended games for user 1423371: ['TRIP Steam Edition', 'Overcast - Walden and the Werewolf - Soundtrack', 'The Counting Kingdom', 'eXceed - Gun Bullet Children', 'Reckless Ruckus']\n"
     ]
    }
   ],
   "source": [
    "def recommend_games(user_id, model, game_to_index, top_n=5):\n",
    "    user_index = userid_to_index[user_id]\n",
    "    user_vector = np.array([user_index])\n",
    "\n",
    "    # game index\n",
    "    game_indices = np.array(range(len(game_to_index)))\n",
    "\n",
    "    # merge user index and game index\n",
    "    user_game_pairs = np.array(np.meshgrid(user_vector, game_indices)).T.reshape(-1, 2)\n",
    "\n",
    "    predictions = model.predict(user_game_pairs).flatten()\n",
    "\n",
    "    # top games\n",
    "    top_games_indices = np.argsort(predictions)[-top_n:][::-1]\n",
    "    recommended_games = [game_names[i] for i in top_games_indices]\n",
    "\n",
    "    return recommended_games\n",
    "\n",
    "\n",
    "recommended_games = recommend_games(1423371, best_ncf, game_to_index)\n",
    "print(\"Recommended games for user 1423371:\", recommended_games)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Comparison"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 125,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "  Model      RMSE\n",
      "0   ALS  0.099449\n",
      "1   NCF  0.016287\n"
     ]
    }
   ],
   "source": [
    "comparison = pd.DataFrame({\n",
    "    'Model': ['ALS', 'NCF'],\n",
    "    'RMSE': [rmse_als, best_rmse_ncf]\n",
    "})\n",
    "print(comparison)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The NCF's RMSE is lower than ALS's, indicating that the NCF model is more effective in capturing user preferences and behavior compared to the ALS model."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "base",
   "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.12.0"
  },
  "orig_nbformat": 4
 },
 "nbformat": 4,
 "nbformat_minor": 2
}