{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Assignment on Spark (Scala)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Gosia Migut and Georgios Gousios"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this assignment, we will use Spark to have a look at the [Movie Lens dataset](https://drive.google.com/file/d/1rmvXfdnM-JBsDLGm-072z0ZmtrvzC0Lw) containing user generated ratings for movies.  The dataset comes in 3 files:\n",
    "\n",
    "* `ratings.dat` contains the ratings in the following format: `UserID::MovieID::Rating::Timestamp`\n",
    "* `users.dat` contains demographic information about the users: `UserID::Gender::Age::Occupation::Zip-code`\n",
    "* `movies.dat` contains meta information about the movies: `MovieID::Title::Genres`\n",
    "\n",
    "Refer to the README for the detailed description of the data.\n",
    "\n",
    "**Note:** when using the files use the filepath `data/[file].dat`, otherwise automatic grading will fail.\n",
    "\n",
    "**Grade:** This assignment consists of 105 points. You need to collect them all to get a 10! All cells that are graded include the expected answer. Your task is to write the code that comes up with the expected result. The automated grading process will be run on a different dataset."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Loading and parsing the file"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q1 (5 points):** Download the ratings file, parse it and load it in an RDD named `ratings`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "cell_id": "Q1.1"
   },
   "outputs": [],
   "source": [
    "import org.apache.spark.rdd.RDD\n",
    "\n",
    "case class Rating(user_ID: Integer, movie_ID: Integer, rating: Integer, timestamp: String)\n",
    "case class Movie(movie_ID: Integer, title: String, genre: String)\n",
    "case class User(user_ID: Integer, gender: String, age: Integer, occupation: String, zip_code: String)\n",
    "\n",
    "def parseRatings(row: String): Rating = {\n",
    "    return null\n",
    "}\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "cell_id": "Q1.2"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "MapPartitionsRDD[7] at map at <console>:25"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//load data to RDD and use parseRatings function to parse it.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q2 (5 points):** How many lines does the `ratings` RDD contain? "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "cell_id": "Q2"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1000209"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Basic filtering and counting"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q3 (5 points):** Count how many times the rating '1' has been given."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "cell_id": "Q3"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "56174"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q4 (5 points):** Count how many unique movies have been rated."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "cell_id": "Q4"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "3706"
      ]
     },
     "execution_count": 10,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q5 (5 points):** Which user gave most ratings? Return the `userID` and number of ratings. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "cell_id": "Q5"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Array((4169,2314))"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q6 (5 points):** Which user gave most '5' ratings? Return the `userID` and number of ratings. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "cell_id": "Q6"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Array((4277,571))"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q7 (5 points):** Which movie was rated most times? Return the `movieID` and number of ratings."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "cell_id": "Q7"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Array((2858,3428))"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Joining"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we will look at two additional files from the Movie Lens dataset."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q8 (5 points):** Read the `movies` and `users` files into RDDs. How many records are there in each RDD?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "cell_id": "Q8.1"
   },
   "outputs": [],
   "source": [
    "//load movies dataset to RDD, parse and cache it.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "cell_id": "Q8.2"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "3883"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//how many records are in movies RDD's?\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "cell_id": "Q8.3"
   },
   "outputs": [],
   "source": [
    "//load users dataset to RDD, parse and cache it.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "cell_id": "Q8.4"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "6040"
      ]
     },
     "execution_count": 17,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "//how many records are in users RDD's?\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As you probably have noticed there are more movies in the movies dataset than rated movies."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q9 (5 points):** How many of the movies are a comedy? "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "cell_id": "Q9"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1200"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q10 (10 points):** Which comedy has the most ratings? Return the title and the number of rankings. Answer this question by joining two datasets."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "cell_id": "Q10"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Array((American Beauty (1999),3428))"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q11 (10 points):** For users under 18 years old (category 1), what is the frequency of each star rating? Return a list/array with the rating and the number of times it appears, e.g. `Array((4,16), (1,3), (3,9), (5,62), (2,2))`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "cell_id": "Q11"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Array((4,8808), (2,2983), (1,2238), (3,6380), (5,6802))"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Indexing"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As you have noticed, typical operations on RDDs require grouping on a specific part of each record and then calculating specific counts given the groups. While this operation can be achieved with the `groupBy` family of functions, it is often useful to create a structure called an inverted index. An inverted index creates an `1..n` mapping from the record part to all occurencies of the record in the dataset. For example, if the dataset looks like the following:\n",
    "\n",
    "> col1,col2,col3   \n",
    " A,1,foo   \n",
    "B,1,bar   \n",
    "C,2,foo   \n",
    "D,3,baz   \n",
    "E,1,foobar  \n",
    "\n",
    "an inverted index on col2 would look like\n",
    "\n",
    ">1 -> [(A,1,foo), (B,1,bar), (E,1,foobar)]  \n",
    "2 -> [(C,2,foo)]   \n",
    "3 -> [(D,3,baz)]  \n",
    "\n",
    "Inverted indexes enable us to quickly access precalculated partitions of the dataset. Let's compute an inverted index on the `rating` field of `ratings.dat`."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q12 (5 points)**: Compute the number of unique users that rated the movies with `movie_ID`s 2858, 356 and 2329."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "cell_id": "Q12.1"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4213"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Measure the time (in seconds) it takes to make this computation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "cell_id": "Q12.2"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1.022\n"
     ]
    }
   ],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q13 (5 points)**: Create an inverted index on `ratings`, field `movie_ID`. Print the first item."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "cell_id": "Q13"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "ArrayBuffer(CompactBuffer(Rating(1,1,5,978824268), Rating(6,1,4,978237008), Rating(8,1,4,978233496), Rating(9,1,5,978225952), Rating(10,1,5,978226474), Rating(18,1,4,978154768), Rating(19,1,5,978555994), Rating(21,1,3,978139347), Rating(23,1,4,978463614), Rating(26,1,3,978130703), Rating(28,1,3,978985309), Rating(34,1,5,978102970), Rating(36,1,5,978061285), Rating(38,1,5,978046225), Rating(44,1,5,978019369), Rating(45,1,4,977990044), Rating(48,1,4,977975909), Rating(49,1,5,977972501), Rating(51,1,5,977947828), Rating(56,1,5,977938855), Rating(60,1,4,977931983), Rating(65,1,5,991368774), Rating(68,1,3,991376026), Rating(73,1,3,977867812), Rating(75,1,5,977851099), Rating(76,1,5,977847069), Rating(78,1,4,978570648), Rating(80,1,3,977786904), Rating(90,1,3,99..."
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q14 (5 points)**: Compute the number of unique users that rated the movies with `movie_ID`s 2858, 356 and 2329 using the index"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "cell_id": "Q14.1"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4213"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Measure the time (in seconds) it takes to compute the same result using the index."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "cell_id": "Q14.2"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "19.245\n"
     ]
    }
   ],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "You should have noticed difference in performance.  Is the indexed version faster? If yes, why? If not, why not?  Discuss this with your partner."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Dataframes"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q15 (5 points)**: Create a data frame from the `ratings` RDD and count the number of lines in it. Also register the data frame as an SQL table."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "cell_id": "Q15"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1000209"
      ]
     },
     "execution_count": 26,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q16 (5 points):** Provide the statistical summary of the column containing ratings (use Spark function that returns a table with count, mean, stddev, min, max). \n",
    "\n",
    "_Hint_: To select the correct column you might first want to print the datatypes and names of each of the columns."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "cell_id": "Q16"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------------------+\n",
      "|summary|            rating|\n",
      "+-------+------------------+\n",
      "|  count|           1000209|\n",
      "|   mean| 3.581564453029317|\n",
      "| stddev|1.1171018453732544|\n",
      "|    min|                 1|\n",
      "|    max|                 5|\n",
      "+-------+------------------+\n",
      "\n"
     ]
    }
   ],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q17 (5 points):** Count how many times the rating '1' has been given, by filtering it from the ratings DataFrame. Measure the execution time and compare with the execution time of the same query using RDD. \n",
    "Think for yourself when it would be usefull to use DataFrames and when not."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "cell_id": "Q17"
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "56174"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q18 (5 points):** Count how many times the rating '1' has been given, using an SQL query"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "cell_id": "Q18"
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|count(1)|\n",
      "+--------+\n",
      "|   56174|\n",
      "+--------+\n",
      "\n"
     ]
    }
   ],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Q19 (5 points):** Which user gave most '5' ratings? Return the `userID` and number of ratings, using an SQL query"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "cell_id": "Q19",
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----------+\n",
      "|user_ID|num_ratings|\n",
      "+-------+-----------+\n",
      "|   4277|        571|\n",
      "+-------+-----------+\n",
      "only showing top 1 row\n",
      "\n"
     ]
    }
   ],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Apache Toree - Scala",
   "language": "scala",
   "name": "apache_toree_scala"
  },
  "language_info": {
   "file_extension": ".scala",
   "name": "scala",
   "version": "2.11.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}