t3.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "eac4cb42",
   "metadata": {},
   "source": [
    "<img src=\"AUEB.png\" /> <img src=\"MSc_BA.png\" />"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cbda0a08",
   "metadata": {},
   "source": [
    "# Athens University of Economics and Business\n",
    "# School of Business\n",
    "# Department of Management Science & Technology\n",
    "# Master of Science in Business Analytics\n",
    "___"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c14c74b6",
   "metadata": {},
   "source": [
    "<table style='float:left;font-size: 20px;'>\n",
    "    <tr>\n",
    "        <th style='text-align: left;'>Program:</th>\n",
    "        <td style='text-align: left;'>Full-Time</td>\n",
    "    </tr>\n",
    "    <tr>\n",
    "        <th style='text-align: left;'>Quarter:</th>\n",
    "        <td style='text-align: left;'>2nd (Winter Quarter)</td>\n",
    "    </tr>\n",
    "    <tr>\n",
    "        <th style='text-align: left;'>Course:</th>\n",
    "        <td style='text-align: left;'>Big Data Systems & Architectures</td>\n",
    "    </tr>\n",
    "    <tr>\n",
    "        <th style='text-align: left;'>Assignment:</th>\n",
    "        <td style='text-align: left;'>Spark Assignment</td>\n",
    "    </tr> \n",
    "    <tr>\n",
    "        <th style='text-align: left;'>Student (Registration No):</th>\n",
    "        <td style='text-align: left;'>Souflas Eleftherios-Efthymios (f2822217)</td>\n",
    "    </tr>\n",
    "</table>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "def54897",
   "metadata": {},
   "source": [
    "___\n",
    "# Task 3"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2f06e0b3",
   "metadata": {},
   "source": [
    "### Instantiate a Spark Session and load the dataset"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fc9974d9",
   "metadata": {},
   "source": [
    "In order to explore the dataset, we must firstly be able to read it. \n",
    "\n",
    "PySpark is the Python API for Apache Spark. In order to install it, uncomment the following first line of the code, or if it does not work, uncomment the third and fourth line of the code.\n",
    "\n",
    "From the `pyspark.sql` module, we imported `SparkSession`, which is the entry point to programming Spark with the Dataset and DataFrame API. We then used the Builder to construct a SparkSession instance, having as a name \"task3\".\n",
    "\n",
    "Spark SQL can load a JSON file as a DataFrame using `SparkSession.read.json()` function. We then set multiline option to true to read JSON records from multiple lines of the JSON file provided into a Spark DataFrame named \"movies\"."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "6b296397",
   "metadata": {},
   "outputs": [],
   "source": [
    "# pip install pyspark\n",
    "# or if it produces error then run the following 2 lines:\n",
    "# import sys\n",
    "# !{sys.executable} -m pip install pyspark\n",
    "\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = SparkSession.builder.appName(\"task3\").getOrCreate()\n",
    "\n",
    "file = \"movie.json\" \n",
    "movies = spark.read.option(\"multiline\",\"true\").json(file)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8f45e5fb",
   "metadata": {},
   "source": [
    "### Prepare the feature vectors"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "48e3a859",
   "metadata": {},
   "source": [
    "Firstly, we casted the \"metascore\", \"users_rating\" and \"votes\" columns into integer, float and integer data types respectively.\n",
    "\n",
    "We used then the `printSchema()` function to observe that everything worked well and then used the `toPandas()` function, which as the name suggests, returns the contents of this DataFrame as Pandas `pandas.DataFrame`, in order to view it in a more pretty way than it is printed with the `show()` function."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "5becb248",
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- actors: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- countries: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- description: string (nullable = true)\n",
      " |-- directors: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- genre: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- imdb_url: string (nullable = true)\n",
      " |-- img_url: string (nullable = true)\n",
      " |-- languages: array (nullable = true)\n",
      " |    |-- element: string (containsNull = true)\n",
      " |-- metascore: integer (nullable = true)\n",
      " |-- rating: string (nullable = true)\n",
      " |-- runtime: string (nullable = true)\n",
      " |-- tagline: string (nullable = true)\n",
      " |-- title: string (nullable = true)\n",
      " |-- users_rating: float (nullable = true)\n",
      " |-- votes: integer (nullable = true)\n",
      " |-- year: string (nullable = true)\n",
      "\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>actors</th>\n",
       "      <th>countries</th>\n",
       "      <th>description</th>\n",
       "      <th>directors</th>\n",
       "      <th>genre</th>\n",
       "      <th>imdb_url</th>\n",
       "      <th>img_url</th>\n",
       "      <th>languages</th>\n",
       "      <th>metascore</th>\n",
       "      <th>rating</th>\n",
       "      <th>runtime</th>\n",
       "      <th>tagline</th>\n",
       "      <th>title</th>\n",
       "      <th>users_rating</th>\n",
       "      <th>votes</th>\n",
       "      <th>year</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>[Timothée Chalamet, Elle Fanning, Liev Schreib...</td>\n",
       "      <td>[USA]</td>\n",
       "      <td>A young couple arrives in New York for a weeke...</td>\n",
       "      <td>[Woody Allen]</td>\n",
       "      <td>[Comedy, Romance]</td>\n",
       "      <td>https://www.imdb.com/title/tt7139936/</td>\n",
       "      <td>https://m.media-amazon.com/images/M/MV5BODAwZD...</td>\n",
       "      <td>[English]</td>\n",
       "      <td>44.0</td>\n",
       "      <td>PG-13</td>\n",
       "      <td>92 min</td>\n",
       "      <td>None</td>\n",
       "      <td>A Rainy Day in New York</td>\n",
       "      <td>6.6</td>\n",
       "      <td>NaN</td>\n",
       "      <td>2019</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>[Emilia Clarke, Hadley Fraser, Sylvia Panacion...</td>\n",
       "      <td>[USA]</td>\n",
       "      <td>Creepy, terrifying chapters from our book of h...</td>\n",
       "      <td>[Michael Escobedo, Kelly Hallmark, 4 more cred...</td>\n",
       "      <td>[Horror, Thriller]</td>\n",
       "      <td>https://www.imdb.com/title/tt12384178/</td>\n",
       "      <td>https://m.media-amazon.com/images/M/MV5BMDI5ZD...</td>\n",
       "      <td>[English]</td>\n",
       "      <td>NaN</td>\n",
       "      <td>18</td>\n",
       "      <td>91 min</td>\n",
       "      <td>When it comes to murder, they wrote the book</td>\n",
       "      <td>Murder Manual</td>\n",
       "      <td>2.4</td>\n",
       "      <td>192.0</td>\n",
       "      <td>2020</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>[Matthew Broderick, Alan Ruck, Mia Sara, Jeffr...</td>\n",
       "      <td>[USA]</td>\n",
       "      <td>A high school wise guy is determined to have a...</td>\n",
       "      <td>[John Hughes]</td>\n",
       "      <td>[Comedy]</td>\n",
       "      <td>https://www.imdb.com/title/tt0091042/</td>\n",
       "      <td>https://m.media-amazon.com/images/M/MV5BMDA0Nj...</td>\n",
       "      <td>[English, German]</td>\n",
       "      <td>61.0</td>\n",
       "      <td>PG-13</td>\n",
       "      <td>103 min</td>\n",
       "      <td>Because life is too beautiful a thing to waste.</td>\n",
       "      <td>Ferris Bueller's Day Off</td>\n",
       "      <td>7.8</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1986</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>[Robert De Niro, Nick Nolte, Jessica Lange, Ju...</td>\n",
       "      <td>[USA]</td>\n",
       "      <td>A convicted rapist, released from prison after...</td>\n",
       "      <td>[Martin Scorsese]</td>\n",
       "      <td>[Crime, Thriller]</td>\n",
       "      <td>https://www.imdb.com/title/tt0101540/</td>\n",
       "      <td>https://m.media-amazon.com/images/M/MV5BNjdhNz...</td>\n",
       "      <td>[English]</td>\n",
       "      <td>73.0</td>\n",
       "      <td>R</td>\n",
       "      <td>128 min</td>\n",
       "      <td>There is nothing in the dark that isn't there ...</td>\n",
       "      <td>Cape Fear</td>\n",
       "      <td>7.3</td>\n",
       "      <td>NaN</td>\n",
       "      <td>1991</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>[Lindsay Lohan, Rachel McAdams, Tina Fey, Tim ...</td>\n",
       "      <td>[USA, Canada]</td>\n",
       "      <td>Cady Heron is a hit with The Plastics, the A-l...</td>\n",
       "      <td>[Mark Waters]</td>\n",
       "      <td>[Comedy]</td>\n",
       "      <td>https://www.imdb.com/title/tt0377092/</td>\n",
       "      <td>https://m.media-amazon.com/images/M/MV5BMjE1MD...</td>\n",
       "      <td>[English, German, Vietnamese, Swahili]</td>\n",
       "      <td>66.0</td>\n",
       "      <td>PG-13</td>\n",
       "      <td>97 min</td>\n",
       "      <td>Survival of the Ruthless</td>\n",
       "      <td>Mean Girls</td>\n",
       "      <td>7.0</td>\n",
       "      <td>NaN</td>\n",
       "      <td>2004</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "                                              actors      countries  \\\n",
       "0  [Timothée Chalamet, Elle Fanning, Liev Schreib...          [USA]   \n",
       "1  [Emilia Clarke, Hadley Fraser, Sylvia Panacion...          [USA]   \n",
       "2  [Matthew Broderick, Alan Ruck, Mia Sara, Jeffr...          [USA]   \n",
       "3  [Robert De Niro, Nick Nolte, Jessica Lange, Ju...          [USA]   \n",
       "4  [Lindsay Lohan, Rachel McAdams, Tina Fey, Tim ...  [USA, Canada]   \n",
       "\n",
       "                                         description  \\\n",
       "0  A young couple arrives in New York for a weeke...   \n",
       "1  Creepy, terrifying chapters from our book of h...   \n",
       "2  A high school wise guy is determined to have a...   \n",
       "3  A convicted rapist, released from prison after...   \n",
       "4  Cady Heron is a hit with The Plastics, the A-l...   \n",
       "\n",
       "                                           directors               genre  \\\n",
       "0                                      [Woody Allen]   [Comedy, Romance]   \n",
       "1  [Michael Escobedo, Kelly Hallmark, 4 more cred...  [Horror, Thriller]   \n",
       "2                                      [John Hughes]            [Comedy]   \n",
       "3                                  [Martin Scorsese]   [Crime, Thriller]   \n",
       "4                                      [Mark Waters]            [Comedy]   \n",
       "\n",
       "                                 imdb_url  \\\n",
       "0   https://www.imdb.com/title/tt7139936/   \n",
       "1  https://www.imdb.com/title/tt12384178/   \n",
       "2   https://www.imdb.com/title/tt0091042/   \n",
       "3   https://www.imdb.com/title/tt0101540/   \n",
       "4   https://www.imdb.com/title/tt0377092/   \n",
       "\n",
       "                                             img_url  \\\n",
       "0  https://m.media-amazon.com/images/M/MV5BODAwZD...   \n",
       "1  https://m.media-amazon.com/images/M/MV5BMDI5ZD...   \n",
       "2  https://m.media-amazon.com/images/M/MV5BMDA0Nj...   \n",
       "3  https://m.media-amazon.com/images/M/MV5BNjdhNz...   \n",
       "4  https://m.media-amazon.com/images/M/MV5BMjE1MD...   \n",
       "\n",
       "                                languages  metascore rating  runtime  \\\n",
       "0                               [English]       44.0  PG-13   92 min   \n",
       "1                               [English]        NaN     18   91 min   \n",
       "2                       [English, German]       61.0  PG-13  103 min   \n",
       "3                               [English]       73.0      R  128 min   \n",
       "4  [English, German, Vietnamese, Swahili]       66.0  PG-13   97 min   \n",
       "\n",
       "                                             tagline  \\\n",
       "0                                               None   \n",
       "1       When it comes to murder, they wrote the book   \n",
       "2    Because life is too beautiful a thing to waste.   \n",
       "3  There is nothing in the dark that isn't there ...   \n",
       "4                           Survival of the Ruthless   \n",
       "\n",
       "                      title  users_rating  votes  year  \n",
       "0   A Rainy Day in New York           6.6    NaN  2019  \n",
       "1             Murder Manual           2.4  192.0  2020  \n",
       "2  Ferris Bueller's Day Off           7.8    NaN  1986  \n",
       "3                 Cape Fear           7.3    NaN  1991  \n",
       "4                Mean Girls           7.0    NaN  2004  "
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from pyspark.sql.functions import col\n",
    "\n",
    "movies = movies.withColumn(\"metascore\", col(\"metascore\").cast('integer'))\n",
    "movies = movies.withColumn(\"users_rating\", col(\"users_rating\").cast('float'))\n",
    "movies = movies.withColumn(\"votes\", col(\"votes\").cast('integer'))\n",
    "\n",
    "movies.printSchema()\n",
    "movies.toPandas().head()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "75ef5a8a",
   "metadata": {},
   "source": [
    "Because \"genre\" and \"language\" columns contained array data types and we wanted to train our model only by using the first value of the arrays, we used `getItem(0)` function, that gets an item at the first ordinal position and created two new columns (\"genre1\" and \"languages1\" respectively) that contained those values for each row (movie).\n",
    "\n",
    "Then, from the \"movies\" DataFrame we selected only the important features (\"users_rating\", \"metascore\", \"runtime\", \"genre1\" and \"languages1\") and saved the result into a new DataFrame named \"df\".\n",
    "\n",
    "We then used `cache()` method that Spark provides to cache the \"df\" DataFrame. `Cache()` is an optimization mechanism to store the intermediate computation of a Spark DataFrame to memory as a deserialized object (or if required storage is greater than available memory, it stores some of the excess partitions into a disk and reads the data from the disk when required), so that it can be reused in subsequent actions. Then we printed the first 20 rows of new DataFrame and we observed its schema in a tree format, using the `printSchema()` function."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "5a7000f6",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+---------+-------+---------+----------+\n",
      "|users_rating|metascore|runtime|   genre1|languages1|\n",
      "+------------+---------+-------+---------+----------+\n",
      "|         6.6|       44| 92 min|   Comedy|   English|\n",
      "|         2.4|     null| 91 min|   Horror|   English|\n",
      "|         7.8|       61|103 min|   Comedy|   English|\n",
      "|         7.3|       73|128 min|    Crime|   English|\n",
      "|         7.0|       66| 97 min|   Comedy|   English|\n",
      "|         5.3|       64| 87 min|  Fantasy|   English|\n",
      "|         6.2|       51|128 min|   Action|   English|\n",
      "|         8.0|       69|143 min|   Action|   English|\n",
      "|         8.3|       70|126 min|    Drama|   English|\n",
      "|         8.3|       84|149 min|Adventure|   English|\n",
      "|         8.3|       65|170 min|    Crime|   English|\n",
      "|         7.6|       78|102 min|   Comedy|   English|\n",
      "|         5.3|       45| 90 min|    Drama|   English|\n",
      "|         8.3|       76|116 min|    Drama|   English|\n",
      "|         7.4|       63|161 min|Adventure|   English|\n",
      "|         8.1|       72|116 min|    Drama|   English|\n",
      "|         8.2|       88|115 min|   Comedy|   English|\n",
      "|         7.8|       73|143 min|Adventure|   English|\n",
      "|         7.5|       68|140 min|   Action|   English|\n",
      "|         6.6|       51|124 min|    Drama|   English|\n",
      "+------------+---------+-------+---------+----------+\n",
      "only showing top 20 rows\n",
      "\n",
      "root\n",
      " |-- users_rating: float (nullable = true)\n",
      " |-- metascore: integer (nullable = true)\n",
      " |-- runtime: string (nullable = true)\n",
      " |-- genre1: string (nullable = true)\n",
      " |-- languages1: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "movies = movies.withColumn(\"genre1\",movies.genre.getItem(0)).withColumn(\"languages1\",movies.languages.getItem(0))\n",
    "df = movies.select(\"users_rating\",\"metascore\",\"runtime\",\"genre1\",\"languages1\")\n",
    "df.cache().show()\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ecb423c7",
   "metadata": {},
   "source": [
    "From observing the tree schema above, we noticed that only the \"runtime\" column was at a data type different from that implied from its contained values. This was happening, because it contained the metric of the column, which was in minutes. \n",
    "\n",
    "Firstly, we splitted the \"runtime\" column, using the `split` function, into an array of two values, the first containing the value of the metric and the second the metric. Then, we checked if minutes was the only metric implied. We found that there also existed rows with missing value in the metric field (2nd value) and we checked if for them the value of the metric (1st field) existed and found that there did not exist rows with runtime value with no metric.\n",
    "\n",
    "We then retained only the first value of the \"runtime\" column, dropping the second value from the DataFrame, printed the top 20 rows and the schema of it, in order to observe that everything worked well."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "ae52915f",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Check that runtime is expressed only in minutes:\n",
      "\n",
      "+-----------------------+-----+\n",
      "|element_at(run_time, 2)|count|\n",
      "+-----------------------+-----+\n",
      "|                   null|12165|\n",
      "|                    min|49893|\n",
      "+-----------------------+-----+\n",
      "\n",
      "Check if exists runtime value with no metric (e.g. minutes):\n",
      "\n",
      "[]\n",
      "\n",
      "Top 20 rows after all transformations:\n",
      "\n",
      "+------------+---------+-------+---------+----------+\n",
      "|users_rating|metascore|runtime|   genre1|languages1|\n",
      "+------------+---------+-------+---------+----------+\n",
      "|         6.6|       44|     92|   Comedy|   English|\n",
      "|         2.4|     null|     91|   Horror|   English|\n",
      "|         7.8|       61|    103|   Comedy|   English|\n",
      "|         7.3|       73|    128|    Crime|   English|\n",
      "|         7.0|       66|     97|   Comedy|   English|\n",
      "|         5.3|       64|     87|  Fantasy|   English|\n",
      "|         6.2|       51|    128|   Action|   English|\n",
      "|         8.0|       69|    143|   Action|   English|\n",
      "|         8.3|       70|    126|    Drama|   English|\n",
      "|         8.3|       84|    149|Adventure|   English|\n",
      "|         8.3|       65|    170|    Crime|   English|\n",
      "|         7.6|       78|    102|   Comedy|   English|\n",
      "|         5.3|       45|     90|    Drama|   English|\n",
      "|         8.3|       76|    116|    Drama|   English|\n",
      "|         7.4|       63|    161|Adventure|   English|\n",
      "|         8.1|       72|    116|    Drama|   English|\n",
      "|         8.2|       88|    115|   Comedy|   English|\n",
      "|         7.8|       73|    143|Adventure|   English|\n",
      "|         7.5|       68|    140|   Action|   English|\n",
      "|         6.6|       51|    124|    Drama|   English|\n",
      "+------------+---------+-------+---------+----------+\n",
      "only showing top 20 rows\n",
      "\n",
      "DataFrame Schema after all transformations:\n",
      "\n",
      "root\n",
      " |-- users_rating: float (nullable = true)\n",
      " |-- metascore: integer (nullable = true)\n",
      " |-- runtime: integer (nullable = true)\n",
      " |-- genre1: string (nullable = true)\n",
      " |-- languages1: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import split, col, element_at\n",
    "\n",
    "df = df.withColumn(\"run_time\",split(col(\"runtime\"),\" \"))\n",
    "print(\"Check that runtime is expressed only in minutes:\\n\")\n",
    "df.groupBy(element_at(col(\"run_time\"),2)).count().show()\n",
    "print(\"Check if exists runtime value with no metric (e.g. minutes):\\n\")\n",
    "print(df.filter(element_at(col(\"run_time\"),2).isNull() & element_at(col(\"run_time\"),2).isNotNull()).collect())\n",
    "df = df.withColumn(\"runtime\",element_at(col(\"run_time\"),1))\n",
    "df = df.withColumn(\"runtime\", col(\"runtime\").cast('integer'))\n",
    "df = df.drop(\"run_time\")\n",
    "print(\"\\nTop 20 rows after all transformations:\\n\")\n",
    "df.show()\n",
    "print(\"DataFrame Schema after all transformations:\\n\")\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cda4debd",
   "metadata": {},
   "source": [
    "In the following block of code, we present that there exist in total a lot of null values. Our response variable (\"users_rating\") has 2 missing values (out of the 62058), but the \"genre\", \"runtime\" and \"metascore\" variables have approximately 2%, 20% and 85% respectively of their values missing."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "06490eb9",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+---------+-------+------+----------+\n",
      "|users_rating|metascore|runtime|genre1|languages1|\n",
      "+------------+---------+-------+------+----------+\n",
      "|           2|    52447|  12167|  1252|         0|\n",
      "+------------+---------+-------+------+----------+\n",
      "\n",
      "Column-wise percentages of missing values per total number of rows:\n",
      "\n",
      "+------------+---------+-------+------+----------+\n",
      "|users_rating|metascore|runtime|genre1|languages1|\n",
      "+------------+---------+-------+------+----------+\n",
      "|         0.0|    84.51|  19.61|  2.02|       0.0|\n",
      "+------------+---------+-------+------+----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import when, isnan, round, count\n",
    "\n",
    "df.select([count(when(isnan(c) | col(c).isNull(), c)).alias(c) for c in df.columns]).show()\n",
    "print(\"Column-wise percentages of missing values per total number of rows:\\n\")\n",
    "df.select([round((count(when(isnan(c) | col(c).isNull(), c))/df.count()*100),2).alias(c) for c in df.columns]).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "91412ed9",
   "metadata": {},
   "source": [
    "Linear Regression requires numeric features. Because we want to consider the \"genre1\" and \"languages1\" variables that are categorical, we must use One-Hot Encoding, that converts categorical variables into binary vector variables that only take on values 0 or 1. One-Hot Encoding in Spark is a two-step process. We firstly use the StringIndexer, followed by the OneHotEncoder. In particular:\n",
    "\n",
    "- `StringIndexer()` creates a label indexer that maps a string column of labels to a column of label indices. For example, it might convert the values \"English\", \"German\", and \"Spanish\" to 0, 1, and 2.\n",
    "- `OneHotEncoder()` maps a column of category indices to a column of binary vectors, with at most one \"1\" in each row that indicates the category index for that row. For example, for a variable with 5 categories, an input value of 2.0 would map to an output vector of [0.0, 0.0, 1.0, 0.0].\n",
    "\n",
    "The following block of code defines the `StringIndexer()` and `OneHotEncoder()` but does not apply it to any data yet.\n",
    "Because we have missing values, we specify to StringIndexer to handle invalid values i.e., to put unseen labels in a special additional bucket, at index numLabels, by setting the parameter handleInvalid = \"keep\"."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "3f9af2bc",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import StringIndexer, OneHotEncoder\n",
    "\n",
    "# We determine which of the columns are categorical.\n",
    "categoricalCols = [\"genre1\", \"languages1\"]\n",
    "\n",
    "# The following two lines are estimators. They return functions that we will later apply to transform the dataset.\n",
    "stringIndexer = StringIndexer(inputCols=categoricalCols, outputCols=[x + \"Index\" for x in categoricalCols], \\\n",
    "                              handleInvalid = \"keep\")\n",
    "encoder = OneHotEncoder(inputCols=stringIndexer.getOutputCols(), outputCols=[x + \"OHE\" for x in categoricalCols])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "41adef7a",
   "metadata": {},
   "source": [
    "Because Linear Regression algorithm requires a single features column as input, we must use a feature transformer that merges multiple columns into a vector column i.e., `VectorAssembler()`.\n",
    "\n",
    "The following block of code defines the `VectorAssembler()` but does not apply it to any data yet. Because we have missing values and we do not want to throw an error (default: handleInvalid = \"error\") or return relevant number of NaN in the output (handleInvalid = \"keep\") we specify to VectorAssembler to filter out rows with invalid data i.e., setting the parameter handleInvalid = \"skip\"."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "44b0decc",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.feature import VectorAssembler\n",
    "\n",
    "# This includes both the numeric columns and the one-hot encoded binary vector columns in our dataset.\n",
    "numericCols = [\"metascore\", \"runtime\"]\n",
    "assemblerInputs = [c + \"OHE\" for c in categoricalCols] + numericCols\n",
    "vecAssembler = VectorAssembler(inputCols=assemblerInputs, outputCol=\"features\", handleInvalid = \"skip\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "270133d4",
   "metadata": {},
   "source": [
    "We are going to use a [Linear Regression](https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.regression.LinearRegression.html) model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "9dd5c59a",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.regression import LinearRegression\n",
    "\n",
    "lr = LinearRegression(featuresCol = 'features', labelCol='users_rating')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0486950b",
   "metadata": {},
   "source": [
    "### Prepare the training and testing datasets (80%-20%)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "15481238",
   "metadata": {},
   "source": [
    "For predictive regression models we need data to train our model (the \"training set\") and data to evaluate the effectiveness/accuracy of the model (the \"test set\"). In the following, we divide our data accordingly.\n",
    "\n",
    "We made the training dataset to contain the 80% of the data, and the testing dataset to contain the rest. Also, we are using a \"seed\" so that for each run we have exactly the same split (for reproducibility). Finally, we cache our training dataset to memory, using `cache()` function, for efficiency reasons, since the training dataset will be used multiple times during the training phase."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "70942d6e",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Training: 49579\n",
      "Testing: 12479\n"
     ]
    }
   ],
   "source": [
    "trainDF, testDF = df.randomSplit([0.8, 0.2], seed=1)\n",
    "print(f\"Training: {trainDF.cache().count()}\") # Cache because accessing training data multiple times\n",
    "print(f\"Testing: {testDF.count()}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "73b6a4ed",
   "metadata": {},
   "source": [
    "### Train the model"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "10f74006",
   "metadata": {},
   "source": [
    "A Pipeline is an ordered list of transformers and estimators. You can define a pipeline to automate and ensure repeatability of the transformations to be applied to a dataset. In this step, we define the pipeline and then apply it to the test dataset. A Pipeline is an estimator. The pipeline.fit() method returns a PipelineModel, which is a transformer.\n",
    "\n",
    "We then defined a simple pipeline, which acts as an estimator. A `Pipeline()` consists of a sequence of stages, each of which is either an Estimator or a Transformer. When Pipeline.fit() is called to the \"training dataset\", the stages are executed in order. If a stage is an Estimator, its Estimator.fit() method will be called on the input dataset to fit a model. Then the model, which is a transformer, will be used to transform the dataset as the input to the next stage. If a stage is a Transformer, its Transformer.transform() method will be called to produce the dataset for the next stage. \n",
    "\n",
    "The fitted model from a Pipeline is a PipelineModel, which consists of fitted models and transformers, corresponding to the pipeline stages. We then apply it to the \"testing dataset\" and we display the top 10 rows with the VectorAssembler Features, the actual response variable (\"users_rating\") of the \"testing dataset\" and the predicted value for each row."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "a7018690",
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+------------+------------------+\n",
      "|            features|users_rating|        prediction|\n",
      "+--------------------+------------+------------------+\n",
      "|(104,[0,25,102,10...|         1.4| 4.724988855630935|\n",
      "|(104,[0,25,102,10...|         2.1| 4.618920202712481|\n",
      "|(104,[2,25,102,10...|         2.1| 5.332971605582712|\n",
      "|(104,[2,25,102,10...|         2.4| 4.687398431089768|\n",
      "|(104,[2,25,102,10...|         2.5|4.7297133675361245|\n",
      "|(104,[2,25,102,10...|         2.5| 5.696608766845739|\n",
      "|(104,[0,25,102,10...|         2.8| 4.520454673745588|\n",
      "|(104,[3,25,102,10...|         2.8| 4.287562131953017|\n",
      "|(104,[2,25,102,10...|         2.9| 5.772452465159654|\n",
      "|(104,[0,25,102,10...|         3.0|4.4750234712379235|\n",
      "+--------------------+------------+------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml import Pipeline\n",
    "\n",
    "# Define the pipeline based on the stages created in previous steps.\n",
    "pipeline = Pipeline(stages=[stringIndexer, encoder, vecAssembler, lr])\n",
    "\n",
    "# Define the pipeline model.\n",
    "pipelineModel = pipeline.fit(trainDF)\n",
    "\n",
    "# Apply the pipeline model to the test dataset to classify the respective samples.\n",
    "predDF = pipelineModel.transform(testDF)\n",
    "\n",
    "# Display the predictions from the model\n",
    "predDF.select(\"features\", \"users_rating\", \"prediction\").show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "735b910d",
   "metadata": {},
   "source": [
    "### Evaluate the accuracy of the model (based on the Rsquared metric)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a73f8ba5",
   "metadata": {},
   "source": [
    "To evaluate the accuracy of the model, based on the R^2 metric, we use the `RegressionEvaluator()`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "c2d596ef",
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "R Squared (R^2) on test data = 0.533488\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.evaluation import RegressionEvaluator\n",
    "\n",
    "lr_evaluator = RegressionEvaluator(predictionCol=\"prediction\", labelCol=\"users_rating\",metricName=\"r2\")\n",
    "print(\"R Squared (R^2) on test data = %g\" % lr_evaluator.evaluate(predDF))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7048ebd0",
   "metadata": {},
   "source": [
    "### Tune hyperparameters"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1eb2ab77",
   "metadata": {},
   "source": [
    "We then used `ParamGridBuilder()` and `CrossValidator()` to tune the model with 20 values for \"regParam\" and 10 for \"elasticNetParam\", for a total of 20 x 10 = 200 hyperparameter combinations for `CrossValidator` to examine and with the use of the pipeline we created before as the estimator. \n",
    "\n",
    "K-fold cross validation performs model selection by splitting the dataset into a set of non-overlapping randomly partitioned folds which are used as separate training and test datasets e.g., with k=10 folds, K-fold cross validation will generate 10 (training, test) dataset pairs, each of which uses 9/10 of the data for training and 1/10 for testing. Each fold is used as the test set exactly once. As evaluator of the 10-fold CrossValidator, we used the RegressionEvaluator used before, based on R^2 metric, thus the output of it will definetely lead to a better model. The procedure can be parallelized, so we used 4 threads for parallel computations.\n",
    "\n",
    "We commented the 10-fold Cross Validation procedure taken below and saved the hyperparameters for instant creation of the tuned Linear Regression model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "b1f9a66f",
   "metadata": {},
   "outputs": [],
   "source": [
    "# import numpy\n",
    "# from pyspark.ml.tuning import ParamGridBuilder, CrossValidator\n",
    "\n",
    "# paramGrid = (ParamGridBuilder()\n",
    "#              .addGrid(lr.regParam, list(numpy.arange(0.01, 2.10, 0.10)))\n",
    "#              .addGrid(lr.elasticNetParam, list(numpy.arange(0.0, 1.1, 0.1)))\n",
    "#              .build())\n",
    "\n",
    "# # Create a 10-fold CrossValidator\n",
    "# cv = CrossValidator(estimator=pipeline, estimatorParamMaps=paramGrid, evaluator=lr_evaluator, numFolds=10, parallelism = 4)\n",
    "\n",
    "# # Run cross validations. This step takes a few minutes and returns the best model found from the cross validation.\n",
    "# cvModel = cv.fit(trainDF)\n",
    "\n",
    "# lr = LinearRegression(featuresCol = 'features', labelCol='users_rating', \\\n",
    "#                       regParam=cvModel.bestModel.stages[-1]._java_obj.parent().getRegParam(), \\\n",
    "#                       elasticNetParam=cvModel.bestModel.stages[-1]._java_obj.parent().getElasticNetParam())\n",
    "\n",
    "lr = LinearRegression(featuresCol = 'features', labelCol='users_rating', regParam=0.01, elasticNetParam=0.8)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9f4d30d1",
   "metadata": {},
   "source": [
    "Finally, we fitted the tuned model on the \"train dataset\" and tested it on the \"test dataset\". From the evaluation of the accuracy of the model, we found a slightly better R^2 metric."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "01ce83f4",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "R Squared (R^2) on test data = 0.537574\n"
     ]
    }
   ],
   "source": [
    "# Define the pipeline based on the stages created in previous steps.\n",
    "pipeline = Pipeline(stages=[stringIndexer, encoder, vecAssembler, lr])\n",
    "\n",
    "# Define the pipeline model.\n",
    "pipelineModel = pipeline.fit(trainDF)\n",
    "\n",
    "# Apply the pipeline model to the test dataset to classify the respective samples.\n",
    "predDF = pipelineModel.transform(testDF)\n",
    "\n",
    "lr_evaluator = RegressionEvaluator(predictionCol=\"prediction\", labelCol=\"users_rating\",metricName=\"r2\")\n",
    "print(\"R Squared (R^2) on test data = %g\" % lr_evaluator.evaluate(predDF))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1f4aca74",
   "metadata": {},
   "source": [
    "### End Spark Session"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c4c8fb9f",
   "metadata": {},
   "source": [
    "Finally, we stopped the underlying Spark Session."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "f0203adc",
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.stop()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.9.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}