Pyspark dataframe ALS on MovieLens data.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 130,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.mllib.recommendation import ALS, MatrixFactorizationModel\n",
    "from pyspark import SparkContext\n",
    "from pyspark.sql import SQLContext\n",
    "from pyspark.sql.types import *\n",
    "import pandas as pd\n",
    "sc = SparkContext()\n",
    "sql_sc = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.evaluation import RegressionEvaluator\n",
    "from pyspark.ml.tuning import TrainValidationSplit, ParamGridBuilder"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "pd_df_ratings = pd.read_csv('C://Users/user/Desktop/ml-latest-small/ratings.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "pd_df_movies = pd.read_csv('C://Users/user/Desktop/ml-latest-small/movies.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "pyspark_df_ratings = sql_sc.createDataFrame(pd_df_ratings)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "pyspark_df_movies = sql_sc.createDataFrame(pd_df_movies)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 133,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "RDD_df_ratings = sc.textFile(\"C://Users/user/Desktop/ml-latest-small/ratings.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 134,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "RDD_df_movies = sc.textFile(\"C://Users/user/Desktop/ml-latest-small/movies.csv\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [],
   "source": [
    "d = pyspark_df_ratings.join(pyspark_df_movies, on = 'movieId', how =  'full')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [],
   "source": [
    "pyspark_df_ratings = pyspark_df_ratings.drop('Timestamp')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+------+\n",
      "|userId|movieId|rating|\n",
      "+------+-------+------+\n",
      "|1     |31     |2.5   |\n",
      "|1     |1029   |3.0   |\n",
      "|1     |1061   |3.0   |\n",
      "|1     |1129   |2.0   |\n",
      "|1     |1172   |4.0   |\n",
      "+------+-------+------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "pyspark_df_ratings.show(5, truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[movieId: bigint, title: string]"
      ]
     },
     "execution_count": 50,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pyspark_df_ratings.cache()\n",
    "pyspark_df_movies.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "pyspark_df_movies = pyspark_df_movies.drop('genres')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----------------------------------+\n",
      "|movieId|title                             |\n",
      "+-------+----------------------------------+\n",
      "|1      |Toy Story (1995)                  |\n",
      "|2      |Jumanji (1995)                    |\n",
      "|3      |Grumpier Old Men (1995)           |\n",
      "|4      |Waiting to Exhale (1995)          |\n",
      "|5      |Father of the Bride Part II (1995)|\n",
      "+-------+----------------------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "pyspark_df_movies.show(5, truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "joint_ratings_movies = pyspark_df_ratings.join(pyspark_df_movies, on='movieId')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------+------+--------------------+\n",
      "|movieId|userId|rating|               title|\n",
      "+-------+------+------+--------------------+\n",
      "|     31|     1|   2.5|Dangerous Minds (...|\n",
      "|   1029|     1|   3.0|        Dumbo (1941)|\n",
      "|   1061|     1|   3.0|     Sleepers (1996)|\n",
      "|   1129|     1|   2.0|Escape from New Y...|\n",
      "|   1172|     1|   4.0|Cinema Paradiso (...|\n",
      "+-------+------+------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "joint_ratings_movies.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "training_df, validation_df, test_df = joint_ratings_movies.randomSplit([0.6, 0.2, 0.2], seed = 42)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[movieId: bigint, userId: bigint, rating: double, title: string]"
      ]
     },
     "execution_count": 56,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "training_df.cache()\n",
    "validation_df.cache()\n",
    "test_df.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "MAX_ITERATIONS = 10\n",
    "REG_PARAM = 0.1\n",
    "SEED_VALUE = 42"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 114,
   "metadata": {},
   "outputs": [],
   "source": [
    "als = ALS(maxIter = MAX_ITERATIONS, regParam=REG_PARAM, userCol= 'userId', itemCol='movieId',ratingCol='rating')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 115,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "ALS_4a6e86fda199a64a02a7"
      ]
     },
     "execution_count": 115,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "als"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 116,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "reg_eval = RegressionEvaluator(predictionCol=\"prediction\", labelCol=\"rating\", metricName=\"rmse\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 117,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pyspark.ml.util.JavaMLReader at 0x148cfac8>"
      ]
     },
     "execution_count": 117,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "reg_eval.read()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 118,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "ranks = [1, 2, 3, 4]\n",
    "errors = [0, 0, 0, 0]\n",
    "models = [0, 0, 0, 0]\n",
    "count = 0\n",
    "min_error = float('inf')\n",
    "best_rank = -1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 119,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Rank: 1  RMSE: 0.91536003322\n",
      "Rank: 2  RMSE: 0.91947340264\n",
      "Rank: 3  RMSE: 0.931341261591\n",
      "Rank: 4  RMSE: 0.937812092655\n",
      "Best model: 1\n"
     ]
    }
   ],
   "source": [
    "for rank in ranks:\n",
    "    als.setRank(rank)\n",
    "    model =als.fit(training_df)\n",
    "    predict_df = model.transform(validation_df)\n",
    "    predicted_ratings_df = predict_df.filter(predict_df.prediction != float('nan'))\n",
    "    error = reg_eval.evaluate(predicted_ratings_df)\n",
    "    errors[count] = error\n",
    "    models[count]= model\n",
    "    print('Rank: %s  RMSE: %s' % (rank,error))\n",
    "    \n",
    "    if error < min_error:\n",
    "        min_error = error\n",
    "        best_rank = count\n",
    "    count += 1\n",
    "\n",
    "als.setRank(ranks[best_rank])\n",
    "print('Best model: %s' % ranks[best_rank])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 120,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "best_model = models[best_rank]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 121,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_predict_df = best_model.transform(test_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 122,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------+------+--------------------+----------+\n",
      "|movieId|userId|rating|               title|prediction|\n",
      "+-------+------+------+--------------------+----------+\n",
      "|    463|    30|   4.0|Guilty as Sin (1993)| 3.4663312|\n",
      "|    463|   311|   3.0|Guilty as Sin (1993)| 2.7467191|\n",
      "|    471|   588|   3.0|Hudsucker Proxy, ...|  4.026064|\n",
      "|    471|   126|   5.0|Hudsucker Proxy, ...| 4.1335073|\n",
      "|    471|    19|   3.0|Hudsucker Proxy, ...| 3.9036355|\n",
      "|    471|   309|   4.0|Hudsucker Proxy, ...| 4.3257484|\n",
      "|    471|    15|   3.0|Hudsucker Proxy, ...| 2.9689806|\n",
      "|    471|    73|   4.0|Hudsucker Proxy, ...|  3.865151|\n",
      "|    471|   487|   4.0|Hudsucker Proxy, ...|  4.436184|\n",
      "|    471|   508|   4.0|Hudsucker Proxy, ...| 4.3334002|\n",
      "+-------+------+------+--------------------+----------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "test_predict_df.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 123,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "test_predict_df = test_predict_df.filter(test_predict_df.prediction != float('nan'))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 124,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "RMSE on test set: 0.918710445384\n"
     ]
    }
   ],
   "source": [
    "test_RMSE = reg_eval.evaluate(test_predict_df)\n",
    "print('RMSE on test set: {0}'.format(test_RMSE))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "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.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}