als_bagging_on_movielens.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# ALS Bagging Ensemble for Movie Recommendation\n",
    "<br>  \n",
    "This notebook shows an example of using bagging ensemble w/ pySpark ASL on [MovieLens](https://grouplens.org/datasets/movielens/) data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "import pyspark.sql.functions as F\n",
    "from pyspark.ml.recommendation import ALS\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.types import (\n",
    "    StructType,\n",
    "    StructField,\n",
    "    StringType,\n",
    "    FloatType,\n",
    "    IntegerType,\n",
    "    LongType\n",
    ")\n",
    "\n",
    "import pandas as pd\n",
    "\n",
    "from bagging import RecoBagging\n",
    "from ranking_evaluator import RankingEvaluator\n",
    "\n",
    "spark = SparkSession.builder.appName(\"SAR pySpark\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Number of models to combine\n",
    "NUM_MODELS = 10\n",
    "\n",
    "# Number of items to recommend for each user\n",
    "TOP_K = 10"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Data loading\n",
    "Load 100k MovieLens data and randomly split into training and test sets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>userId</th>\n",
       "      <th>movieId</th>\n",
       "      <th>rating</th>\n",
       "      <th>timestamp</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>196</td>\n",
       "      <td>242</td>\n",
       "      <td>3.0</td>\n",
       "      <td>881250949</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>186</td>\n",
       "      <td>302</td>\n",
       "      <td>3.0</td>\n",
       "      <td>891717742</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>22</td>\n",
       "      <td>377</td>\n",
       "      <td>1.0</td>\n",
       "      <td>878887116</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>244</td>\n",
       "      <td>51</td>\n",
       "      <td>2.0</td>\n",
       "      <td>880606923</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>166</td>\n",
       "      <td>346</td>\n",
       "      <td>1.0</td>\n",
       "      <td>886397596</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "   userId  movieId  rating  timestamp\n",
       "0     196      242     3.0  881250949\n",
       "1     186      302     3.0  891717742\n",
       "2      22      377     1.0  878887116\n",
       "3     244       51     2.0  880606923\n",
       "4     166      346     1.0  886397596"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "url = \"http://files.grouplens.org/datasets/movielens/ml-100k/u.data\"\n",
    "data_pd = pd.read_csv(url, sep='\\t', names=['userId', 'movieId', 'rating', 'timestamp'])\n",
    "assert len(data_pd) == 100000\n",
    "\n",
    "data_pd['rating'] = data_pd['rating'].astype(float)\n",
    "data_pd.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Train vs test: 69854 vs 30146\n"
     ]
    }
   ],
   "source": [
    "schema = StructType([\n",
    "    StructField('userId', IntegerType()),\n",
    "    StructField('movieId', IntegerType()),\n",
    "    StructField('rating', FloatType()),\n",
    "    StructField('timestamp', LongType()),\n",
    "])\n",
    "data_df = spark.createDataFrame(data_pd, schema=schema)\n",
    "\n",
    "train, test = data_df.randomSplit([0.7, 0.3], 123)\n",
    "print(\"Train vs test: {} vs {}\".format(\n",
    "    train.cache().count(),\n",
    "    test.cache().count()))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Training\n",
    "Train multiple ALS with bootstraping sampling. To add more diversity in the ensemble, randomize some of the ALS hyper params too.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Training model 0 {'userCol': 'userId', 'itemCol': 'movieId', 'ratingCol': 'rating', 'rank': 31, 'maxIter': 15, 'implicitPrefs': True, 'alpha': 15.023938133735445, 'regParam': 0.012762680426829758, 'coldStartStrategy': 'drop', 'nonnegative': True}\n",
      "Training model 1 {'userCol': 'userId', 'itemCol': 'movieId', 'ratingCol': 'rating', 'rank': 22, 'maxIter': 15, 'implicitPrefs': True, 'alpha': 33.476069414287075, 'regParam': 0.1558492757743099, 'coldStartStrategy': 'drop', 'nonnegative': True}\n",
      "Training model 2 {'userCol': 'userId', 'itemCol': 'movieId', 'ratingCol': 'rating', 'rank': 48, 'maxIter': 15, 'implicitPrefs': True, 'alpha': 35.92503483617226, 'regParam': 0.1519447821933119, 'coldStartStrategy': 'drop', 'nonnegative': True}\n",
      "Training model 3 {'userCol': 'userId', 'itemCol': 'movieId', 'ratingCol': 'rating', 'rank': 40, 'maxIter': 15, 'implicitPrefs': True, 'alpha': 25.814073409363534, 'regParam': 0.172423724155834, 'coldStartStrategy': 'drop', 'nonnegative': True}\n",
      "Training model 4 {'userCol': 'userId', 'itemCol': 'movieId', 'ratingCol': 'rating', 'rank': 44, 'maxIter': 15, 'implicitPrefs': True, 'alpha': 23.639870337785517, 'regParam': 0.18802679059097302, 'coldStartStrategy': 'drop', 'nonnegative': True}\n",
      "Training model 5 {'userCol': 'userId', 'itemCol': 'movieId', 'ratingCol': 'rating', 'rank': 34, 'maxIter': 15, 'implicitPrefs': True, 'alpha': 37.56181477549837, 'regParam': 0.12818447130643557, 'coldStartStrategy': 'drop', 'nonnegative': True}\n",
      "Training model 6 {'userCol': 'userId', 'itemCol': 'movieId', 'ratingCol': 'rating', 'rank': 31, 'maxIter': 15, 'implicitPrefs': True, 'alpha': 22.016407353245434, 'regParam': 0.08303686256944531, 'coldStartStrategy': 'drop', 'nonnegative': True}\n",
      "Training model 7 {'userCol': 'userId', 'itemCol': 'movieId', 'ratingCol': 'rating', 'rank': 32, 'maxIter': 15, 'implicitPrefs': True, 'alpha': 22.004864102931514, 'regParam': 0.09936984198460991, 'coldStartStrategy': 'drop', 'nonnegative': True}\n",
      "Training model 8 {'userCol': 'userId', 'itemCol': 'movieId', 'ratingCol': 'rating', 'rank': 39, 'maxIter': 15, 'implicitPrefs': True, 'alpha': 32.10002281523691, 'regParam': 0.12394289598599252, 'coldStartStrategy': 'drop', 'nonnegative': True}\n",
      "Training model 9 {'userCol': 'userId', 'itemCol': 'movieId', 'ratingCol': 'rating', 'rank': 38, 'maxIter': 15, 'implicitPrefs': True, 'alpha': 38.60438669407002, 'regParam': 0.0852279119008048, 'coldStartStrategy': 'drop', 'nonnegative': True}\n"
     ]
    }
   ],
   "source": [
    "params = {\n",
    "    'userCol': 'userId',\n",
    "    'itemCol': 'movieId',\n",
    "    'ratingCol': 'rating',\n",
    "    'rank': (20, 50),\n",
    "    'maxIter': 15,\n",
    "    'implicitPrefs': True,\n",
    "    'alpha': (0.1, 40.0),\n",
    "    'regParam': (0.01, 0.2),\n",
    "    'coldStartStrategy': 'drop',\n",
    "    'nonnegative': True\n",
    "}\n",
    "\n",
    "bagging = RecoBagging(\n",
    "    ALS,\n",
    "    num_models=NUM_MODELS,\n",
    "    user_col='userId', item_col='movieId', rating_col='rating',\n",
    "    **params\n",
    ")\n",
    "\n",
    "bagging.fit(train)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Testing\n",
    "Recommend top-k movies for each user"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Recommending by 0\n",
      "Recommending by 1\n",
      "Recommending by 2\n",
      "Recommending by 3\n",
      "Recommending by 4\n",
      "Recommending by 5\n",
      "Recommending by 6\n",
      "Recommending by 7\n",
      "Recommending by 8\n",
      "Recommending by 9\n",
      "+------+--------------------+\n",
      "|userId|     recommendations|\n",
      "+------+--------------------+\n",
      "|   148|[[432, 1.26218629...|\n",
      "|   463|[[277, 0.98066174...|\n",
      "|   471|[[220, 0.78397213...|\n",
      "|   496|[[1084, 0.8240563...|\n",
      "|   833|[[39, 1.758559763...|\n",
      "|   243|[[283, 1.38636025...|\n",
      "|   392|[[242, 0.95355752...|\n",
      "|   540|[[288, 1.18128086...|\n",
      "|   623|[[210, 0.89439146...|\n",
      "|   737|[[191, 1.15225412...|\n",
      "|   858|[[286, 2.58612694...|\n",
      "|   897|[[148, 0.87238724...|\n",
      "|    31|[[513, 1.44179504...|\n",
      "|   516|[[169, 0.76956390...|\n",
      "|    85|[[1020, 1.3456176...|\n",
      "|   137|[[50, 2.460963522...|\n",
      "|   251|[[181, 1.76799780...|\n",
      "|   451|[[881, 2.24169102...|\n",
      "|   580|[[300, 1.04780128...|\n",
      "|   808|[[327, 1.28745436...|\n",
      "+------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "recommendations = bagging.recommend_k_items(test, top_k=TOP_K, merge_by='sum', scale=True)\n",
    "recommendations.cache().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Evaluation\n",
    "Evaluate the recommendation results by using [Spark's ranking metrics](https://spark.apache.org/docs/2.2.0/mllib-evaluation-metrics.html#ranking-systems). Since the Spark mllib's is based on RDD, a DataFrame wrapper class `RankingEvaluator` is implemented and used here. \n",
    "For more information about the evaluation metrics, see [link](https://en.wikipedia.org/wiki/Evaluation_measures_(information_retrieval))."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"\n",
    "Result of bagging ensemble of ALS\n",
    "\"\"\"\n",
    "ranking_evaluator = RankingEvaluator(test, reco_df=recommendations,\n",
    "    user_col='userId', item_col='movieId', rating_col='rating', reco_col='recommendations')\n",
    "\n",
    "result_bagging = {}\n",
    "result_bagging['model'] = \"Bagging(ALS)\"\n",
    "result_bagging['ndcg@k'] = ranking_evaluator.ndcgAt(TOP_K) \n",
    "result_bagging['precision@k'] = ranking_evaluator.precisionAt(TOP_K)\n",
    "result_bagging['recall@k'] = ranking_evaluator.recallAt(TOP_K)\n",
    "result_bagging['map'] = ranking_evaluator.meanAveragePrecision()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"\n",
    "Results of the individual ALS in the ensemble and their max and mean\n",
    "\"\"\"\n",
    "results = {\n",
    "    'model': [],\n",
    "    'ndcg@k': [],\n",
    "    'precision@k': [],\n",
    "    'recall@k': [],\n",
    "    'map': []\n",
    "}\n",
    "\n",
    "for i in range(bagging.num_models):\n",
    "    reco = bagging.reco_lists_df.filter(F.col('model') == i)\n",
    "        \n",
    "    rank_eval = RankingEvaluator(test, reco_df=reco,\n",
    "        user_col='userId', item_col='movieId', rating_col='rating', reco_col='recommendations')\n",
    "    results['model'].append(\"ALS \" + str(i+1))\n",
    "    results['ndcg@k'].append(rank_eval.ndcgAt(TOP_K))\n",
    "    results['precision@k'].append(rank_eval.precisionAt(TOP_K))\n",
    "    results['recall@k'].append(rank_eval.recallAt(TOP_K))\n",
    "    results['map'].append(rank_eval.meanAveragePrecision())\n",
    "\n",
    "result_max = {}\n",
    "result_max['model'] = \"Max(ALS)\"\n",
    "result_max['ndcg@k'] = max(results['ndcg@k'])\n",
    "result_max['precision@k'] = max(results['precision@k'])\n",
    "result_max['recall@k'] = max(results['recall@k'])\n",
    "result_max['map'] = max(results['map'])\n",
    "\n",
    "result_avg = {}\n",
    "result_avg['model'] = \"Avg(ALS)\"\n",
    "result_avg['ndcg@k'] = sum(results['ndcg@k']) /  len(results['ndcg@k']) \n",
    "result_avg['precision@k'] = sum(results['precision@k']) / len(results['precision@k'])\n",
    "result_avg['recall@k'] = sum(results['recall@k']) / len(results['recall@k'])\n",
    "result_avg['map'] = sum(results['map']) / len(results['map'])\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>model</th>\n",
       "      <th>ndcg@k</th>\n",
       "      <th>precision@k</th>\n",
       "      <th>recall@k</th>\n",
       "      <th>map</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>ALS 1</td>\n",
       "      <td>0.147004</td>\n",
       "      <td>0.140297</td>\n",
       "      <td>0.075325</td>\n",
       "      <td>0.029800</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>ALS 2</td>\n",
       "      <td>0.116004</td>\n",
       "      <td>0.112513</td>\n",
       "      <td>0.061447</td>\n",
       "      <td>0.022492</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>ALS 3</td>\n",
       "      <td>0.122784</td>\n",
       "      <td>0.121103</td>\n",
       "      <td>0.065097</td>\n",
       "      <td>0.024557</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>ALS 4</td>\n",
       "      <td>0.130224</td>\n",
       "      <td>0.129692</td>\n",
       "      <td>0.067485</td>\n",
       "      <td>0.025508</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>ALS 5</td>\n",
       "      <td>0.144424</td>\n",
       "      <td>0.141145</td>\n",
       "      <td>0.074903</td>\n",
       "      <td>0.028455</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>5</th>\n",
       "      <td>ALS 6</td>\n",
       "      <td>0.119779</td>\n",
       "      <td>0.117709</td>\n",
       "      <td>0.064514</td>\n",
       "      <td>0.023684</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>6</th>\n",
       "      <td>ALS 7</td>\n",
       "      <td>0.140104</td>\n",
       "      <td>0.134464</td>\n",
       "      <td>0.074453</td>\n",
       "      <td>0.029743</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>7</th>\n",
       "      <td>ALS 8</td>\n",
       "      <td>0.133569</td>\n",
       "      <td>0.130753</td>\n",
       "      <td>0.067657</td>\n",
       "      <td>0.026140</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>8</th>\n",
       "      <td>ALS 9</td>\n",
       "      <td>0.129028</td>\n",
       "      <td>0.125557</td>\n",
       "      <td>0.069247</td>\n",
       "      <td>0.026850</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>9</th>\n",
       "      <td>ALS 10</td>\n",
       "      <td>0.119595</td>\n",
       "      <td>0.118982</td>\n",
       "      <td>0.067190</td>\n",
       "      <td>0.024682</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>10</th>\n",
       "      <td>Max(ALS)</td>\n",
       "      <td>0.147004</td>\n",
       "      <td>0.141145</td>\n",
       "      <td>0.075325</td>\n",
       "      <td>0.029800</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>11</th>\n",
       "      <td>Avg(ALS)</td>\n",
       "      <td>0.130252</td>\n",
       "      <td>0.127222</td>\n",
       "      <td>0.068732</td>\n",
       "      <td>0.026191</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>12</th>\n",
       "      <td>Bagging(ALS)</td>\n",
       "      <td>0.181353</td>\n",
       "      <td>0.165854</td>\n",
       "      <td>0.086904</td>\n",
       "      <td>0.039094</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "           model    ndcg@k  precision@k  recall@k       map\n",
       "0          ALS 1  0.147004     0.140297  0.075325  0.029800\n",
       "1          ALS 2  0.116004     0.112513  0.061447  0.022492\n",
       "2          ALS 3  0.122784     0.121103  0.065097  0.024557\n",
       "3          ALS 4  0.130224     0.129692  0.067485  0.025508\n",
       "4          ALS 5  0.144424     0.141145  0.074903  0.028455\n",
       "5          ALS 6  0.119779     0.117709  0.064514  0.023684\n",
       "6          ALS 7  0.140104     0.134464  0.074453  0.029743\n",
       "7          ALS 8  0.133569     0.130753  0.067657  0.026140\n",
       "8          ALS 9  0.129028     0.125557  0.069247  0.026850\n",
       "9         ALS 10  0.119595     0.118982  0.067190  0.024682\n",
       "10      Max(ALS)  0.147004     0.141145  0.075325  0.029800\n",
       "11      Avg(ALS)  0.130252     0.127222  0.068732  0.026191\n",
       "12  Bagging(ALS)  0.181353     0.165854  0.086904  0.039094"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "results['model'].append(result_max['model'])\n",
    "results['ndcg@k'].append(result_max['ndcg@k'])\n",
    "results['precision@k'].append(result_max['precision@k'])\n",
    "results['recall@k'].append(result_max['recall@k'])\n",
    "results['map'].append(result_max['map'])\n",
    "\n",
    "results['model'].append(result_avg['model'])\n",
    "results['ndcg@k'].append(result_avg['ndcg@k'])\n",
    "results['precision@k'].append(result_avg['precision@k'])\n",
    "results['recall@k'].append(result_avg['recall@k'])\n",
    "results['map'].append(result_avg['map'])\n",
    "\n",
    "results['model'].append(result_bagging['model'])\n",
    "results['ndcg@k'].append(result_bagging['ndcg@k'])\n",
    "results['precision@k'].append(result_bagging['precision@k'])\n",
    "results['recall@k'].append(result_bagging['recall@k'])\n",
    "results['map'].append(result_bagging['map'])\n",
    "\n",
    "\n",
    "result_table = pd.DataFrame.from_dict(results)\n",
    "result_table"
   ]
  },
  {
   "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.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}