notebooks/00_quick_start/als_pyspark_movielens.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Running ALS on MovieLens (pySpark)\n",
    "\n",
    "[ALS](https://spark.apache.org/docs/latest/api/python/_modules/pyspark/ml/recommendation.html#ALS) (Alternating Least Squares) is a well-known collaborative filtering algorithm.\n",
    "\n",
    "This notebook provides an example of how to utilize and evaluate ALS pySpark ML (DataFrame-based API) implementation, meant for large-scale distributed datasets. We use a smaller dataset in this example to run ALS efficiently on Data Science Virtual Machine."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "System version: 3.6.0 | packaged by conda-forge | (default, Feb  9 2017, 14:36:55) \n",
      "[GCC 4.8.2 20140120 (Red Hat 4.8.2-15)]\n",
      "Spark version: 2.3.1\n"
     ]
    }
   ],
   "source": [
    "# set the environment path to find Recommenders\n",
    "import sys\n",
    "sys.path.append(\"../../\")\n",
    "import os\n",
    "import numpy as np\n",
    "from zipfile import ZipFile\n",
    "import papermill as pm\n",
    "import pyspark\n",
    "from pyspark.ml.recommendation import ALS\n",
    "import pyspark.sql.functions as F\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.types import StructType, StructField\n",
    "from pyspark.sql.types import StringType, FloatType, IntegerType\n",
    "\n",
    "from reco_utils.dataset.url_utils import maybe_download\n",
    "from reco_utils.dataset.spark_splitters import spark_random_split\n",
    "from reco_utils.evaluation.spark_evaluation import SparkRatingEvaluation, SparkRankingEvaluation\n",
    "\n",
    "\n",
    "print(\"System version: {}\".format(sys.version))\n",
    "print(\"Spark version: {}\".format(pyspark.__version__))\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Set the default parameters."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "tags": [
     "parameters"
    ]
   },
   "outputs": [],
   "source": [
    "# top k items to recommend\n",
    "TOP_K = 10\n",
    "\n",
    "# Select Movielens data size: 100k, 1m, 10m, or 20m\n",
    "MOVIELENS_DATA_SIZE = '100k'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 0. Set up Spark context\n",
    "\n",
    "The following settings work well for debugging locally on VM - change when running on a cluster. We set up a giant single executor with many threads and specify memory cap. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# the following settings work well for debugging locally on VM - change when running on a cluster\n",
    "# set up a giant single executor with many threads and specify memory cap\n",
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"ALS pySpark\") \\\n",
    "    .master(\"local[*]\") \\\n",
    "    .config(\"spark.driver.memory\", \"2g\")\\\n",
    "    .config(\"spark.executor.cores\", \"32\")\\\n",
    "    .config(\"spark.executor.memory\", \"8g\")\\\n",
    "    .config(\"spark.memory.fraction\", \"0.9\")\\\n",
    "    .config(\"spark.memory.stageFraction\", \"0.3\")\\\n",
    "    .config(\"spark.executor.instances\", 1)\\\n",
    "    .config(\"spark.executor.heartbeatInterval\", \"36000s\")\\\n",
    "    .config(\"spark.network.timeout\", \"10000000s\")\\\n",
    "    .config(\"spark.driver.maxResultSize\", \"50g\")\\\n",
    "    .getOrCreate()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 1. Download the MovieLens dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# MovieLens data have different data-format for each size of dataset\n",
    "data_header = False\n",
    "if MOVIELENS_DATA_SIZE == \"100k\":\n",
    "    separator = \"\\t\"\n",
    "    data_name = \"u.data\"\n",
    "    data_folder = \"ml-100k\"\n",
    "elif MOVIELENS_DATA_SIZE == \"1m\":\n",
    "    separator = \"::\"\n",
    "    data_name = \"ratings.dat\"\n",
    "    data_folder = \"ml-1m\"\n",
    "elif MOVIELENS_DATA_SIZE == \"10m\":\n",
    "    separator = \"::\"\n",
    "    data_name = \"ratings.dat\"\n",
    "    data_folder = \"ml-10M100K\"\n",
    "elif MOVIELENS_DATA_SIZE == \"20m\":\n",
    "    separator = \",\"\n",
    "    data_name = \"ratings.csv\"\n",
    "    data_folder = \"ml-20m\"\n",
    "    data_header = True\n",
    "else:\n",
    "    raise ValueError(\"Invalid data size. Should be one of {100k, 1m, 10m, or 20m}\")\n",
    "\n",
    "# Download dataset zip file and decompress if haven't done yet\n",
    "data_path = os.path.join(data_folder, data_name)\n",
    "if not os.path.exists(data_path):\n",
    "    filename = \"ml-\" + MOVIELENS_DATA_SIZE + \".zip\"\n",
    "    filepath = maybe_download(\n",
    "        \"http://files.grouplens.org/datasets/movielens/\" + filename, filename\n",
    "    )\n",
    "\n",
    "    with ZipFile(filepath, \"r\") as zf:\n",
    "        zf.extractall()\n",
    "\n",
    "    # remove zip file we already used\n",
    "    os.remove(filepath)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+------+---------+\n",
      "|UserId|MovieId|Rating|Timestamp|\n",
      "+------+-------+------+---------+\n",
      "|   196|    242|   3.0|881250949|\n",
      "|   186|    302|   3.0|891717742|\n",
      "|    22|    377|   1.0|878887116|\n",
      "|   244|     51|   2.0|880606923|\n",
      "|   166|    346|   1.0|886397596|\n",
      "|   298|    474|   4.0|884182806|\n",
      "|   115|    265|   2.0|881171488|\n",
      "|   253|    465|   5.0|891628467|\n",
      "|   305|    451|   3.0|886324817|\n",
      "|     6|     86|   3.0|883603013|\n",
      "|    62|    257|   2.0|879372434|\n",
      "|   286|   1014|   5.0|879781125|\n",
      "|   200|    222|   5.0|876042340|\n",
      "|   210|     40|   3.0|891035994|\n",
      "|   224|     29|   3.0|888104457|\n",
      "|   303|    785|   3.0|879485318|\n",
      "|   122|    387|   5.0|879270459|\n",
      "|   194|    274|   2.0|879539794|\n",
      "|   291|   1042|   4.0|874834944|\n",
      "|   234|   1184|   2.0|892079237|\n",
      "+------+-------+------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Note: The DataFrame-based API for ALS currently only supports integers for user and item ids.\n",
    "schema = StructType(\n",
    "    (\n",
    "        StructField(\"UserId\", IntegerType()),\n",
    "        StructField(\"MovieId\", IntegerType()),\n",
    "        StructField(\"Rating\", FloatType()),\n",
    "        StructField(\"Timestamp\", IntegerType()),\n",
    "    )\n",
    ")\n",
    "\n",
    "# pySpark's read csv currently doesn't support multi-character delimiter, thus we manually handle that\n",
    "if len(separator) > 1:\n",
    "    raw_data = spark.sparkContext.textFile(data_path) \\\n",
    "        .map(lambda l: l.split(separator)).map(lambda s: ','.join(map(str, s)))\n",
    "    data = spark.read.csv(raw_data, schema=schema, sep=',', header=data_header)\n",
    "else:\n",
    "    data = spark.read.csv(data_path, schema=schema, sep=separator, header=data_header)\n",
    "\n",
    "data.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 2. Split the data using the Spark random splitter provided in utilities"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "N train 75193\n",
      "N test 24807\n"
     ]
    }
   ],
   "source": [
    "train, test = spark_random_split(data, ratio=0.75, seed=123)\n",
    "print (\"N train\", train.count())\n",
    "print (\"N test\", test.count())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3. Train the ALS model on the training data, and get the top-k recommendations for our testing data\n",
    "\n",
    "To predict movie ratings, we use the rating data in the training set as users' explicit feedbacks.\n",
    "\n",
    "When our goal is to recommend top k movies a user is likely to watch, on the other hand, we utilize the ratings as implicit feedbacks."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "header = {\n",
    "    \"userCol\": \"UserId\",\n",
    "    \"itemCol\": \"MovieId\",\n",
    "    \"ratingCol\": \"Rating\",\n",
    "}\n",
    "\n",
    "\n",
    "# implicitPrefs=True for recommendation, False for rating prediction\n",
    "als = ALS(\n",
    "    rank=40,\n",
    "    maxIter=15,\n",
    "    implicitPrefs=True,\n",
    "    alpha=0.1,\n",
    "    regParam=0.01,\n",
    "    coldStartStrategy='drop',\n",
    "    nonnegative=True,\n",
    "    **header\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "model = als.fit(train)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------------------+\n",
      "|UserId|     recommendations|\n",
      "+------+--------------------+\n",
      "|   471|[[71, 0.57152903]...|\n",
      "|   463|[[285, 1.1577708]...|\n",
      "|   833|[[185, 1.1997313]...|\n",
      "|   496|[[204, 0.974352],...|\n",
      "|   148|[[181, 0.9232193]...|\n",
      "|   540|[[100, 1.0301069]...|\n",
      "|   392|[[302, 1.0355105]...|\n",
      "|   243|[[275, 0.8969344]...|\n",
      "|   623|[[181, 1.0735292]...|\n",
      "|   737|[[181, 0.47953993...|\n",
      "|   897|[[69, 1.0488142],...|\n",
      "|   858|[[286, 1.1212317]...|\n",
      "|    31|[[268, 0.78563535...|\n",
      "|   516|[[286, 0.6772065]...|\n",
      "|   580|[[405, 1.0718408]...|\n",
      "|   251|[[121, 1.228331],...|\n",
      "|   451|[[259, 1.2970893]...|\n",
      "|    85|[[170, 1.2421695]...|\n",
      "|   137|[[405, 1.103399],...|\n",
      "|   808|[[288, 0.8915927]...|\n",
      "+------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "recommendations = model.recommendForUserSubset(test, TOP_K)\n",
    "recommendations.show()\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+----------+\n",
      "|UserId|MovieId|    rating|\n",
      "+------+-------+----------+\n",
      "|   471|     71|0.57152903|\n",
      "|   471|    588|0.55487996|\n",
      "|   471|    432|0.55319667|\n",
      "|   471|    418|0.54289067|\n",
      "|   471|    501| 0.5107881|\n",
      "|   471|     99| 0.4807951|\n",
      "|   471|    419|0.44345152|\n",
      "|   471|     95|0.44070157|\n",
      "|   471|     91|0.43045348|\n",
      "|   471|    143|0.41670722|\n",
      "|   463|    285| 1.1577708|\n",
      "|   463|    124| 1.0746053|\n",
      "|   463|    286| 1.0422602|\n",
      "|   463|    116| 0.9322226|\n",
      "|   463|     14|0.91807663|\n",
      "|   463|    302| 0.9128054|\n",
      "|   463|    269|0.88212854|\n",
      "|   463|      1| 0.8625988|\n",
      "|   463|    283| 0.8478566|\n",
      "|   463|     13|0.84707767|\n",
      "+------+-------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Convert to reco util's ranking evaluator format\n",
    "top_k = recommendations.select('UserId', F.explode('recommendations').alias('r')) \\\n",
    "    .select('UserId', 'r.*')\n",
    "top_k.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4. Evaluate how well ALS performs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+------+---------+\n",
      "|UserId|MovieId|Rating|Timestamp|\n",
      "+------+-------+------+---------+\n",
      "|     1|      2|   3.0|876893171|\n",
      "|     1|      3|   4.0|878542960|\n",
      "|     1|      4|   3.0|876893119|\n",
      "|     1|      9|   5.0|878543541|\n",
      "|     1|     11|   2.0|875072262|\n",
      "|     1|     17|   3.0|875073198|\n",
      "|     1|     25|   4.0|875071805|\n",
      "|     1|     28|   4.0|875072173|\n",
      "|     1|     30|   3.0|878542515|\n",
      "|     1|     33|   4.0|878542699|\n",
      "|     1|     43|   4.0|878542869|\n",
      "|     1|     48|   5.0|875072520|\n",
      "|     1|     49|   3.0|878542478|\n",
      "|     1|     52|   4.0|875072205|\n",
      "|     1|     59|   5.0|876892817|\n",
      "|     1|     62|   3.0|878542282|\n",
      "|     1|     65|   4.0|875072125|\n",
      "|     1|     66|   4.0|878543030|\n",
      "|     1|     71|   3.0|876892425|\n",
      "|     1|     78|   1.0|878543176|\n",
      "+------+-------+------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "test.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "rank_eval = SparkRankingEvaluation(test, top_k, k = TOP_K, col_user=\"UserId\", col_item=\"MovieId\", \n",
    "                                    col_rating=\"Rating\", col_prediction=\"rating\", \n",
    "                                    relevancy_method=\"top_k\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Model:\tALS\n",
      "Top K:\t10\n",
      "MAP:\t0.022188\n",
      "NDCG:\t0.091370\n",
      "Precision@K:\t0.090127\n",
      "Recall@K:\t0.073033\n"
     ]
    }
   ],
   "source": [
    "print(\"Model:\\tALS\",\n",
    "      \"Top K:\\t%d\" % rank_eval.k,\n",
    "      \"MAP:\\t%f\" % rank_eval.map_at_k(),\n",
    "      \"NDCG:\\t%f\" % rank_eval.ndcg_at_k(),\n",
    "      \"Precision@K:\\t%f\" % rank_eval.precision_at_k(),\n",
    "      \"Recall@K:\\t%f\" % rank_eval.recall_at_k(), sep='\\n')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 5. Evaluate rating prediction"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+------+---------+----------+\n",
      "|UserId|MovieId|Rating|Timestamp|prediction|\n",
      "+------+-------+------+---------+----------+\n",
      "|   406|    148|   3.0|879540276| 3.0810654|\n",
      "|    27|    148|   3.0|891543129|  2.908636|\n",
      "|   606|    148|   3.0|878150506|  3.621529|\n",
      "|   916|    148|   2.0|880843892| 2.1849582|\n",
      "|   236|    148|   4.0|890117028|  4.351115|\n",
      "|   602|    148|   4.0|888638517|  4.241503|\n",
      "|   663|    148|   4.0|889492989|   4.01702|\n",
      "|   372|    148|   5.0|876869915| 4.4350476|\n",
      "|   190|    148|   4.0|891033742|   4.93678|\n",
      "|     1|    148|   2.0|875240799| 2.7891634|\n",
      "|   297|    148|   3.0|875239619|  3.849859|\n",
      "|   178|    148|   4.0|882824325| 3.9544783|\n",
      "|   308|    148|   3.0|887740788| 2.6153567|\n",
      "|   923|    148|   4.0|880387474|  4.268126|\n",
      "|    54|    148|   3.0|880937490| 4.2782016|\n",
      "|   430|    148|   2.0|877226047|  3.016966|\n",
      "|    92|    148|   2.0|877383934| 2.1105278|\n",
      "|   447|    148|   4.0|878854729|  4.211315|\n",
      "|   374|    148|   4.0|880392992|  4.876349|\n",
      "|   891|    148|   5.0|891639793|  3.756253|\n",
      "+------+-------+------+---------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "als_prediction = ALS(\n",
    "    rank=40,\n",
    "    maxIter=15,\n",
    "    implicitPrefs=False,\n",
    "    regParam=0.01,\n",
    "    coldStartStrategy='drop',\n",
    "    nonnegative=True,\n",
    "    **header\n",
    ")\n",
    "\n",
    "model_prediction = als_prediction.fit(train)\n",
    "\n",
    "prediction = model_prediction.transform(test)\n",
    "prediction.show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Model:\tALS rating prediction\n",
      "RMSE:\t1.12\n",
      "MAE:\t0.869609\n",
      "Explained variance:\t0.005666\n",
      "R squared:\t0.002667\n"
     ]
    }
   ],
   "source": [
    "rating_eval = SparkRatingEvaluation(test, prediction, col_user=\"UserId\", col_item=\"MovieId\", \n",
    "                                    col_rating=\"Rating\", col_prediction=\"prediction\")\n",
    "\n",
    "print(\"Model:\\tALS rating prediction\",\n",
    "      \"RMSE:\\t%.2f\" % rating_eval.rmse(),\n",
    "      \"MAE:\\t%f\" % rating_eval.mae(),\n",
    "      \"Explained variance:\\t%f\" % rating_eval.exp_var(),\n",
    "      \"R squared:\\t%f\" % rating_eval.rsquared(), sep='\\n')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/papermill.record+json": {
       "map": 0.022187601696326647
      }
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/papermill.record+json": {
       "ndcg": 0.09136992931956911
      }
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/papermill.record+json": {
       "precision": 0.09012738853503184
      }
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/papermill.record+json": {
       "recall": 0.07303304832753388
      }
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/papermill.record+json": {
       "rmse": 1.1195442239211102
      }
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/papermill.record+json": {
       "mae": 0.8696094802050408
      }
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/papermill.record+json": {
       "exp_var": 0.005666353905053212
      }
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/papermill.record+json": {
       "rsquared": 0.0026673840895518053
      }
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Record results with papermill for tests\n",
    "pm.record(\"map\", rank_eval.map_at_k())\n",
    "pm.record(\"ndcg\", rank_eval.ndcg_at_k())\n",
    "pm.record(\"precision\", rank_eval.precision_at_k())\n",
    "pm.record(\"recall\", rank_eval.recall_at_k())\n",
    "\n",
    "pm.record(\"rmse\", rating_eval.rmse())\n",
    "pm.record(\"mae\", rating_eval.mae())\n",
    "pm.record(\"exp_var\", rating_eval.exp_var())\n",
    "pm.record(\"rsquared\", rating_eval.rsquared())\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "celltoolbar": "Tags",
  "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": 1
}