02-data-preprocess.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import count, col, countDistinct, monotonically_increasing_id"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/12/25 12:54:50 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n"
     ]
    }
   ],
   "source": [
    "spark = SparkSession.builder.master(\"local[4]\") \\\n",
    "                    .appName('recommendation_system') \\\n",
    "                    .config(\"spark.driver.memory\", \"15g\") \\\n",
    "                    .getOrCreate()"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "df = spark.read.options(\n",
    "            header=True,\n",
    "            inferSchema=True\n",
    "        ) \\\n",
    "        .csv('data/ratings_edited.csv')"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- userId: integer (nullable = true)\n",
      " |-- rating: double (nullable = true)\n",
      " |-- movieIdx: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": "(20000263, 3)"
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.count(), len(df.columns)"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "### Removing Less common movies and users from the dataset"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "I am going to extract the ratings corresponding to the most common users and movies.\n",
    "The threshold I am setting here is number of most common users is N = 10000 and number of most common movies is M = 2000."
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "outputs": [],
   "source": [
    "N = 10000\n",
    "M = 2000"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "most_common_users_df = df.groupBy('userId') \\\n",
    "    .agg(count(\"*\").alias(\"frequency\")) \\\n",
    "    .orderBy(col('frequency').desc()) \\\n",
    "    .select('userId').head(N)\n",
    "\n",
    "most_common_movies_df = df.groupBy('movieIdx') \\\n",
    "    .agg(count(\"*\").alias(\"frequency\")) \\\n",
    "    .orderBy(col('frequency').desc()) \\\n",
    "    .select('movieIdx').head(M)\n",
    "\n",
    "most_common_users = [user[0] for user in most_common_users_df]\n",
    "most_common_movies = [movie[0] for movie in most_common_movies_df]"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "outputs": [],
   "source": [
    "df = df.filter(\n",
    "    (col(\"userId\").isin(most_common_users))\n",
    "    &\n",
    "    (col(\"movieIdx\").isin(most_common_movies))\n",
    ")"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": "[Row(count(DISTINCT movieIdx)=2000)]"
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.select(countDistinct('movieIdx')).collect()"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 26:==============================================>           (4 + 1) / 5]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The unique number of users is 10000\n",
      "User ID starts from 10\n",
      "User ID ends at 138473\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "print(f\"The unique number of users is {df.select(countDistinct('userId')).collect()[0][0]}\\n\"\n",
    "      f\"User ID starts from {df.agg({'userId': 'min'}).collect()[0][0]}\\n\"\n",
    "      f\"User ID ends at {df.agg({'userId': 'max'}).collect()[0][0]}\"\n",
    ")"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 29:==============================================>           (4 + 1) / 5]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+----------+\n",
      "|userId|new_userId|\n",
      "+------+----------+\n",
      "|    10|         0|\n",
      "|    23|         1|\n",
      "|    53|         2|\n",
      "|    57|         3|\n",
      "|    90|         4|\n",
      "|    95|         5|\n",
      "|   103|         6|\n",
      "|   115|         7|\n",
      "|   130|         8|\n",
      "|   133|         9|\n",
      "|   155|        10|\n",
      "|   207|        11|\n",
      "|   246|        12|\n",
      "|   250|        13|\n",
      "|   257|        14|\n",
      "|   270|        15|\n",
      "|   277|        16|\n",
      "|   293|        17|\n",
      "|   297|        18|\n",
      "|   317|        19|\n",
      "+------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "user_id_map = df.select('userId').distinct().sort('userId')\n",
    "user_id_map = user_id_map.withColumn(\"new_userId\", monotonically_increasing_id())\n",
    "user_id_map.show()"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "outputs": [],
   "source": [
    "movie_id_map = df.select('movieIdx').distinct().sort('movieIdx')\n",
    "movie_id_map = movie_id_map.withColumn(\"movieId\", monotonically_increasing_id())\n",
    "# movie_id_map.show()"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "outputs": [],
   "source": [
    "df = df.join(\n",
    "    user_id_map,\n",
    "    df.userId == user_id_map.userId,\n",
    "    \"inner\"\n",
    ") \\\n",
    "    .drop('userId') \\\n",
    "    .withColumnRenamed('new_userId', 'userId')\n",
    "\n",
    "df = df.join(\n",
    "    movie_id_map,\n",
    "    df.movieIdx == movie_id_map.movieIdx,\n",
    "    \"inner\"\n",
    ") \\\n",
    "    .drop('movieIdx')\n",
    "# ratings.show()"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": "5392234"
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.count()"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "# df.toPandas().to_csv('./data/ratings_preprocessed.csv')"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "df.write.options(header='True', delimiter=',').csv('./data/ratings_processed.csv')"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "outputs": [],
   "source": [
    "spark.stop()"
   ],
   "metadata": {
    "collapsed": false
   }
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}