recomendador_pyspark.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Recomendador en Python\n",
    "\n",
    "#### Kevin Craig Alisauskas\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Importamos librerías y cargamos pyspark (modificar como sea necesario según el pc)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Nothing to stop\n"
     ]
    }
   ],
   "source": [
    "import numpy as np\n",
    "import pandas as pd\n",
    "import warnings\n",
    "warnings.filterwarnings('ignore')\n",
    "\n",
    "import findspark\n",
    "findspark.init('/home/kubote/spark/spark-2.4.1-bin-hadoop2.7/')\n",
    "\n",
    "import pyspark\n",
    "from pyspark.sql import Row\n",
    "\n",
    "try:\n",
    "    sc.stop()\n",
    "    print(\"Stopped and restarted\")\n",
    "except:\n",
    "    print(\"Nothing to stop\")\n",
    "\n",
    "from pyspark import SparkContext\n",
    "from pyspark.sql import SQLContext\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "sc = SparkContext(master = \"local[*]\")\n",
    "sqlContext = SQLContext(sc)\n",
    "\n",
    "spark = SparkSession.builder.getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Importamos los datos"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Leemos las filas del dataset una por una\n",
    "lines = spark.read.text(\"./Datos/ratings.dat\").rdd\n",
    "lines2 = spark.read.text(\"./Datos/movies.dat\").rdd\n",
    "\n",
    "# Dividimos las filas en columnas.\n",
    "parts = lines.map(lambda row: row.value.split(\"::\"))\n",
    "parts2 = lines2.map(lambda row: row.value.split(\"::\"))\n",
    "\n",
    "# Definimos cada variable con su tipo y nombre\n",
    "ratingsRDD = parts.map(lambda p: Row(UserID = int(p[0]), MovieID = int(p[1]), Rating = float(p[2]), Timestamp = int(p[3])))\n",
    "moviesRDD = parts2.map(lambda p: Row(MovieID = int(p[0]), Title = str(p[1]), Genres = str(p[2])))\n",
    "\n",
    "# Lo transformamos en Dataframe.\n",
    "ratings = spark.createDataFrame(ratingsRDD)\n",
    "ratings = ratings.drop('Timestamp')\n",
    "\n",
    "movies = spark.createDataFrame(moviesRDD)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "En la implementación en Python realizamos dos pasos, una selección de usuarios y películas posibles por Nearest Neighbors y una selección de películas por la correlación entre las posibles y las vistas por el usuario a recomendar. \n",
    "En esta ocasión, existe bastante documentación sobre el filtrado colaborativo en pyspark utilizando la función ALS (Alternating Least Squares), sin embargo, la implementación es muy sencilla y se limita a cargar los datos y ejecutar ALS, indicando las columnas de Item, Usuarios y Ratings, con la posibilidad de mejorar los parámetros del modelo para aumentar su rendimiento. Dada la sencillez de ese método, he decidido adaptar de la mejor manera posible el método utilizado en la implementación en Python, que procedo a explicar paso a paso junto con el código, ya que he tenido que enfrentar algunos problemas."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Para empezar, guardamos los valores únicos de películas y usuarios."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "distinct_movies = ratings.select('MovieID').distinct().count()\n",
    "distinct_users = ratings.select('UserID').distinct().count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Creamos una tabla de contingencia para obtener posibles usuarios similares según las películas vistas por ambos."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "crossed = ratings.crosstab(\"UserID\", \"MovieID\")\n",
    "\n",
    "# Necesitaremos los id de los usuarios posteriormente.\n",
    "users_total_id = [int(row['UserID_MovieID']) for row in crossed.select('UserID_MovieID').collect()]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Comprobamos que la tabla tenga las dimensiones correctas."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "El número de filas de la tabla de contingencia es correcto:  True\n",
      "El número de columnas de la tabla de contingencia es correcto:  True\n"
     ]
    }
   ],
   "source": [
    "cierto = True if distinct_users == crossed.count() else False\n",
    "print(\"El número de filas de la tabla de contingencia es correcto: \", cierto)\n",
    "\n",
    "cierto = True if distinct_movies == len(crossed.columns) - 1 else False\n",
    "print(\"El número de columnas de la tabla de contingencia es correcto: \", cierto)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Ahora bien, puestos a utilizar Nearest Neighbors como en Python, nos encotramos con que no existe en Spark. Así pues, vamos a usar la tabla de contingencia para obtener un KMeans de 100 clústers, de esa forma encontraremos como pareja del seleccionado por una variable user_id_position (posición en la lista de usuarios) al perteneciente a su mismo cluster."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "user_id_position = 1\n",
    "\n",
    "# Número de clústers\n",
    "n_clus = 100"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Transformamos la matriz para el KMeans.\n",
    "from pyspark.ml.feature import VectorAssembler\n",
    "\n",
    "vec_assembler = VectorAssembler(inputCols = crossed.columns[1:], outputCol='features')\n",
    "kmeans_data = vec_assembler.transform(crossed)\n",
    "\n",
    "# Ajustamos el kmeans.\n",
    "from pyspark.ml.clustering import KMeans\n",
    "\n",
    "kmeans = KMeans(featuresCol='features', k = n_clus)\n",
    "model = kmeans.fit(kmeans_data)\n",
    "\n",
    "# Obtenemos las predicciones (esto añade una columna \"prediction\" al dataframe).\n",
    "predictions = model.transform(kmeans_data)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Buscamos el clúster al que pertenece nuestro usuario en la posición 1 (0 en python, el primero) y lo mostramos."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "El clúster que buscamos es el : 9\n"
     ]
    }
   ],
   "source": [
    "user_id = users_total_id[user_id_position]\n",
    "\n",
    "cluster_id = predictions.filter(predictions.UserID_MovieID == user_id).select('prediction').collect()[0]['prediction']\n",
    "print(\"El clúster que buscamos es el :\", cluster_id)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Filtramos los usuarios pertenecientes a el cluster buscado y seleccionamos uno aleatorio (similar_user) a partir del cual recomendar películas a user_id."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Posibles usuarios (pertenecientes al mismo cluster)\n",
    "possible_users = predictions.filter(predictions.prediction == cluster_id).filter(predictions.UserID_MovieID != user_id).select('UserID_MovieID')\n",
    "\n",
    "# Usuario aleatorio dentro del mismo cluster.\n",
    "similar_user = np.random.randint(possible_users.count())\n",
    "similar_user_movies = predictions.filter(predictions.UserID_MovieID == possible_users.collect()[similar_user]['UserID_MovieID'])\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Calculamos las diferencias con el vector del usuario seleccionado con el similar. Primero definimos los arrays de películas similares y películas vistas por el usuario.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "similar_user_movies_array = np.array(similar_user_movies.drop(\"UserID_MovieID\", \"features\", \"prediction\").collect())\n",
    "\n",
    "user_movies_array = np.array(predictions.filter(predictions.UserID_MovieID == user_id).drop(\"UserID_MovieID\", \"features\", \"prediction\").collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Ahroa obtendremos las películas posibles, es decir, las películas vistas por el usuario similar y no vistas por el usuario para el cual recomendar."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Películas vistas por el usuario 645\n",
      "['1' '111' '1136' '1148' '1198' '1204' '1206' '1208' '1213' '1221' '1225'\n",
      " '1227' '1228' '1230' '1252' '1256' '1288' '1293' '1299' '1537' '1836'\n",
      " '2028' '2064' '2300' '2324' '2396' '260' '2858' '2937' '2966' '3077'\n",
      " '3114' '3210' '34' '3424' '3425' '3929' '593' '608' '68' '745' '750'\n",
      " '903' '904']\n",
      "Películas Posibles: \n",
      "['1022' '1028' '1032' '1035' '1081' '1083' '1088' '1210' '1380' '1937'\n",
      " '1943' '1947' '1951' '2078' '2081' '2087' '2096' '2565' '2628' '2746'\n",
      " '2941' '2946' '3061' '3199' '3545' '3549' '3604' '3605' '364' '3675'\n",
      " '588' '899' '900' '912' '918' '938']\n"
     ]
    }
   ],
   "source": [
    "# Y ahora la diferencia.\n",
    "diff = similar_user_movies_array - user_movies_array\n",
    "\n",
    "# Los códigos de las películas.\n",
    "id_peliculas = np.array(similar_user_movies.drop(\"UserID_MovieID\", \"features\", \"prediction\").columns)\n",
    "\n",
    "# Ahora, las posiciones de las posibles películas (las que tengan valor 1).\n",
    "Lposibles = id_peliculas[np.where(diff == 1)[1]]\n",
    "\n",
    "# Y las películas vistas por el usuario user_id.\n",
    "vistas_user = id_peliculas[np.where(user_movies_array == 1)[1]]\n",
    "\n",
    "# Mostramos las películas vistas y posibles.\n",
    "print(\"Películas vistas por el usuario\", user_id)\n",
    "print(vistas_user)\n",
    "\n",
    "print(\"Películas Posibles: \")\n",
    "print(Lposibles)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Ahora queda el paso de elegir la película a recomendar de entre todas las posibles. La idea original era crear una matriz cruzada con los usuarios como filas, las películas como columnas y los ratings como valores, para de ahí calcular la correlación entre películas, sin embargo, aunque he implementado ese método tanto con la distancia euclidea como con la similitud del coseno entre dos columnas del dataframe, es imposible de calcular con mi equipo, surgen errores de memoria que no se solucionar. Dejaré el código de la implementación comentado al final del trabajo.\n",
    "\n",
    "Así pues, simplemente vamos a seleccionar la película con mayor rating medio de todas las películas posibles."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Recomendacion Final: \n",
      "Película con MovieID:  912\n",
      "+-----------------+\n",
      "|            Title|\n",
      "+-----------------+\n",
      "|Casablanca (1942)|\n",
      "+-----------------+\n",
      "\n",
      "None\n"
     ]
    }
   ],
   "source": [
    "# Calculo las medias\n",
    "from pyspark.sql.functions import mean\n",
    "medias = ratings.groupby('MovieID').agg({'Rating': \"mean\"})\n",
    "\n",
    "medias_movies_id = np.array([int(row['MovieID']) for row in medias.select('MovieID').collect()])\n",
    "medias_movies_avg = [float(row['avg(Rating)']) for row in medias.select('avg(Rating)').collect()]\n",
    "\n",
    "# Almaceno las medias de únicamente las películas posibles y las comparo entre si para obtener la que tiene la mayor\n",
    "# que será la película elegida.\n",
    "if Lposibles.shape[0] > 0:\n",
    "\n",
    "    Lscores = []\n",
    "    i = 0\n",
    "\n",
    "    for p in Lposibles:\n",
    "\n",
    "        media = medias_movies_avg[np.where(medias_movies_id == int(p))[0][0]]\n",
    "        # media = ratings_pivoted.select(p).dropna().select(mean(p)).collect()[0]['avg(' + p + ')']\n",
    "\n",
    "        # Almacenamos la suma de las correlaciones de la posible película con las vistas.\n",
    "        Lscores.append(media)\n",
    "\n",
    "\n",
    "    irecom = int(np.array(Lscores).argmax())\n",
    "    print(\"Recomendacion Final: \")\n",
    "    print(\"Película con MovieID: \", Lposibles[irecom])\n",
    "    print(movies.filter(movies.MovieID == Lposibles[irecom]).select('Title').show())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "######################################################################################################################\n",
    "#####################################################################################################################"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A continuación dejo el código de la implementación original, que sería funcional si no fuera por los errores de memoria."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "# Creo la matriz cruzada.\n",
    "#ratings_pivoted = ratings.groupby(\"UserID\").pivot(\"MovieID\").avg(\"Rating\").fillna(0)\n",
    "\n",
    "# Defino las funciones de similitud del coseno y distancia euclidea.\n",
    "#import pyspark.sql.functions as func\n",
    "\n",
    "#def cosine_similarity(df, col1, col2):\n",
    "\n",
    "#    df_cosine = df.select(func.sum(df[col1] * df[col2]).alias('dot'),\n",
    "#                          func.sqrt(func.sum(df[col1]**2)).alias('norm1'),\n",
    "#                          func.sqrt(func.sum(df[col2] **2)).alias('norm2'))\n",
    "\n",
    "#    d = df_cosine.rdd.collect()[0].asDict()\n",
    "\n",
    "#    return d['dot']/(d['norm1'] * d['norm2'])\n",
    "\n",
    "#def euclidean_distance(df, col1, col2):\n",
    "\n",
    "#    df_cosine = df.select(func.sqrt(func.sum((df[col1] - df[col2])**2)).alias('dot'))\n",
    "\n",
    "#    d = df_cosine.rdd.collect()[0].asDict()\n",
    "\n",
    "#    return d['dot']\n",
    "\n",
    "\n",
    "# Ahora vamos a trabajar con la matriz ratings_pivoted, buscando la correlación entre películas a partir\n",
    "# de las valoraciones de usuarios.\n",
    "\n",
    "#if Lposibles.shape[0] > 0:\n",
    "\n",
    "#    Lscores = []\n",
    "#    i = 0\n",
    "\n",
    "#    for p in Lposibles:\n",
    "\n",
    "#        simil = 0\n",
    "\n",
    "#        for u in vistas_user:\n",
    "\n",
    "#            print(\"Iteración \", i, \" de \", Lposibles.shape[0]*vistas_user.shape[0])\n",
    "#            i = i + 1\n",
    "\n",
    "            # Calculamos la similitud entre la película posible y las vistas por el usuario.\n",
    "#            simil = simil + euclidean_distance(ratings_pivoted, p, u) # ratings_pivoted_filled.corr(p, u) la correlación \n",
    "            # da errores de memoria.\n",
    "\n",
    "        # Almacenamos la suma de las correlaciones de la posible película con las vistas.\n",
    "#        Lscores.append(simil)\n",
    "\n",
    "\n",
    "#    irecom = int(np.array(Lscores).argmin())\n",
    "#    print(\"Recomendacion Final: \")\n",
    "#    print(movies.filter(movies.MovieID == Lposibles[irecom]).select('Title').show())\n"
   ]
  }
 ],
 "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.7.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}