movie-similarities.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import sys\n",
    "from pyspark import SparkConf, SparkContext\n",
    "from math import sqrt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def loadMovieNames():\n",
    "    movieNames = {}\n",
    "    with open(\"ml-100k/u.ITEM\", 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": 27,
   "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": 25,
   "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": 32,
   "metadata": {},
   "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": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "conf = SparkConf().setMaster(\"local[*]\").setAppName(\"MovieSimilarities\")\n",
    "sc = SparkContext(conf = conf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "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": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'Get Shorty (1995)'"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "nameDict[4]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "data = sc.textFile(\"ml-100k/u.data\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['99\\t98\\t5\\t885679596',\n",
       " '99\\t978\\t3\\t885679382',\n",
       " '99\\t975\\t3\\t885679472',\n",
       " '99\\t963\\t3\\t885679998',\n",
       " '99\\t931\\t2\\t886780147']"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "data.top(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.rdd.RDD"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "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": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(943, (1330, 3.0)),\n",
       " (943, (1228, 3.0)),\n",
       " (943, (1188, 3.0)),\n",
       " (943, (1074, 4.0)),\n",
       " (943, (1067, 2.0))]"
      ]
     },
     "execution_count": 22,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "ratings.top(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Emit every movie rated together by the same user.\n",
    "# Self-join to find every combination.\n",
    "joinedRatings = ratings.join(ratings)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(943, ((1330, 3.0), (1330, 3.0))),\n",
       " (943, ((1330, 3.0), (1228, 3.0))),\n",
       " (943, ((1330, 3.0), (1188, 3.0))),\n",
       " (943, ((1330, 3.0), (1074, 4.0))),\n",
       " (943, ((1330, 3.0), (1067, 2.0)))]"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# At this point our RDD consists of userID => ((movieID, rating), (movieID, rating))\n",
    "joinedRatings.top(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Filter out duplicate pairs\n",
    "uniqueJoinedRatings = joinedRatings.filter(filterDuplicates)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "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": 29,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.rdd.PipelinedRDD"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(moviePairs)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[((1679, 1680), (3.0, 2.0)),\n",
       " ((1678, 1680), (1.0, 2.0)),\n",
       " ((1678, 1679), (1.0, 3.0)),\n",
       " ((1675, 1676), (3.0, 2.0)),\n",
       " ((1672, 1681), (2.0, 3.0))]"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "moviePairs.top(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "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": 33,
   "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": 34,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[((1679, 1680), (1.0, 1)),\n",
       " ((1678, 1680), (1.0, 1)),\n",
       " ((1678, 1679), (1.0, 1)),\n",
       " ((1675, 1676), (1.0, 1)),\n",
       " ((1672, 1681), (1.0, 1))]"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "moviePairSimilarities.top(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Save the results if desired\n",
    "#moviePairSimilarities.sortByKey()\n",
    "#moviePairSimilarities.saveAsTextFile(\"movie-sims\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['/usr/local/Cellar/python3/3.6.4_2/Frameworks/Python.framework/Versions/3.6/lib/python3.6/site-packages/ipykernel_launcher.py',\n",
       " '-f',\n",
       " '/Users/paolo/Library/Jupyter/runtime/kernel-d55aa5e4-00ab-4be0-b3fd-0d127317e17e.json']"
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sys.argv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {},
   "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": 46,
   "metadata": {},
   "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": 47,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Top 10 similar movies for Star Wars (1977)\n",
      "Empire Strikes Back, The (1980)\tscore: 0.9895522078385338\tstrength: 345\n",
      "Return of the Jedi (1983)\tscore: 0.9857230861253026\tstrength: 480\n",
      "Raiders of the Lost Ark (1981)\tscore: 0.981760098872619\tstrength: 380\n",
      "20,000 Leagues Under the Sea (1954)\tscore: 0.9789385605497993\tstrength: 68\n",
      "12 Angry Men (1957)\tscore: 0.9776576120448436\tstrength: 109\n",
      "Close Shave, A (1995)\tscore: 0.9775948291054827\tstrength: 92\n",
      "African Queen, The (1951)\tscore: 0.9764692222674887\tstrength: 138\n",
      "Sting, The (1973)\tscore: 0.9751512937740359\tstrength: 204\n",
      "Wrong Trousers, The (1993)\tscore: 0.9748681355460885\tstrength: 103\n",
      "Wallace & Gromit: The Best of Aardman Animation (1996)\tscore: 0.9741816128302572\tstrength: 58\n"
     ]
    }
   ],
   "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
}