Movie Recommender System Using ALS.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 147,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import sys\n",
    "from pyspark.rdd import RDD\n",
    "from pyspark.sql import Row\n",
    "from pyspark.sql import DataFrame\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import lit, col\n",
    "from pyspark.sql.functions import desc\n",
    "from pyspark.ml.evaluation import RegressionEvaluator\n",
    "from pyspark.ml.recommendation import ALS\n",
    "import numpy as np"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Initialize a spark session.\n",
    "def init_spark():\n",
    "    spark = SparkSession \\\n",
    "        .builder \\\n",
    "        .appName(\"Python Spark SQL basic example\") \\\n",
    "        .config(\"spark.some.config.option\", \"some-value\") \\\n",
    "        .getOrCreate()\n",
    "    return spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Useful functions to print RDDs and Dataframes.\n",
    "def toCSVLineRDD(rdd):\n",
    "    '''\n",
    "    This function convert an RDD or a DataFrame into a CSV string\n",
    "    '''\n",
    "    a = rdd.map(lambda row: \",\".join([str(elt) for elt in row]))\\\n",
    "           .reduce(lambda x,y: os.linesep.join([x,y]))\n",
    "    return a + os.linesep\n",
    "\n",
    "def toCSVLine(data):\n",
    "    '''\n",
    "    Convert an RDD or a DataFrame into a CSV string\n",
    "    '''\n",
    "    if isinstance(data, RDD):\n",
    "        return toCSVLineRDD(data)\n",
    "    elif isinstance(data, DataFrame):\n",
    "        return toCSVLineRDD(data.rdd)\n",
    "    return None"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {},
   "outputs": [],
   "source": [
    "def basic_als_recommender(filename, seed):\n",
    "    '''\n",
    "    This function must print the RMSE of recommendations obtained\n",
    "    through ALS collaborative filtering, similarly to the example at\n",
    "    http://spark.apache.org/docs/latest/ml-collaborative-filtering.html\n",
    "    The training ratio must be 80% and the test ratio must be 20%. The\n",
    "    random seed used to sample the training and test sets (passed to\n",
    "    ''DataFrame.randomSplit') is an argument of the function. The seed\n",
    "    must also be used to initialize the ALS optimizer (use\n",
    "    *ALS.setSeed()*). The following parameters must be used in the ALS\n",
    "    optimizer:\n",
    "    - maxIter: 5\n",
    "    - rank: 70\n",
    "    - regParam: 0.01\n",
    "    - coldStartStrategy: 'drop'\n",
    "    Test file: tests/test_basic_als.py\n",
    "    '''\n",
    "    spark=init_spark()\n",
    "    lines = spark.read.text(filename).rdd\n",
    "    parts = lines.map(lambda row: row.value.split(\"::\"))\n",
    "    \n",
    "    ratingsRDD = parts.map(lambda p: Row(userId=int(p[0]), movieId=int(p[1]), rating=float(p[2])))\n",
    "    ratings = spark.createDataFrame(ratingsRDD)\n",
    "    \n",
    "    (training, test) = ratings.randomSplit([0.8, 0.2],seed=seed)\n",
    "    als = ALS(maxIter=5, rank=70, regParam=0.01, userCol=\"userId\", itemCol=\"movieId\", ratingCol=\"rating\",\n",
    "              coldStartStrategy=\"drop\", seed=seed)\n",
    "    \n",
    "    predictions = als.fit(training).transform(test)\n",
    "    evaluator = RegressionEvaluator(metricName=\"rmse\", labelCol=\"rating\", predictionCol=\"prediction\")\n",
    "    rmse = evaluator.evaluate(predictions)\n",
    "\n",
    "    return rmse"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1.6229741581612676"
      ]
     },
     "execution_count": 41,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "a = basic_als_recommender(\"/Users/manikhossain/IdeaProjects/bigdata-la2-ManikHossain08/data/sample_movielens_ratings.txt\", 123)\n",
    "a"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 83,
   "metadata": {},
   "outputs": [],
   "source": [
    "filename = \"/Users/manikhossain/IdeaProjects/bigdata-la2-ManikHossain08/data/sample_movielens_ratings.txt\"\n",
    "spark=init_spark()\n",
    "movie_df = spark.read.csv(\"/Users/manikhossain/IdeaProjects/bigdata-la2-ManikHossain08/data/movies.csv\", header = True)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {},
   "outputs": [],
   "source": [
    "def global_average(filename, seed):\n",
    "    '''\n",
    "    This function must print the global average rating for all users and\n",
    "    all movies in the training set. Training and test\n",
    "    sets should be determined as before (e.g: as in function basic_als_recommender).\n",
    "    Test file: tests/test_global_average.py\n",
    "    '''\n",
    "    spark = init_spark()\n",
    "    lines = spark.read.text(filename).rdd\n",
    "    parts = lines.map(lambda row: row.value.split(\"::\"))\n",
    "    ratingsRDD = parts.map(lambda p: Row(userId=int(p[0]), movieId=int(p[1]),\n",
    "                                         rating=float(p[2])))\n",
    "    ratings = spark.createDataFrame(ratingsRDD)\n",
    "    (training, test) = ratings.randomSplit([0.8, 0.2], seed=seed)\n",
    "\n",
    "    result = training.agg({\"rating\": \"avg\"}).collect()[0]\n",
    "    return result[\"avg(rating)\"]\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1.7774979009235936"
      ]
     },
     "execution_count": 43,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "result = global_average(filename, 123)\n",
    "result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {},
   "outputs": [],
   "source": [
    "def global_average_recommender(filename, seed):\n",
    "    '''\n",
    "    This function must print the RMSE of recommendations obtained\n",
    "    through global average, that is, the predicted rating for each\n",
    "    user-movie pair must be the global average computed in the previous\n",
    "    task. Training and test\n",
    "    sets should be determined as before. You can add a column to an existing DataFrame with function *.withColumn(...)*.\n",
    "    Test file: tests/test_global_average_recommender.py\n",
    "    '''\n",
    "    spark = init_spark()\n",
    "    lines = spark.read.text(filename).rdd\n",
    "    parts = lines.map(lambda row: row.value.split(\"::\"))\n",
    "    ratingsRDD = parts.map(lambda p: Row(userId=int(p[0]), movieId=int(p[1]),\n",
    "                                         rating=float(p[2])))\n",
    "    ratings = spark.createDataFrame(ratingsRDD)\n",
    "    (training, test) = ratings.randomSplit([0.8, 0.2], seed=seed)\n",
    "    average = training.agg({\"rating\": \"avg\"}).collect()[0]['avg(rating)']\n",
    "    training = training.withColumn('avg(rating)', lit(average))\n",
    "\n",
    "    als = ALS(maxIter=5, rank=70, regParam=0.01, userCol=\"userId\", itemCol=\"movieId\", ratingCol=\"avg(rating)\",\n",
    "              coldStartStrategy=\"drop\", seed=seed)\n",
    "    \n",
    "    predictions = als.fit(training).transform(test)\n",
    "\n",
    "    evaluator = RegressionEvaluator(metricName=\"rmse\", labelCol=\"rating\",\n",
    "                                    predictionCol=\"prediction\")\n",
    "    rmse = evaluator.evaluate(predictions)\n",
    "\n",
    "    return rmse"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1.1684684492180035"
      ]
     },
     "execution_count": 67,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "result = global_average_recommender(filename, 123)\n",
    "result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 114,
   "metadata": {},
   "outputs": [],
   "source": [
    "def top_movie_recommendation_touser(filename, seed):\n",
    "    '''\n",
    "    This function must print the RMSE of recommendations obtained\n",
    "    through global average, that is, the predicted rating for each\n",
    "    user-movie pair must be the global average computed in the previous\n",
    "    task. Training and test\n",
    "    sets should be determined as before. You can add a column to an existing DataFrame with function *.withColumn(...)*.\n",
    "    Test file: tests/test_global_average_recommender.py\n",
    "    '''\n",
    "    spark = init_spark()\n",
    "    lines = spark.read.text(filename).rdd\n",
    "    parts = lines.map(lambda row: row.value.split(\"::\"))\n",
    "    ratingsRDD = parts.map(lambda p: Row(userId=int(p[0]), movieId=int(p[1]),\n",
    "                                         rating=float(p[2])))\n",
    "    ratings = spark.createDataFrame(ratingsRDD)\n",
    "    (training, test) = ratings.randomSplit([0.8, 0.2], seed=seed)\n",
    "    average = training.agg({\"rating\": \"avg\"}).collect()[0]['avg(rating)']\n",
    "    training = training.withColumn('avg(rating)', lit(average))\n",
    "\n",
    "    als = ALS(maxIter=5, rank=70, regParam=0.01, userCol=\"userId\", itemCol=\"movieId\", ratingCol=\"rating\",\n",
    "              coldStartStrategy=\"drop\", seed=seed)\n",
    "    model = als.fit(training)\n",
    "    \n",
    "    # Generate top 10 movie recommendations for each user\n",
    "    userRecs = model.recommendForAllUsers(10)\n",
    "    # Generate top 10 user recommendations for each movie\n",
    "    movieRecs = model.recommendForAllItems(10)\n",
    "\n",
    "    return (userRecs, movieRecs)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 115,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+--------------------+\n",
      "|userId|     recommendations|\n",
      "+------+--------------------+\n",
      "|    28|[[92, 4.9826636],...|\n",
      "|    26|[[88, 4.999562], ...|\n",
      "|    27|[[18, 3.9991288],...|\n",
      "|    12|[[17, 4.9900746],...|\n",
      "|    22|[[74, 5.04252], [...|\n",
      "|     1|[[68, 3.9480424],...|\n",
      "|    13|[[93, 3.931377], ...|\n",
      "|     6|[[25, 4.982175], ...|\n",
      "|    16|[[90, 4.9941545],...|\n",
      "|     3|[[18, 3.9945285],...|\n",
      "+------+--------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "(userRecs, movieRecs)  = top_movie_recommendation_touser(filename, 123)\n",
    "userRecs.show(n = 10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 112,
   "metadata": {},
   "outputs": [],
   "source": [
    "def movie_recommendation_toOneuser(filename, seed, userId):\n",
    "    '''\n",
    "    This function must print the RMSE of recommendations obtained\n",
    "    through global average, that is, the predicted rating for each\n",
    "    user-movie pair must be the global average computed in the previous\n",
    "    task. Training and test\n",
    "    sets should be determined as before. You can add a column to an existing DataFrame with function *.withColumn(...)*.\n",
    "    Test file: tests/test_global_average_recommender.py\n",
    "    '''\n",
    "    spark = init_spark()\n",
    "    lines = spark.read.text(filename).rdd\n",
    "    parts = lines.map(lambda row: row.value.split(\"::\"))\n",
    "    ratingsRDD = parts.map(lambda p: Row(userId=int(p[0]), movieId=int(p[1]),\n",
    "                                         rating=float(p[2])))\n",
    "    ratings = spark.createDataFrame(ratingsRDD)\n",
    "    (training, test) = ratings.randomSplit([0.8, 0.2], seed=seed)\n",
    "    average = training.agg({\"rating\": \"avg\"}).collect()[0]['avg(rating)']\n",
    "    training = training.withColumn('avg(rating)', lit(average))\n",
    "\n",
    "    als = ALS(maxIter=5, rank=70, regParam=0.01, userCol=\"userId\", itemCol=\"movieId\", ratingCol=\"rating\",\n",
    "              coldStartStrategy=\"drop\", seed=seed)\n",
    "    model = als.fit(training)\n",
    "    predictions = model.transform(test)\n",
    "\n",
    "    oneUser = predictions.filter(col(\"userId\") == userId).join(movie_df, \"movieId\") \\\n",
    "              .select(\"userId\",\"movieId\", \"title\", \"genres\", \"prediction\")\n",
    "    \n",
    "    return oneUser"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 113,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+--------------------+--------------------+----------+\n",
      "|userId|movieId|               title|              genres|prediction|\n",
      "+------+-------+--------------------+--------------------+----------+\n",
      "|    15|     85|Angels and Insect...|       Drama|Romance| 1.7286502|\n",
      "|    15|     65|     Bio-Dome (1996)|              Comedy| 0.5836302|\n",
      "|    15|     22|      Copycat (1995)|Crime|Drama|Horro...|0.46404094|\n",
      "|    15|      1|    Toy Story (1995)|Adventure|Animati...|0.40661034|\n",
      "|    15|     52|Mighty Aphrodite ...|Comedy|Drama|Romance|0.40930358|\n",
      "|    15|     64|Two if by Sea (1996)|      Comedy|Romance|  1.077209|\n",
      "|    15|     35|   Carrington (1995)|       Drama|Romance| 0.5394808|\n",
      "|    15|     39|     Clueless (1995)|      Comedy|Romance| 1.0550858|\n",
      "|    15|     87|Dunston Checks In...|     Children|Comedy| 1.0461016|\n",
      "|    15|     51|Guardian Angel (1...|Action|Drama|Thri...|0.30425352|\n",
      "|    15|     97|Hate (Haine, La) ...|         Crime|Drama| 0.7158899|\n",
      "|    15|     73|Misérables, Les (...|           Drama|War|0.41770607|\n",
      "|    15|      2|      Jumanji (1995)|Adventure|Childre...| 1.5322835|\n",
      "|    15|     36|Dead Man Walking ...|         Crime|Drama|0.29546544|\n",
      "+------+-------+--------------------+--------------------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "singleUserPredictions = movie_recommendation_toOneuser(filename, 123, 15)\n",
    "singleUserPredictions.show() #userId = 15"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 126,
   "metadata": {},
   "outputs": [],
   "source": [
    " # Generate top 10 movie recommendations for a specified set of users\n",
    "def top_movie_recommendation_toSetOfuser(filename, seed):\n",
    "    '''\n",
    "    This function must print the RMSE of recommendations obtained\n",
    "    through global average, that is, the predicted rating for each\n",
    "    user-movie pair must be the global average computed in the previous\n",
    "    task. Training and test\n",
    "    sets should be determined as before. You can add a column to an existing DataFrame with function *.withColumn(...)*.\n",
    "    Test file: tests/test_global_average_recommender.py\n",
    "    '''\n",
    "    spark = init_spark()\n",
    "    lines = spark.read.text(filename).rdd\n",
    "    parts = lines.map(lambda row: row.value.split(\"::\"))\n",
    "    ratingsRDD = parts.map(lambda p: Row(userId=int(p[0]), movieId=int(p[1]),\n",
    "                                         rating=float(p[2])))\n",
    "    ratings = spark.createDataFrame(ratingsRDD)\n",
    "    (training, test) = ratings.randomSplit([0.8, 0.2], seed=seed)\n",
    "    average = training.agg({\"rating\": \"avg\"}).collect()[0]['avg(rating)']\n",
    "    training = training.withColumn('avg(rating)', lit(average))\n",
    "\n",
    "    als = ALS(maxIter=5, rank=70, regParam=0.01, userCol=\"userId\", itemCol=\"movieId\", ratingCol=\"rating\",\n",
    "              coldStartStrategy=\"drop\", seed=seed)\n",
    "    model = als.fit(training)\n",
    "    predictions = model.transform(test)\n",
    "\n",
    "    # Generate top 10 movie recommendations for a specified set of users (here 5 users)\n",
    "    users = ratings.select(als.getUserCol()).distinct().limit(5)\n",
    "    userSubsetRecs = model.recommendForUserSubset(users, 10) # top 10 movie recommend from model\n",
    "    \n",
    "    return userSubsetRecs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 137,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- userId: integer (nullable = false)\n",
      " |-- recommendations: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- movieId: integer (nullable = true)\n",
      " |    |    |-- rating: float (nullable = true)\n",
      "\n",
      "+------+----------------------------------------+\n",
      "|userId|movieId                                 |\n",
      "+------+----------------------------------------+\n",
      "|26    |[88, 94, 22, 23, 68, 54, 36, 81, 57, 6] |\n",
      "|22    |[74, 88, 22, 30, 51, 32, 69, 68, 62, 98]|\n",
      "|19    |[90, 98, 74, 54, 22, 88, 94, 85, 68, 87]|\n",
      "|29    |[46, 90, 32, 94, 22, 68, 17, 49, 10, 38]|\n",
      "|0     |[92, 9, 91, 2, 26, 25, 41, 95, 89, 77]  |\n",
      "+------+----------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "topMovie = top_movie_recommendation_toSetOfuser(filename, 123)\n",
    "topMovie.printSchema()\n",
    "topMovie.select(\"userId\", \"recommendations.movieId\").show(10,False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 134,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Generate top 10 user recommendations for a specified set of movies\n",
    "def top_user_recommendation_setOfItems(filename, seed):\n",
    "    '''\n",
    "    This function must print the RMSE of recommendations obtained\n",
    "    through global average, that is, the predicted rating for each\n",
    "    user-movie pair must be the global average computed in the previous\n",
    "    task. Training and test\n",
    "    sets should be determined as before. You can add a column to an existing DataFrame with function *.withColumn(...)*.\n",
    "    Test file: tests/test_global_average_recommender.py\n",
    "    '''\n",
    "    spark = init_spark()\n",
    "    lines = spark.read.text(filename).rdd\n",
    "    parts = lines.map(lambda row: row.value.split(\"::\"))\n",
    "    ratingsRDD = parts.map(lambda p: Row(userId=int(p[0]), movieId=int(p[1]),\n",
    "                                         rating=float(p[2])))\n",
    "    ratings = spark.createDataFrame(ratingsRDD)\n",
    "    (training, test) = ratings.randomSplit([0.8, 0.2], seed=seed)\n",
    "    average = training.agg({\"rating\": \"avg\"}).collect()[0]['avg(rating)']\n",
    "    training = training.withColumn('avg(rating)', lit(average))\n",
    "\n",
    "    als = ALS(maxIter=5, rank=70, regParam=0.01, userCol=\"userId\", itemCol=\"movieId\", ratingCol=\"rating\",\n",
    "              coldStartStrategy=\"drop\", seed=seed)\n",
    "    model = als.fit(training)\n",
    "    predictions = model.transform(test)\n",
    "\n",
    "    # Generate top 10 user recommendations for a specified set of movies(here 5 items)\n",
    "    items = ratings.select(als.getItemCol()).distinct().limit(5)\n",
    "    itemSubsetRecs = model.recommendForItemSubset(items, 10)\n",
    "    return itemSubsetRecs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 138,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- movieId: integer (nullable = false)\n",
      " |-- recommendations: array (nullable = true)\n",
      " |    |-- element: struct (containsNull = true)\n",
      " |    |    |-- userId: integer (nullable = true)\n",
      " |    |    |-- rating: float (nullable = true)\n",
      "\n",
      "+-------+--------------------------------------+\n",
      "|movieId|userId                                |\n",
      "+-------+--------------------------------------+\n",
      "|65     |[23, 5, 3, 1, 29, 24, 28, 22, 10, 12] |\n",
      "|26     |[15, 0, 25, 28, 10, 29, 19, 7, 13, 20]|\n",
      "|54     |[16, 26, 19, 22, 15, 3, 29, 1, 7, 12] |\n",
      "|19     |[11, 2, 29, 15, 18, 23, 13, 25, 5, 3] |\n",
      "|29     |[14, 21, 4, 7, 22, 13, 10, 20, 8, 1]  |\n",
      "+-------+--------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "topUser = top_user_recommendation_setOfItems(filename, 123)\n",
    "topUser.printSchema()\n",
    "topUser.select(\"movieId\", \"recommendations.userId\").show(10, False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 178,
   "metadata": {},
   "outputs": [],
   "source": [
    "# predictions some movie for specific users\n",
    "def specific_user_movie_recommendations(filename, seed):\n",
    "    '''\n",
    "    This function must print the RMSE of recommendations obtained\n",
    "    through global average, that is, the predicted rating for each\n",
    "    user-movie pair must be the global average computed in the previous\n",
    "    task. Training and test\n",
    "    sets should be determined as before. You can add a column to an existing DataFrame with function *.withColumn(...)*.\n",
    "    Test file: tests/test_global_average_recommender.py\n",
    "    '''\n",
    "    spark = init_spark()\n",
    "    lines = spark.read.text(filename).rdd\n",
    "    parts = lines.map(lambda row: row.value.split(\"::\"))\n",
    "    ratingsRDD = parts.map(lambda p: Row(userId=int(p[0]), movieId=int(p[1]),\n",
    "                                         rating=float(p[2])))\n",
    "    ratings = spark.createDataFrame(ratingsRDD)\n",
    "    (training, test) = ratings.randomSplit([0.8, 0.2], seed=seed)\n",
    "    average = training.agg({\"rating\": \"avg\"}).collect()[0]['avg(rating)']\n",
    "    training = training.withColumn('avg(rating)', lit(average))\n",
    "\n",
    "    als = ALS(maxIter=5, rank=70, regParam=0.01, userCol=\"userId\", itemCol=\"movieId\", ratingCol=\"rating\",\n",
    "              coldStartStrategy=\"drop\", seed=seed)\n",
    "    model = als.fit(training)\n",
    "    predictions = model.transform(test) # predict for test data\n",
    "    \n",
    "    # predictions some movie for specific users\n",
    "    movies = ratings.select(als.getItemCol()).distinct().limit(5)\n",
    "    users = ratings.select(als.getUserCol()).distinct().limit(5)\n",
    "    movie_ids = []\n",
    "    user_ids = []\n",
    "    \n",
    "    for i in range(0, movies.count()):\n",
    "            movie_ids.append(movies.collect()[i][0])\n",
    "            user_ids.append(users.collect()[i][0])\n",
    "\n",
    "    newUserPred = spark.createDataFrame(zip(movie_ids, user_ids), schema = ['movieId', 'userId'])\n",
    "    \n",
    "    newPredictions = predictions = model.transform(newUserPred)\n",
    "    \n",
    "    return newPredictions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 179,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- movieId: long (nullable = true)\n",
      " |-- userId: long (nullable = true)\n",
      " |-- prediction: float (nullable = false)\n",
      "\n",
      "+-------+------+----------+\n",
      "|movieId|userId|prediction|\n",
      "+-------+------+----------+\n",
      "|     65|    19|0.97528976|\n",
      "|     26|    26|0.25505516|\n",
      "|     54|    22| 1.9031708|\n",
      "|     19|     0| 1.0041881|\n",
      "|     29|    29| 1.0008368|\n",
      "+-------+------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "topUserItem = specific_user_movie_recommendations(filename, 123)\n",
    "topUserItem.printSchema()\n",
    "topUserItem.show()"
   ]
  },
  {
   "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.7.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}