movie-similarities-1m.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import sys\n",
    "from pyspark import SparkConf, SparkContext\n",
    "from math import sqrt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def loadMovieNames():\n",
    "    movieNames = {}\n",
    "    with open(\"ml-1m/movies.dat\", encoding='latin-1') as f:\n",
    "        for line in f:\n",
    "            fields = line.split('::')\n",
    "            movieNames[int(fields[0])] = fields[1]\n",
    "    return movieNames"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def makePairs(x):\n",
    "    (movie1, rating1) = x[1][0]\n",
    "    (movie2, rating2) = x[1][1]\n",
    "    return ((movie1, movie2), (rating1, rating2))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def filterDuplicates(x):\n",
    "    (movie1, rating1) = x[1][0]\n",
    "    (movie2, rating2) = x[1][1]\n",
    "    return movie1 < movie2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def computeCosineSimilarity(ratingPairs):\n",
    "    numPairs = 0\n",
    "    sum_xx = sum_yy = sum_xy = 0\n",
    "    for ratingX, ratingY in ratingPairs:\n",
    "        sum_xx += ratingX * ratingX\n",
    "        sum_yy += ratingY * ratingY\n",
    "        sum_xy += ratingX * ratingY\n",
    "        numPairs += 1\n",
    "\n",
    "    numerator = sum_xy\n",
    "    denominator = sqrt(sum_xx) * sqrt(sum_yy)\n",
    "\n",
    "    score = 0\n",
    "    if (denominator):\n",
    "        score = (numerator / (float(denominator)))\n",
    "\n",
    "    return (score, numPairs)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Change to use cluster config\n",
    "conf = SparkConf().setMaster(\"local[*]\").setAppName(\"MovieSimilarities\")\n",
    "sc = SparkContext(conf = conf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n",
      "Loading movie names...\n"
     ]
    }
   ],
   "source": [
    "print(\"\\nLoading movie names...\")\n",
    "nameDict = loadMovieNames()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'Waiting to Exhale (1995)'"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "nameDict[4]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "data = sc.textFile(\"ml-1m/ratings.dat\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['9::994::4::978226328',\n",
       " '9::920::3::978225401',\n",
       " '9::912::4::978224879',\n",
       " '9::861::2::978226665',\n",
       " '9::838::3::978226495']"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "data.top(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.rdd.RDD"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Map ratings to key / value pairs: user ID => movie ID, rating\n",
    "ratings = data.map(lambda l: l.split('::')).map(lambda l: (int(l[0]), (int(l[1]), float(l[2]))))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "list"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(ratings.top(50))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Emit every movie rated together by the same user.\n",
    "# Self-join to find every combination.\n",
    "ratingsPartitioned = ratings.partitionBy(100)\n",
    "joinedRatings = ratingsPartitioned.join(ratingsPartitioned)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# At this point our RDD consists of userID => ((movieID, rating), (movieID, rating))\n",
    "joinedRatings.top(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Filter out duplicate pairs\n",
    "uniqueJoinedRatings = joinedRatings.filter(filterDuplicates)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Now key by (movie1, movie2) pairs.\n",
    "moviePairs = uniqueJoinedRatings.map(makePairs)\n",
    "# We now have (movie1, movie2) => (rating1, rating2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "type(moviePairs)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "moviePairs.top(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Now collect all ratings for each movie pair and compute similarity\n",
    "moviePairRatings = moviePairs.groupByKey()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# We now have (movie1, movie2) = > (rating1, rating2), (rating1, rating2) ...\n",
    "# Can now compute similarities.\n",
    "moviePairSimilarities = moviePairRatings.mapValues(computeCosineSimilarity).cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "moviePairSimilarities.top(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Save the results if desired\n",
    "#moviePairSimilarities.sortByKey()\n",
    "#moviePairSimilarities.saveAsTextFile(\"movie-sims\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sys.argv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "defaultId = 50\n",
    "# Extract similarities for the movie we care about that are \"good\".\n",
    "if (len(sys.argv) > 1):\n",
    "    try:\n",
    "        movieID = int(sys.argv[1])\n",
    "    except:\n",
    "        movieID = defaultId    \n",
    "else: \n",
    "    movieID = defaultId"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "scoreThreshold = 0.97\n",
    "coOccurenceThreshold = 50\n",
    "\n",
    "\n",
    "# Filter for movies with this sim that are \"good\" as defined by\n",
    "# our quality thresholds above\n",
    "filteredResults = moviePairSimilarities.filter(lambda x: \\\n",
    "        (x[0][0] == movieID or x[0][1] == movieID) \\\n",
    "        and x[1][0] > scoreThreshold and x[1][1] > coOccurenceThreshold)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Sort by quality score.\n",
    "results = filteredResults.map(lambda x: (x[1],x[0])).sortByKey(ascending = False).take(10)\n",
    "\n",
    "print(\"Top 10 similar movies for \" + nameDict[movieID])\n",
    "for result in results:\n",
    "        (sim, pair) = result\n",
    "        # Display the similarity result that isn't the movie we're looking at\n",
    "        similarMovieID = pair[0]\n",
    "        if (similarMovieID == movieID):\n",
    "            similarMovieID = pair[1]\n",
    "        print(nameDict[similarMovieID] + \"\\tscore: \" + str(sim[0]) + \"\\tstrength: \" + str(sim[1]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "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.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}