Nov_Problema_MovieLens_Engine.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# MovieLens API\n",
    "\n",
    "\n",
    "Full: 27,000,000 ratings and 1,100,000 tag applications applied to 58,000 movies by 280,000 users. Includes tag genome data with 14 million relevance scores across 1,100 tags. Last updated 9/2018\n",
    "Usando el Dataset MovieLens conseguir una solución a las siguientes actividades:\n",
    "\n",
    "Parte 1:\n",
    "\n",
    "- Descargar de manera automatizada el dataset.\n",
    "- Crear un RDD con el dataset\n",
    "- Entrenar un motor de recomendaciones usando ALS\n",
    "- Evaluar el rendimiento del motor\n",
    "\n",
    "Parte 2. Crear tres archivos que cumplan con las siguientes capacidades:\n",
    "\n",
    "**engine.py**\n",
    "\n",
    "Funciones del motor de recomendaciones.\n",
    "\n",
    "- `Función get_counts_and_averages`: Dada una tupla (movieID, ratings_iterable), regresa (movieID, (ratings_count, ratings_avg)\n",
    "- Clase `RecommendationEngine` con las siguientes funciones:\n",
    "    - `count_and_average_ratings`: Actualiza el conteo deratings de películas usando el DataFrame de ratings\n",
    "    - `train_model`: Entrena un modelo ALS con el DataFrame de ratings\n",
    "    - `predict_ratings`: Dada una tupla (userID, movieID) genera predicciones.\n",
    "        Regresa: Un RDD con formato (movieTitle, movieRating, numRatings)\n",
    "    - `add_ratings`: \n",
    "        - Añade nuevos ratings al DataFrame existente con el formato (user_id, movie_id, rating).\n",
    "        - calcula el conteo de ratings (count_and_average_ratings())\n",
    "        - Re-entrena el modelo con los nuevos ratings\n",
    "    - `get_ratings_for_movie_id`s: Dado un user_id y una lista de movie_ids, predice ratings para las películas\n",
    "    - `get_top_ratings`: Dada la tupla (user_id, movies_count) recomienda hasta  el valor establecido en movies_count para películas que no han sido votadas por el usuario user_id \n",
    "    - `__init__`: Instancia el DataFrame de Ratings, calcula el conteo de ratings y entrenamiento del modelo\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "**app.py**\n",
    "\n",
    "Router  de Fask-Blueprint que maneje las siguientes rutas y funciones:\n",
    "\n",
    "```python\n",
    "@main.route(\"/<int:user_id>/ratings/top/<int:count>\", methods=[\"GET\"])\n",
    "def top_ratings(user_id, count):\n",
    "    ...\n",
    "    return json.dumps(top_ratings)\n",
    "    \n",
    "\n",
    "```\n",
    "\n",
    "----\n",
    "\n",
    "```python\n",
    "@main.route(\"/<int:user_id>/ratings/<int:movie_id>\", methods=[\"GET\"])\n",
    "def movie_ratings(user_id, movie_id):\n",
    "    ...\n",
    "    return json.dumps(ratings)\n",
    "\n",
    "```\n",
    "\n",
    "----\n",
    "\n",
    "```python\n",
    "@main.route(\"/<int:user_id>/ratings\", methods = [\"POST\"])\n",
    "def add_ratings(user_id):\n",
    "    # get the ratings from the Flask POST request object\n",
    "    # create a list with the format required by the negine (user_id, movie_id, rating)\n",
    "    # add them to the model using then engine API\n",
    "    ...\n",
    "    return json.dumps(ratings)\n",
    "\n",
    "```\n",
    "\n",
    "----\n",
    "\n",
    "```python\n",
    "def create_app(spark_context, dataset_path):\n",
    "    global recommendation_engine \n",
    "\n",
    "    recommendation_engine = RecommendationEngine(spark_context, dataset_path)    \n",
    "    \n",
    "    app = Flask(__name__)\n",
    "    app.register_blueprint(main)\n",
    "    return app \n",
    "\n",
    "```\n",
    "\n",
    "Pista:\n",
    "Este router, debe importar la clase RecommendationEngine del script `engine.py`:\n",
    "```python\n",
    "from engine import RecommendationEngine\n",
    "```\n",
    "\n",
    "\n",
    "\n",
    "**server.py**\n",
    "\n",
    "El script que instanciará:\n",
    " * El servidor web para el router en **app.py**\n",
    " * Contexto de Spark para que la API creada en **engine.py** pueda funcionar\n",
    "     \n",
    "Este es el script que se somete a **spark-submit**:\n",
    "`$SPARK_HOME/bin/spark-submit --master local[*] --total-executor-cores 4 --executor-memory 8g server.py`\n",
    "\n",
    "---\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "## Parte 1:\n",
    "\n",
    "### dataset_downloader.sh\n",
    "\n",
    "\n",
    "Creamos el archivo `dataset_downloader.sh` con el siguiente contenido:\n",
    "\n",
    "```bash\n",
    "#!/usr/bin/env bash\n",
    "\n",
    "hash wget 2>/dev/null || { echo >&2 \"Wget required.  Aborting.\"; exit 1; }\n",
    "hash unzip 2>/dev/null || { echo >&2 \"unzip required.  Aborting.\"; exit 1; }\n",
    "\n",
    "wget http://files.grouplens.org/datasets/movielens/ml-latest.zip\n",
    "     http://files.grouplens.org/datasets/movielens/ml-latest.zip\n",
    "unzip -o \"ml-latest.zip\"\n",
    "DESTINATION=\"./datasets/\"\n",
    "mkdir -p $DESTINATION\n",
    "mv ml-latest $DESTINATION\n",
    "```\n",
    "---\n",
    "\n",
    "Una vez creado, ejecutamos los comandos:\n",
    "\n",
    "```bash\n",
    "\n",
    "chmod +x dataset_downloader.sh\n",
    "\n",
    "./dataset_downloader.sh\n",
    "```\n",
    "\n",
    "\n",
    "\n",
    "---\n",
    "\n",
    "## Modelo ALS\n",
    "\n",
    "Importamos las dependencias:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "\n",
    "from pyspark import SparkContext, SparkConf\n",
    "\n",
    "\n",
    "from pyspark.sql import Column, Row\n",
    "from pyspark.sql.functions import col, expr\n",
    "\n",
    "\n",
    "# Machine Learning\n",
    "\n",
    "from pyspark.ml.recommendation import ALS\n",
    "\n",
    "# Metrics / Eval\n",
    "\n",
    "from pyspark.ml.evaluation import RegressionEvaluator\n",
    "\n",
    "from pyspark.mllib.evaluation import RegressionMetrics, RankingMetrics, RegressionMetrics\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Instanciamos una sesión de Spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = SparkSession.builder \\\n",
    "    .master(\"local\") \\\n",
    "    .appName(\"API\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Generamos un DataFrame que lea la fuente `ratings.csv`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# dataset_path = \"./datasets/ml-latest/\"\n",
    "\n",
    "dataset_path = \"/home/jovyan/work/spark-movie-lens/datasets/ml-latest/\"\n",
    "\n",
    "# ratings = spark.read.option(\"header\",\"false\").text(os.path.join(dataset_path, 'ratings.csv'))\\\n",
    "#     .rdd.toDF()\\\n",
    "#     .selectExpr(\"split(value, ',') as col\")\\\n",
    "#     .selectExpr(\n",
    "#                 \"cast(col[0] as int) as userId\",\n",
    "#                 \"cast(col[1] as int) as movieId\",\n",
    "#                 \"cast(col[2] as float) as rating\",\n",
    "#                 \"cast(col[3] as long) as timestamp\")\\\n",
    "#     .dropna()\\\n",
    "#     .cache()\n",
    "ratings = spark.read.option(\"header\",\"false\").text(os.path.join(dataset_path, 'ratings.csv'))\\\n",
    "    .rdd.toDF()\\\n",
    "    .selectExpr(\"split(value, ',') as col\")\\\n",
    "    .selectExpr(\n",
    "                \"cast(col[0] as int) as userId\",\n",
    "                \"cast(col[1] as int) as movieId\",\n",
    "                \"cast(col[2] as float) as rating\")\\\n",
    "    .dropna()\\\n",
    "    .cache()\n",
    "\n",
    "\n",
    "\n",
    "ratings_short = ratings.sample(fraction=.0001, seed=42)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "ratings_short.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "training, test = ratings_short.randomSplit([0.8, 0.2])\n",
    "als = ALS()\\\n",
    "    .setMaxIter(5)\\\n",
    "    .setRegParam(0.01)\\\n",
    "    .setUserCol(\"userId\")\\\n",
    "    .setItemCol(\"movieId\")\\\n",
    "    .setRatingCol(\"rating\")\n",
    "print(als.explainParams())\n",
    "\n",
    "alsModel = als.fit(training)\n",
    "predictions = alsModel.transform(test)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can now output the top 𝘬 recommendations for each user or movie. The model’s recommendForAllUsers method returns a DataFrame of a userId, an array of recommendations, as well as a rating for each of those movies. recommendForAllItems returns a DataFrame of a movieId, as well as the top users for that movie:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "alsModel.recommendForAllUsers(10)\\\n",
    "  .selectExpr(\"userId\", \"explode(recommendations)\").show()\n",
    "alsModel.recommendForAllItems(10)\\\n",
    "  .selectExpr(\"movieId\", \"explode(recommendations)\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Evaluators for Recommendation\n",
    "\n",
    "\n",
    "When covering the cold-start strategy, we can set up an automatic model evaluator when working with ALS. One thing that may not be immediately obvious is that this recommendation problem is really just a kind of regression problem. Since we’re predicting values (ratings) for given users, we want to optimize for reducing the total difference between our users’ ratings and the true values. We can do this using the same RegressionEvaluator that we saw in Chapter 27. You can place this in a pipeline to automate the training process. When doing this, you should also set the cold-start strategy to be drop instead of NaN and then switch it back to NaN when it comes time to actually make predictions in your production system:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "evaluator = RegressionEvaluator()\\\n",
    "  .setMetricName(\"rmse\")\\\n",
    "  .setLabelCol(\"rating\")\\\n",
    "  .setPredictionCol(\"prediction\")\n",
    "rmse = evaluator.evaluate(predictions)\n",
    "\n",
    "print(\"Root-mean-square error = %f\" % rmse)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Metrics\n",
    "Recommendation results can be measured using both the standard regression metrics and some recommendation-specific metrics. It should come as no surprise that there are more sophisticated ways of measuring recommendation success than simply evaluating based on regression. These metrics are particularly useful for evaluating your final model.\n",
    "\n",
    "\n",
    "### Regression Metrics\n",
    "\n",
    "\n",
    "We can recycle the regression metrics for recommendation. This is because we can simply see how close each prediction is to the actual rating for that user and item:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "regComparison = predictions.select(\"rating\", \"prediction\")\\\n",
    "  .rdd.map(lambda x: (x(0), x(1)))\n",
    "metrics = RegressionMetrics(regComparison)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Ranking Metrics\n",
    "\n",
    "More interestingly, we also have another tool: ranking metrics. A RankingMetric allows us to compare our recommendations with an actual set of ratings (or preferences) expressed by a given user. RankingMetric does not focus on the value of the rank but rather whether or not our algorithm recommends an already ranked item again to a user. This does require some data preparation on our part. You may want to refer to Part II for a refresher on some of the methods. First, we need to collect a set of highly ranked movies for a given user. In our case, we’re going to use a rather low threshold: movies ranked above 2.5. Tuning this value will largely be a business decision:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "perUserActual = predictions\\\n",
    "  .where(\"rating > 2.5\")\\\n",
    "  .groupBy(\"userId\")\\\n",
    "  .agg(expr(\"collect_set(movieId) as movies\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "At this point, we have a collection of users, along with a truth set of previously ranked movies for each user. Now we will get our top 10 recommendations from our algorithm on a per-user basis. We will then see if the top 10 recommendations show up in our truth set. If we have a well-trained model, it will correctly recommend the movies a user already liked. If it doesn’t, it may not have learned enough about each particular user to successfully reflect their preferences:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "perUserPredictions = predictions\\\n",
    "  .orderBy(col(\"userId\"), expr(\"prediction DESC\"))\\\n",
    "  .groupBy(\"userId\")\\\n",
    "  .agg(expr(\"collect_list(movieId) as movies\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we have two DataFrames, one of predictions and another the top-ranked items for a particular user. We can pass them into the RankingMetrics object. This object accepts an RDD of these combinations, as you can see in the following join and RDD conversion:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "perUserActualvPred = perUserActual.join(perUserPredictions, [\"userId\"]).rdd\\\n",
    "  .map(lambda row: (row[1], row[2][:15]))\n",
    "ranks = RankingMetrics(perUserActualvPred)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we can see the metrics from that ranking. For instance, we can see how precise our algorithm is with the mean average precision. We can also get the precision at certain ranking points, for instance, to see where the majority of the positive recommendations fall:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ranks.meanAveragePrecision"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ranks.precisionAt(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  },
  {
   "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.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}