movies_pyspark_recommendations.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## DETAILED DESCRIPTIONS OF DATA FILES\n",
    "Here are brief descriptions of the data.\n",
    "\n",
    "ml-data.tar.gz -- Compressed tar file. To rebuild the u data files do this:\n",
    "gunzip ml-data.tar.gz\n",
    "tar xvf ml-data.tar\n",
    "mku.sh\n",
    "\n",
    "**u.data** -- The full u data set, 100000 ratings by 943 users on 1682 items.\n",
    "Each user has rated at least 20 movies. Users and items are\n",
    "numbered consecutively from 1. The data is randomly\n",
    "ordered. This is a tab separated list of\n",
    "user id | item id | rating | timestamp.\n",
    "The time stamps are unix seconds since 1/1/1970 UTC\n",
    "\n",
    "**u.info** -- The number of users, items, and ratings in the u data set.\n",
    "\n",
    "**u.item** -- Information about the items (movies); this is a tab separated\n",
    "list of\n",
    "movie id | movie title | release date | video release date |\n",
    "IMDb URL | unknown | Action | Adventure | Animation |\n",
    "Children's | Comedy | Crime | Documentary | Drama | Fantasy |\n",
    "Film-Noir | Horror | Musical | Mystery | Romance | Sci-Fi |\n",
    "Thriller | War | Western |\n",
    "The last 19 fields are the genres, a 1 indicates the movie\n",
    "is of that genre, a 0 indicates it is not; movies can be in\n",
    "several genres at once.\n",
    "The movie ids are the ones used in the u.data data set.\n",
    "\n",
    "**u.genre** -- A list of the genres.\n",
    "\n",
    "**u.user** -- Demographic information about the users; this is a tab\n",
    "separated list of\n",
    "user id | age | gender | occupation | zip code\n",
    "The user ids are the ones used in the u.data data set.\n",
    "\n",
    "**u.occupation** -- A list of the occupations.\n",
    "\n",
    "**u1.base** -- The data sets u1.base and u1.test through u5.base and u5.test\n",
    "**u1.test** are 80%/20% splits of the u data into training and test data.\n",
    "**u2.base** Each of u1, …, u5 have disjoint test sets; this if for\n",
    "**u2.test** 5 fold cross validation (where you repeat your experiment\n",
    "**u3.base with each training and test set and average the results).\n",
    "**u3.test** These data sets can be generated from u.data by mku.sh.\n",
    "**u4.base**\n",
    "**u4.test**\n",
    "**u5.base**\n",
    "**u5.test**\n",
    "\n",
    "**ua.base** -- The data sets ua.base, ua.test, ub.base, and ub.test\n",
    "**ua.test** split the u data into a training set and a test set with\n",
    "**ub.base** exactly 10 ratings per user in the test set. The sets\n",
    "**ub.test** ua.test and ub.test are disjoint. These data sets can\n",
    "be generated from u.data by mku.sh.\n",
    "\n",
    "**allbut.pl** -- The script that generates training and test sets where\n",
    "all but n of a users ratings are in the training data.\n",
    "\n",
    "**mku.sh** -- A shell script to generate all the u data sets from u.data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.sql import SparkSession\n",
    "import pyspark.sql.functions as sf \n",
    "\n",
    "# Import the required ML functions\n",
    "from pyspark.ml.evaluation import RegressionEvaluator\n",
    "from pyspark.ml.recommendation import ALS\n",
    "from pyspark.ml.tuning import ParamGridBuilder, CrossValidator\n",
    "from pyspark.ml.evaluation import RegressionEvaluator\n",
    "from pyspark.ml.tuning import ParamGridBuilder, CrossValidator\n",
    "\n",
    "\n",
    "import sys"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Following are the parameters of a SparkContext.\n",
    "\n",
    "**Master** − It is the URL of the cluster it connects to.\n",
    "\n",
    "**appName** − Name of your job.\n",
    "\n",
    "**sparkHome** − Spark installation directory.\n",
    "\n",
    "**pyFiles** − The .zip or .py files to send to the cluster and add to the PYTHONPATH.\n",
    "\n",
    "**Environment** − Worker nodes environment variables.\n",
    "\n",
    "**batchSize** − The number of Python objects represented as a single Java object. Set 1 to disable batching, 0 to automatically choose the batch size based on object sizes, or -1 to use an unlimited batch size.\n",
    "\n",
    "**Serializerve** − RDD serializer.\n",
    "\n",
    "**Conf** − An object of L{SparkConf} to set all the Spark properties.\n",
    "\n",
    "**Gateway** − Use an existing gateway and JVM, otherwise initializing a new JVM.\n",
    "\n",
    "**JSC** − The JavaSparkContext instance.\n",
    "**profiler_cls** − A class of custom Profiler used to do profiling (the default is pyspark.profiler.BasicProfiler).\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    " spark = SparkSession \\\n",
    " .builder \\\n",
    " .appName(\"recommendation_engine\") \\\n",
    " .master(\"local\") \\\n",
    " .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "ratings = spark.read.csv(\"./data/ml-100k/ua.base\", sep=\"\\t\")\n",
    "#ratings = spark.read.load('data/ml-100k/ratings.parquet', schema=schema)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "ratings = ratings.toDF('userId','movieId','rating','timestamp')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['userId', 'movieId', 'rating', 'timestamp']"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ratings.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "ratings = ratings.sort(['userId','movieId'], ascending=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+------+---------+\n",
      "|userId|movieId|rating|timestamp|\n",
      "+------+-------+------+---------+\n",
      "|     1|      1|     5|874965758|\n",
      "|     1|     10|     3|875693118|\n",
      "|     1|    100|     5|878543541|\n",
      "|     1|    101|     2|878542845|\n",
      "|     1|    102|     2|889751736|\n",
      "|     1|    103|     1|878542845|\n",
      "|     1|    104|     1|875241619|\n",
      "|     1|    105|     2|875240739|\n",
      "|     1|    106|     4|875241390|\n",
      "|     1|    107|     4|875241619|\n",
      "|     1|    108|     5|875240920|\n",
      "|     1|    109|     5|874965739|\n",
      "|     1|     11|     2|875072262|\n",
      "|     1|    110|     1|878542845|\n",
      "|     1|    111|     5|889751711|\n",
      "|     1|    112|     1|878542441|\n",
      "|     1|    113|     5|878542738|\n",
      "|     1|    114|     5|875072173|\n",
      "|     1|    115|     5|878541637|\n",
      "|     1|    116|     3|878542960|\n",
      "+------+-------+------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# show schema\n",
    "ratings.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "90570"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# count rows\n",
    "ratings.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Count the total number of ratings in the dataset\n",
    "numerator = ratings.select(\"rating\").count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "number of unique users:  943\n"
     ]
    }
   ],
   "source": [
    "# Count the number of distinct userIds and distinct movieIds\n",
    "num_users = ratings.select(\"userId\").distinct().count()\n",
    "print('number of unique users: ', num_users)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "number of unique movies:  1680\n"
     ]
    }
   ],
   "source": [
    "num_movies = ratings.select(\"movieId\").distinct().count()\n",
    "print('number of unique movies: ', num_movies)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "number of possible ratings:  1584240\n"
     ]
    }
   ],
   "source": [
    "# Set the denominator equal to the number of users multiplied by the number of movies\n",
    "denominator = num_users * num_movies\n",
    "print('number of possible ratings: ', denominator)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The ratings dataframe is  94.28% empty.\n"
     ]
    }
   ],
   "source": [
    "# Divide the numerator by the denominator\n",
    "sparsity = (1.0 - (numerator *1.0)/denominator)*100\n",
    "print(\"The ratings dataframe is \", \"%.2f\" % sparsity + \"% empty.\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+------+---------+\n",
      "|userId|movieId|rating|timestamp|\n",
      "+------+-------+------+---------+\n",
      "|     1|      1|     5|874965758|\n",
      "|     1|     10|     3|875693118|\n",
      "|     1|    100|     5|878543541|\n",
      "|     1|    101|     2|878542845|\n",
      "|     1|    102|     2|889751736|\n",
      "|     1|    103|     1|878542845|\n",
      "|     1|    104|     1|875241619|\n",
      "|     1|    105|     2|875240739|\n",
      "|     1|    106|     4|875241390|\n",
      "|     1|    107|     4|875241619|\n",
      "|     1|    108|     5|875240920|\n",
      "|     1|    109|     5|874965739|\n",
      "|     1|     11|     2|875072262|\n",
      "|     1|    110|     1|878542845|\n",
      "|     1|    111|     5|889751711|\n",
      "|     1|    112|     1|878542441|\n",
      "|     1|    113|     5|878542738|\n",
      "|     1|    114|     5|875072173|\n",
      "|     1|    115|     5|878541637|\n",
      "|     1|    116|     3|878542960|\n",
      "+------+-------+------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Filter to show only userIds less than 100\n",
    "ratings.filter(sf.col(\"userId\") < 100).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-----+\n",
      "|userId|count|\n",
      "+------+-----+\n",
      "|   296|  137|\n",
      "|   467|   34|\n",
      "|   675|   24|\n",
      "|   691|   22|\n",
      "|   829|   54|\n",
      "|   125|  172|\n",
      "|   451|   88|\n",
      "|   800|   18|\n",
      "|   853|   31|\n",
      "|   666|  235|\n",
      "|   870|  259|\n",
      "|   919|  207|\n",
      "|   926|   10|\n",
      "|   124|   14|\n",
      "|   447|  129|\n",
      "|    51|   13|\n",
      "|   591|   74|\n",
      "|     7|  393|\n",
      "|   307|  102|\n",
      "|   475|   10|\n",
      "+------+-----+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Group data by userId, count ratings\n",
    "ratings.groupBy(\"userId\").count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Movie with the fewest ratings: \n",
      "+----------+\n",
      "|min(count)|\n",
      "+----------+\n",
      "|         1|\n",
      "+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Min num ratings for movies\n",
    "print(\"Movie with the fewest ratings: \")\n",
    "ratings.groupBy(\"movieId\").count().select(sf.min(\"count\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Movie with the fewest ratings: \n",
      "+------------------+\n",
      "|        avg(count)|\n",
      "+------------------+\n",
      "|53.910714285714285|\n",
      "+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Min num ratings for movies\n",
    "print(\"Movie with the fewest ratings: \")\n",
    "ratings.groupBy(\"movieId\").count().select(sf.avg('count')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "User with the fewest ratings: \n",
      "+----------+\n",
      "|min(count)|\n",
      "+----------+\n",
      "|        10|\n",
      "+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Min num ratings for user\n",
    "print(\"User with the fewest ratings: \")\n",
    "ratings.groupBy(\"userId\").count().select(sf.min(\"count\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Avg num ratings per user: \n",
      "+-----------------+\n",
      "|       avg(count)|\n",
      "+-----------------+\n",
      "|96.04453870625663|\n",
      "+-----------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Avg num ratings per users\n",
    "print(\"Avg num ratings per user: \")\n",
    "ratings.groupBy(\"userId\").count().select(sf.avg(\"count\")).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- userId: string (nullable = true)\n",
      " |-- movieId: string (nullable = true)\n",
      " |-- rating: string (nullable = true)\n",
      " |-- timestamp: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Use .printSchema() to see the datatypes of the ratings dataset\n",
    "ratings.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Tell Spark to convert the columns to the proper data types\n",
    "ratings = ratings.select(ratings.userId.cast(\"integer\"), ratings.movieId.cast(\"integer\"), ratings.rating.cast(\"double\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- userId: integer (nullable = true)\n",
      " |-- movieId: integer (nullable = true)\n",
      " |-- rating: double (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Call .printSchema() again to confirm the columns are now in the correct format\n",
    "ratings.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# ALS Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create test and train set\n",
    "(train, test) = ratings.randomSplit([0.8, 0.2], seed = 1234)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create ALS model\n",
    "als = ALS(userCol=\"userId\", itemCol=\"movieId\", ratingCol=\"rating\", nonnegative = True, implicitPrefs = False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.ml.recommendation.ALS"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Confirm that a model called \"als\" was created\n",
    "type(als)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Add hyperparameters and their respective values to param_grid\n",
    "#param_grid = ParamGridBuilder() \\\n",
    "#            .addGrid(als.rank,  [10, 50, 100, 150]) \\\n",
    "#            .addGrid(als.maxIter, [5, 50, 100, 200]) \\\n",
    "#            .addGrid(als.regParam, [.01, .05, .1, .15]) \\\n",
    "#            .build()\n",
    "\n",
    "param_grid = ParamGridBuilder() \\\n",
    "            .addGrid(als.rank,  [10, 50]) \\\n",
    "            .addGrid(als.maxIter, [5]) \\\n",
    "            .build()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Num models to be tested:  2\n"
     ]
    }
   ],
   "source": [
    "# Define evaluator as RMSE and print length of evaluator\n",
    "evaluator = RegressionEvaluator(metricName=\"rmse\", labelCol=\"rating\", predictionCol=\"prediction\") \n",
    "print (\"Num models to be tested: \", len(param_grid))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CrossValidator_a615bdab980c\n"
     ]
    }
   ],
   "source": [
    "# Build cross validation using CrossValidator\n",
    "cv = CrossValidator(estimator=als, estimatorParamMaps=param_grid, evaluator=evaluator, numFolds=5)\n",
    "\n",
    "# Confirm cv was built\n",
    "print(cv)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Fit cross validator to the 'train' dataset\n",
    "model = cv.fit(train)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [],
   "source": [
    "best_model = model.bestModel"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'pyspark.ml.recommendation.ALSModel'>\n",
      "**Best Model**\n",
      "  Rank: 10\n"
     ]
    }
   ],
   "source": [
    "# Print best_model\n",
    "print(type(best_model))\n",
    "\n",
    "# Complete the code below to extract the ALS model parameters\n",
    "print(\"**Best Model**\")\n",
    "\n",
    "# Print \"Rank\"\n",
    "print(\"  Rank:\", best_model.rank)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_test_01 = spark.read.csv(\"./data/ml-100k/ua.test\", sep=\"\\t\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_test_01 = df_test_01.toDF('userId','movieId','rating','timestamp')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_test_01 = df_test_01.select(df_test_01.userId.cast(\"integer\"), df_test_01.movieId.cast(\"integer\"), df_test_01.rating.cast(\"double\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_predictions = best_model.transform(df_test_01)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+------+----------+\n",
      "|userId|movieId|rating|prediction|\n",
      "+------+-------+------+----------+\n",
      "|   251|    148|   2.0| 3.0283358|\n",
      "|   580|    148|   4.0| 3.0808342|\n",
      "|   633|    148|   1.0| 2.9991403|\n",
      "|   642|    148|   5.0|   4.01511|\n",
      "|   406|    148|   3.0| 2.5808716|\n",
      "|    26|    148|   3.0|  2.611642|\n",
      "|    44|    148|   4.0| 2.8444238|\n",
      "|   271|    148|   3.0| 2.8736029|\n",
      "|   606|    148|   3.0| 3.3448424|\n",
      "|   916|    148|   2.0|  2.479743|\n",
      "|   236|    148|   4.0| 3.1556623|\n",
      "|   602|    148|   4.0| 3.6904817|\n",
      "|   663|    148|   4.0| 2.9024608|\n",
      "|   222|    148|   2.0| 2.8818076|\n",
      "|   601|    148|   3.0| 1.9276849|\n",
      "|   330|    148|   4.0| 4.1763325|\n",
      "|   372|    148|   5.0|  3.984272|\n",
      "|   727|    148|   2.0| 2.9217217|\n",
      "|   190|    148|   4.0| 3.2250655|\n",
      "|   224|    148|   3.0| 3.3781059|\n",
      "+------+-------+------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "test_predictions.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- userId: integer (nullable = true)\n",
      " |-- movieId: integer (nullable = true)\n",
      " |-- rating: double (nullable = true)\n",
      " |-- prediction: float (nullable = false)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "test_predictions.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_predictions = test_predictions.select(test_predictions.userId.cast(\"integer\"),\n",
    "                               test_predictions.movieId.cast(\"integer\"),\n",
    "                               test_predictions.rating.cast(\"double\"),\n",
    "                              test_predictions.prediction.cast(\"double\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "nan\n"
     ]
    }
   ],
   "source": [
    "RMSE = evaluator.evaluate(test_predictions)\n",
    "print(RMSE)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[userId: int, movieId: int, rating: double, prediction: double]"
      ]
     },
     "execution_count": 42,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "test_predictions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [],
   "source": [
    "# recommendations for all users \n",
    "ALS_recommendations = best_model.recommendForAllUsers(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {},
   "outputs": [],
   "source": [
    "ALS_recommendations.registerTempTable(\"ALS_recs_temp\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 82,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[recommendations: array<struct<movieId:int,rating:float>>]"
      ]
     },
     "execution_count": 82,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ALS_recommendations.select('recommendations')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {},
   "outputs": [
    {
     "ename": "NameError",
     "evalue": "name 'getrows' is not defined",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mNameError\u001b[0m                                 Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-77-827dc65f7811>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[0;32m----> 1\u001b[0;31m \u001b[0mgetrows\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mALS_recommendations\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mrownums\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0;36m0\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0;36m2\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mcollect\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[0;31mNameError\u001b[0m: name 'getrows' is not defined"
     ]
    }
   ],
   "source": [
    "getrows(ALS_recommendations, rownums=[0, 2]).collect()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 86,
   "metadata": {},
   "outputs": [],
   "source": [
    "clean_recs = spark.sql(\"SELECT userId, \\\n",
    "                       moviesIds_and_ratings.movieId AS movieId, \\\n",
    "                       moviesIds_And_ratings.rating AS prediction \\\n",
    "                       FROM ALS_recs_temp  \\\n",
    "                       LATERAL VIEW explode(recommendations) exploded_table \\\n",
    "                        moviesIds_and_ratings\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 87,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+----------+\n",
      "|userId|movieId|prediction|\n",
      "+------+-------+----------+\n",
      "|   471|    838| 5.7406297|\n",
      "|   471|   1155| 5.1363497|\n",
      "|   471|    394| 5.1212916|\n",
      "|   471|   1242| 5.0978255|\n",
      "|   471|    867| 5.0426164|\n",
      "|   471|    613| 4.9960318|\n",
      "|   471|    898|  4.980299|\n",
      "|   471|   1313|  4.958345|\n",
      "|   471|    289|  4.943675|\n",
      "|   471|   1344| 4.8187056|\n",
      "|   463|   1431|  4.512798|\n",
      "|   463|   1344|  4.505311|\n",
      "|   463|   1589| 4.4850698|\n",
      "|   463|   1242| 4.2059665|\n",
      "|   463|   1268|  4.190686|\n",
      "|   463|    887| 4.0957394|\n",
      "|   463|    963| 4.0705633|\n",
      "|   463|    838|  4.067269|\n",
      "|   463|   1137|   4.02338|\n",
      "|   463|    613| 4.0128913|\n",
      "+------+-------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "clean_recs.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 100,
   "metadata": {},
   "outputs": [],
   "source": [
    "items_names = spark.read.csv('data/ml-100k/u.item', sep='|')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 102,
   "metadata": {},
   "outputs": [],
   "source": [
    "items_names = items_names.select('_c0', '_c1')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 104,
   "metadata": {},
   "outputs": [],
   "source": [
    "items_names = items_names.toDF('movieId', 'movieName')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 105,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+--------------------+\n",
      "|movieId|           movieName|\n",
      "+-------+--------------------+\n",
      "|      1|    Toy Story (1995)|\n",
      "|      2|    GoldenEye (1995)|\n",
      "|      3|   Four Rooms (1995)|\n",
      "|      4|   Get Shorty (1995)|\n",
      "|      5|      Copycat (1995)|\n",
      "|      6|Shanghai Triad (Y...|\n",
      "|      7|Twelve Monkeys (1...|\n",
      "|      8|         Babe (1995)|\n",
      "|      9|Dead Man Walking ...|\n",
      "|     10|  Richard III (1995)|\n",
      "|     11|Seven (Se7en) (1995)|\n",
      "|     12|Usual Suspects, T...|\n",
      "|     13|Mighty Aphrodite ...|\n",
      "|     14|  Postino, Il (1994)|\n",
      "|     15|Mr. Holland's Opu...|\n",
      "|     16|French Twist (Gaz...|\n",
      "|     17|From Dusk Till Da...|\n",
      "|     18|White Balloon, Th...|\n",
      "|     19|Antonia's Line (1...|\n",
      "|     20|Angels and Insect...|\n",
      "+-------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "items_names.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 107,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------+----------+--------------------+\n",
      "|movieId|userId|prediction|           movieName|\n",
      "+-------+------+----------+--------------------+\n",
      "|    838|   471| 5.7406297|In the Line of Du...|\n",
      "|   1155|   471| 5.1363497|Rendezvous in Par...|\n",
      "|    394|   471| 5.1212916|Radioland Murders...|\n",
      "|   1242|   471| 5.0978255|Old Lady Who Walk...|\n",
      "|    867|   471| 5.0426164|Whole Wide World,...|\n",
      "|    613|   471| 4.9960318|My Man Godfrey (1...|\n",
      "|    898|   471|  4.980299| Postman, The (1997)|\n",
      "|   1313|   471|  4.958345|     Palmetto (1998)|\n",
      "|    289|   471|  4.943675|        Evita (1996)|\n",
      "|   1344|   471| 4.8187056|Story of Xinghua,...|\n",
      "|   1431|   463|  4.512798| Legal Deceit (1997)|\n",
      "|   1344|   463|  4.505311|Story of Xinghua,...|\n",
      "|   1589|   463| 4.4850698|  Schizopolis (1996)|\n",
      "|   1242|   463| 4.2059665|Old Lady Who Walk...|\n",
      "|   1268|   463|  4.190686|  Bitter Moon (1992)|\n",
      "|    887|   463| 4.0957394|  Eve's Bayou (1997)|\n",
      "|    963|   463| 4.0705633|Some Folks Call I...|\n",
      "|    838|   463|  4.067269|In the Line of Du...|\n",
      "|   1137|   463|   4.02338|Beautiful Thing (...|\n",
      "|    613|   463| 4.0128913|My Man Godfrey (1...|\n",
      "+-------+------+----------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# recommended movies by user\n",
    "clean_recs.join(items_names, ['movieId'], \"left\").show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.6.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}