00-data-download.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 81,
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import countDistinct, col, monotonically_increasing_id, count"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 89,
   "outputs": [],
   "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": 3,
   "outputs": [],
   "source": [
    "sc = spark.sparkContext"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Data Download"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 92,
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Downloading movielens-20m-dataset.zip to data\r\n",
      " 99%|███████████████████████████████████████▍| 193M/195M [00:06<00:00, 54.0MB/s]\r\n",
      "100%|████████████████████████████████████████| 195M/195M [00:06<00:00, 32.3MB/s]\r\n"
     ]
    }
   ],
   "source": [
    "!mkdir -p data\n",
    "!kaggle datasets download -d grouplens/movielens-20m-dataset -p data/ --unzip"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 108,
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "df = spark.read.options(\n",
    "            header=True,\n",
    "            inferSchema=True\n",
    "        ) \\\n",
    "        .csv('data/rating.csv')"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 109,
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": "20000263"
     },
     "execution_count": 109,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.count()"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 110,
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- userId: integer (nullable = true)\n",
      " |-- movieId: integer (nullable = true)\n",
      " |-- rating: double (nullable = true)\n",
      " |-- timestamp: timestamp (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.printSchema()"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 111,
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+------+-------------------+\n",
      "|userId|movieId|rating|          timestamp|\n",
      "+------+-------+------+-------------------+\n",
      "|     1|      2|   3.5|2005-04-02 23:53:47|\n",
      "|     1|     29|   3.5|2005-04-02 23:31:16|\n",
      "|     1|     32|   3.5|2005-04-02 23:33:39|\n",
      "+------+-------+------+-------------------+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show(3)"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Fixing User IDs and Movie IDs for easy indexing"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 112,
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 91:================================================>         (5 + 1) / 6]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The unique number of users is 138493\n",
      "User ID starts from 1\n",
      "User ID ends at 138493\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": 113,
   "outputs": [],
   "source": [
    "df = df.withColumn(\"userId\",col(\"userId\")-1)"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 114,
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 103:===============================================>         (5 + 1) / 6]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The unique number of users is 138493\n",
      "User ID starts from 0\n",
      "User ID ends at 138492\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": 115,
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 115:===============================================>         (5 + 1) / 6]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The unique number of movies is 26744\n",
      "Movie ID starts from 1\n",
      "Movie ID ends at 131262\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "print(f\"The unique number of movies is {df.select(countDistinct('movieId')).collect()[0][0]}\\n\"\n",
    "      f\"Movie ID starts from {df.agg({'movieId': 'min'}).collect()[0][0]}\\n\"\n",
    "      f\"Movie ID ends at {df.agg({'movieId': 'max'}).collect()[0][0]}\"\n",
    ")"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 116,
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 118:===============================================>         (5 + 1) / 6]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+--------+\n",
      "|movieId|movieIdx|\n",
      "+-------+--------+\n",
      "|      1|       0|\n",
      "|      2|       1|\n",
      "|      3|       2|\n",
      "|      4|       3|\n",
      "|      5|       4|\n",
      "|      6|       5|\n",
      "|      7|       6|\n",
      "|      8|       7|\n",
      "|      9|       8|\n",
      "|     10|       9|\n",
      "|     11|      10|\n",
      "|     12|      11|\n",
      "|     13|      12|\n",
      "|     14|      13|\n",
      "|     15|      14|\n",
      "|     16|      15|\n",
      "|     17|      16|\n",
      "|     18|      17|\n",
      "|     19|      18|\n",
      "|     20|      19|\n",
      "+-------+--------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "movieId_df = df.select('movieId').distinct().sort('movieId')\n",
    "movieId_df = movieId_df.withColumn(\"movieIdx\", monotonically_increasing_id())\n",
    "movieId_df.show()"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 117,
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- movieId: integer (nullable = true)\n",
      " |-- movieIdx: long (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "movieId_df.printSchema()"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 118,
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+------+-------------------+\n",
      "|userId|movieId|rating|          timestamp|\n",
      "+------+-------+------+-------------------+\n",
      "|     0|      2|   3.5|2005-04-02 23:53:47|\n",
      "|     0|     29|   3.5|2005-04-02 23:31:16|\n",
      "|     0|     32|   3.5|2005-04-02 23:33:39|\n",
      "|     0|     47|   3.5|2005-04-02 23:32:07|\n",
      "|     0|     50|   3.5|2005-04-02 23:29:40|\n",
      "|     0|    112|   3.5|2004-09-10 03:09:00|\n",
      "|     0|    151|   4.0|2004-09-10 03:08:54|\n",
      "|     0|    223|   4.0|2005-04-02 23:46:13|\n",
      "|     0|    253|   4.0|2005-04-02 23:35:40|\n",
      "|     0|    260|   4.0|2005-04-02 23:33:46|\n",
      "|     0|    293|   4.0|2005-04-02 23:31:43|\n",
      "|     0|    296|   4.0|2005-04-02 23:32:47|\n",
      "|     0|    318|   4.0|2005-04-02 23:33:18|\n",
      "|     0|    337|   3.5|2004-09-10 03:08:29|\n",
      "|     0|    367|   3.5|2005-04-02 23:53:00|\n",
      "|     0|    541|   4.0|2005-04-02 23:30:03|\n",
      "|     0|    589|   3.5|2005-04-02 23:45:57|\n",
      "|     0|    593|   3.5|2005-04-02 23:31:01|\n",
      "|     0|    653|   3.0|2004-09-10 03:08:11|\n",
      "|     0|    919|   3.5|2004-09-10 03:07:01|\n",
      "+------+-------+------+-------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df.show()"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 119,
   "outputs": [],
   "source": [
    "ratings = df.join(\n",
    "    movieId_df,\n",
    "    df.movieId == movieId_df.movieId,\n",
    "    \"inner\"\n",
    ") \\\n",
    "    .drop('movieId') \\\n",
    "    .drop('timestamp')"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 120,
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+------+--------+\n",
      "|userId|rating|movieIdx|\n",
      "+------+------+--------+\n",
      "|     0|   3.5|    3903|\n",
      "|     1|   4.0|    1528|\n",
      "|     1|   3.0|    3825|\n",
      "|     2|   4.0|    2281|\n",
      "|     6|   3.0|    1528|\n",
      "|     6|   2.0|    3088|\n",
      "|     8|   2.0|    4424|\n",
      "|    10|   5.0|    1528|\n",
      "|    10|   5.0|    1539|\n",
      "|    13|   5.0|     467|\n",
      "|    13|   3.5|    1528|\n",
      "|    13|   4.5|    3088|\n",
      "|    13|   4.5|   10371|\n",
      "|    13|   4.0|   10864|\n",
      "|    15|   4.0|    1528|\n",
      "|    16|   4.0|    1528|\n",
      "|    17|   2.5|    1528|\n",
      "|    20|   3.0|    2780|\n",
      "|    21|   3.0|    1528|\n",
      "|    22|   5.0|    1528|\n",
      "+------+------+--------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ratings.show()"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 121,
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 165:=============================================>           (4 + 1) / 5]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The unique number of users is 138493\n",
      "User ID starts from 0\n",
      "User ID ends at 138492\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "print(f\"The unique number of users is {ratings.select(countDistinct('userId')).collect()[0][0]}\\n\"\n",
    "      f\"User ID starts from {ratings.agg({'userId': 'min'}).collect()[0][0]}\\n\"\n",
    "      f\"User ID ends at {ratings.agg({'userId': 'max'}).collect()[0][0]}\"\n",
    ")"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 122,
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The unique number of movies is 26744\n",
      "Movie ID starts from 0\n",
      "Movie ID ends at 26743\n"
     ]
    }
   ],
   "source": [
    "print(f\"The unique number of movies is {ratings.select(countDistinct('movieIdx')).collect()[0][0]}\\n\"\n",
    "      f\"Movie ID starts from {ratings.agg({'movieIdx': 'min'}).collect()[0][0]}\\n\"\n",
    "      f\"Movie ID ends at {ratings.agg({'movieIdx': 'max'}).collect()[0][0]}\"\n",
    ")"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 123,
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "ratings.write.options(header='True', delimiter=',').csv('./data/ratings_edited.csv')"
   ],
   "metadata": {
    "collapsed": false
   }
  },
  {
   "cell_type": "code",
   "execution_count": 124,
   "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
}