spark.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "%matplotlib inline\n",
    "import matplotlib.pyplot as plt\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import seaborn as sns\n",
    "from datetime import datetime\n",
    "\n",
    "import pyspark\n",
    "from pyspark import SparkContext\n",
    "from pyspark import SparkConf\n",
    "from pyspark.sql import SparkSession\n",
    "import numpy as np\n",
    "\n",
    "\n",
    "import os\n",
    "from time import time"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(100004, 4)"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#file = 'ratings.csv'\n",
    "file = ('ratings_small.csv')\n",
    "ratings_df = pd.read_csv(file)\n",
    "\n",
    "ratings_df.shape"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc = SparkContext.getOrCreate(SparkConf().setMaster(\"local[*]\"))\n",
    "small_ratings_raw_data = sc.textFile('ratings_small.csv')\n",
    "small_ratings_raw_data_header = small_ratings_raw_data.take(1)[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "small_ratings_data = small_ratings_raw_data.filter(lambda line: line!=small_ratings_raw_data_header)\\\n",
    "    .map(lambda line: line.split(\",\")).map(lambda tokens: (tokens[0],tokens[1],tokens[2])).cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[('1', '31', '2.5'), ('1', '1029', '3.0'), ('1', '1061', '3.0')]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "small_ratings_data.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "training_RDD, validation_RDD, test_RDD = small_ratings_data.randomSplit([6, 2, 2], seed=0)\n",
    "validation_for_predict_RDD = validation_RDD.map(lambda x: (x[0], x[1]))\n",
    "test_for_predict_RDD = test_RDD.map(lambda x: (x[0], x[1]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.mllib.recommendation import ALS\n",
    "import math"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "seed = 5\n",
    "iterations = [5,10, 15, 25]\n",
    "regularization_parameter = [0.05, 0.1, 0.2]\n",
    "ranks = [5, 10, 15]\n",
    "errors = []\n",
    "err = 0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "min_error = float('inf')\n",
    "best_rank = -1\n",
    "best_iteration = -1\n",
    "best_reg_par=-1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "5 5 0.05\n",
      "For rank 5, iterations 5 the RMSE is 0.9909944177563392 in 6.752425670623779 sec\n",
      "5 5 0.1\n",
      "For rank 5, iterations 5 the RMSE is 0.9431141589594201 in 4.5080602169036865 sec\n",
      "5 5 0.2\n",
      "For rank 5, iterations 5 the RMSE is 0.9225513302402039 in 4.220797061920166 sec\n",
      "5 10 0.05\n",
      "For rank 5, iterations 10 the RMSE is 0.9933565693338079 in 3.916440963745117 sec\n",
      "5 10 0.1\n",
      "For rank 5, iterations 10 the RMSE is 0.9366908878153927 in 3.585076093673706 sec\n",
      "5 10 0.2\n",
      "For rank 5, iterations 10 the RMSE is 0.9167557864526742 in 3.908825159072876 sec\n",
      "5 15 0.05\n",
      "For rank 5, iterations 15 the RMSE is 0.9969180104204706 in 3.708817720413208 sec\n",
      "5 15 0.1\n",
      "For rank 5, iterations 15 the RMSE is 0.9387085472995429 in 3.952730894088745 sec\n",
      "5 15 0.2\n",
      "For rank 5, iterations 15 the RMSE is 0.9169477037160668 in 3.632023334503174 sec\n",
      "5 25 0.05\n",
      "For rank 5, iterations 25 the RMSE is 0.9989888606647397 in 4.134925842285156 sec\n",
      "5 25 0.1\n",
      "For rank 5, iterations 25 the RMSE is 0.9396901243067682 in 4.328730821609497 sec\n",
      "5 25 0.2\n",
      "For rank 5, iterations 25 the RMSE is 0.9170309828365204 in 3.5424835681915283 sec\n",
      "10 5 0.05\n",
      "For rank 10, iterations 5 the RMSE is 1.024608954533246 in 2.627504587173462 sec\n",
      "10 5 0.1\n",
      "For rank 10, iterations 5 the RMSE is 0.9460237381924925 in 2.608691453933716 sec\n",
      "10 5 0.2\n",
      "For rank 10, iterations 5 the RMSE is 0.9194012576796401 in 2.924351453781128 sec\n",
      "10 10 0.05\n",
      "For rank 10, iterations 10 the RMSE is 1.031064371583573 in 3.12280011177063 sec\n",
      "10 10 0.1\n",
      "For rank 10, iterations 10 the RMSE is 0.9446204572753562 in 2.906529188156128 sec\n",
      "10 10 0.2\n",
      "For rank 10, iterations 10 the RMSE is 0.9175817721503496 in 3.107532501220703 sec\n",
      "10 15 0.05\n",
      "For rank 10, iterations 15 the RMSE is 1.029064215378703 in 3.319972276687622 sec\n",
      "10 15 0.1\n",
      "For rank 10, iterations 15 the RMSE is 0.9432573228152389 in 3.480044364929199 sec\n",
      "10 15 0.2\n",
      "For rank 10, iterations 15 the RMSE is 0.9176305041885852 in 3.76353120803833 sec\n",
      "10 25 0.05\n",
      "For rank 10, iterations 25 the RMSE is 1.023601981007656 in 4.009577751159668 sec\n",
      "10 25 0.1\n",
      "For rank 10, iterations 25 the RMSE is 0.9427774847305589 in 3.7746527194976807 sec\n",
      "10 25 0.2\n",
      "For rank 10, iterations 25 the RMSE is 0.9176774288637861 in 3.8592143058776855 sec\n",
      "15 5 0.05\n",
      "For rank 15, iterations 5 the RMSE is 1.0447492350945697 in 2.862536907196045 sec\n",
      "15 5 0.1\n",
      "For rank 15, iterations 5 the RMSE is 0.9457652667921567 in 2.6849756240844727 sec\n",
      "15 5 0.2\n",
      "For rank 15, iterations 5 the RMSE is 0.9171150330038783 in 2.8088631629943848 sec\n",
      "15 10 0.05\n",
      "For rank 15, iterations 10 the RMSE is 1.0359148115238757 in 2.9765868186950684 sec\n",
      "15 10 0.1\n",
      "For rank 15, iterations 10 the RMSE is 0.9402815385889816 in 2.942756175994873 sec\n",
      "15 10 0.2\n",
      "For rank 15, iterations 10 the RMSE is 0.9160992581265431 in 3.735441207885742 sec\n",
      "15 15 0.05\n",
      "For rank 15, iterations 15 the RMSE is 1.028366533943588 in 3.9408531188964844 sec\n",
      "15 15 0.1\n",
      "For rank 15, iterations 15 the RMSE is 0.9393781664524706 in 4.026504278182983 sec\n",
      "15 15 0.2\n",
      "For rank 15, iterations 15 the RMSE is 0.9168005597049109 in 3.678649663925171 sec\n",
      "15 25 0.05\n",
      "For rank 15, iterations 25 the RMSE is 1.0190828563050702 in 4.036576271057129 sec\n",
      "15 25 0.1\n",
      "For rank 15, iterations 25 the RMSE is 0.9405711633874365 in 4.6709678173065186 sec\n",
      "15 25 0.2\n",
      "For rank 15, iterations 25 the RMSE is 0.9172893828713985 in 4.224663019180298 sec\n"
     ]
    }
   ],
   "source": [
    "for rank in ranks:\n",
    "    for itert in iterations:\n",
    "        for reg_par in regularization_parameter:\n",
    "            t0=time()\n",
    "            print(rank, itert, reg_par)\n",
    "            model = ALS.train(training_RDD, rank, seed=seed, iterations=itert, lambda_=reg_par)\n",
    "            predictions = model.predictAll(validation_for_predict_RDD).map(lambda r: ((r[0], r[1]), r[2]))\n",
    "            rates_and_preds = validation_RDD.map(lambda r: ((int(r[0]), int(r[1])), float(r[2]))).join(predictions)\n",
    "            error = math.sqrt(rates_and_preds.map(lambda r: (r[1][0] - r[1][1])**2).mean())\n",
    "            t1=time()-t0\n",
    "            errors.append(tuple((rank, itert, reg_par, error)))\n",
    "            err += 1\n",
    "            print ('For rank %s, iterations %s the RMSE is %s in %s sec' % (rank, itert, error, t1))\n",
    "            if error < min_error:\n",
    "                min_error = error\n",
    "                best_rank = rank\n",
    "                best_iteration = itert\n",
    "                best_reg_par=reg_par\n",
    "            "
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "For rank 4, iterations 5 the RMSE is 0.9904031832699854\n",
    "For rank 8, iterations 5 the RMSE is 1.019060324625603\n",
    "For rank 12, iterations 5 the RMSE is 1.0420758059007416\n",
    "The best model was trained with rank 4 interations 5 \n",
    "For rank 4, iterations 10 the RMSE is 0.994537483407379\n",
    "For rank 8, iterations 10 the RMSE is 1.0242180504741996\n",
    "For rank 12, iterations 10 the RMSE is 1.0390124251638786\n",
    "The best model was trained with rank 4 interations 10 \n",
    "For rank 4, iterations 20 the RMSE is 0.988934826644699\n",
    "For rank 8, iterations 20 the RMSE is 1.0177816705795284\n",
    "For rank 12, iterations 20 the RMSE is 1.0229475371529837"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The best model was trained with rank 15 interations 10, reg param 0.2 \n"
     ]
    }
   ],
   "source": [
    "print ('The best model was trained with rank %s interations %s, reg param %s ' % (best_rank, best_iteration, best_reg_par))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[((390, 667), 3.587789609945542),\n",
       " ((48, 44828), 0.4064959666892259),\n",
       " ((428, 5618), 4.091352887329464)]"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "predictions.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[((1, 1061), (3.0, 2.638034823999758)),\n",
       " ((1, 1129), (2.0, 2.515218704015662)),\n",
       " ((1, 1371), (2.5, 2.1242737804306535))]"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "rates_and_preds.take(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "For testing data the RMSE is 0.9226471804373351\n"
     ]
    }
   ],
   "source": [
    "model = ALS.train(training_RDD, best_rank, seed=seed, iterations=best_iteration, lambda_=best_reg_par)\n",
    "predictions = model.predictAll(test_for_predict_RDD).map(lambda r: ((r[0], r[1]), r[2]))\n",
    "rates_and_preds = test_RDD.map(lambda r: ((int(r[0]), int(r[1])), float(r[2]))).join(predictions)\n",
    "error = math.sqrt(rates_and_preds.map(lambda r: (r[1][0] - r[1][1])**2).mean())\n",
    "    \n",
    "print ('For testing data the RMSE is %s' % (error))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "complete_ratings_raw_data = sc.textFile('ratings.csv')\n",
    "complete_ratings_raw_data_header  = complete_ratings_raw_data.take(1)[0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "There are 26024289 recommendations in the complete dataset\n"
     ]
    }
   ],
   "source": [
    "complete_ratings_data = complete_ratings_raw_data.filter(lambda line: line!=complete_ratings_raw_data_header)\\\n",
    "    .map(lambda line: line.split(\",\")).map(lambda tokens: (int(tokens[0]),int(tokens[1]),float(tokens[2]))).cache()\n",
    "    \n",
    "print (\"There are %s recommendations in the complete dataset\" % (complete_ratings_data.count()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "training_RDD, test_RDD = complete_ratings_data.randomSplit([7, 3], seed=0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "t0 = time()\n",
    "complete_model = ALS.train(training_RDD, best_rank, seed=seed, \n",
    "                           iterations=best_iteration, lambda_=best_reg_par)\n",
    "tt = time() - t0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Time taken to train the full data for 15 rank, 10 iterations 213.7662742137909\n"
     ]
    }
   ],
   "source": [
    "print (\"Time taken to train the full data for %s rank, %s iterations %s\" % (best_rank, best_iteration, tt))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "test_for_predict_RDD = test_RDD.map(lambda x: (x[0], x[1]))"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "# predictions.take(5)\n",
    "[((256762, 1), 3.874252825623201, 4.0),\n",
    " ((256762, 419), 2.784599184781533, 3.0),\n",
    " ((256762, 364), 3.3881877011285795, 3.5),\n",
    " ((256762, 552), 3.1986723088649924, 3.0),\n",
    " ((256762, 300), 3.314560192444367, 3.5)]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "t0=time()\n",
    "predictions = complete_model.predictAll(test_for_predict_RDD).map(lambda r: ((r[0], r[1]), r[2] ))\n",
    "rates_and_preds = test_RDD.map(lambda r: ((int(r[0]), int(r[1])), float(r[2]))).join(predictions)\n",
    "error = math.sqrt(rates_and_preds.map(lambda r: (r[1][0] - r[1][1])**2).mean())\n",
    "#print ('For testing data the RMSE is %s' % (error))\n",
    "t1=time()-t0"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "For testing data the RMSE is 0.8649513155281516 in 498.9620997905731 secs\n"
     ]
    }
   ],
   "source": [
    "print ('For testing data the RMSE is %s in %s secs' % (error, t1))"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "rounded_error = math.sqrt(rates_and_preds.map(lambda r: (r[1][0] - r[1][2])**2).mean())\n",
    "    \n",
    "print ('For testing data the Rounded RMSE is %s' % (error))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "predictions_rounded = complete_model.predictAll(test_for_predict_RDD).map(lambda r: ((r[0], r[1]), (r[2]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "predictions_rounded.take(5)"
   ]
  },
  {
   "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.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}