notebooks/HandsOn-SparkSQL_exercises.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## This notebook is part of the Apache Spark training delivered by CERN-IT\n",
    "### Spark SQL Hands-On Lab\n",
    "Contact: Luca.Canali@cern.ch"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Run this notebook from Jupyter with Python kernel\n",
    "- When using on CERN SWAN, do not attach the notebook to a Spark cluster, but rather run locally on the SWAN container\n",
    "- If running this outside CERN SWAN, please make sure to have PySpark installed: `pip install pyspark`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Examples datasets\n",
    "The following examples use sample data provided in the repository.  \n",
    "We will use the movielens dataset from Kaggle, credits: https://www.kaggle.com/grouplens/movielens-20m-dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create Spark Session, you need this to work with Spark\n",
    "from pyspark.sql import SparkSession\n",
    "spark = SparkSession.builder \\\n",
    "        .appName(\"My spark example app\")  \\\n",
    "        .master(\"local[*]\") \\\n",
    "        .config(\"spark.driver.memory\",\"8g\") \\\n",
    "        .config(\"spark.ui.showConsoleProgress\", \"false\") \\\n",
    "        .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# sets the path to the directory with datafiles\n",
    "PATH = \"../data/\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "ratings = spark.read.option(\"header\",\"true\").csv(PATH + \"ratings1.csv.gz\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ratings.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ratings.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# infer schema needs to go through the data to estimate the schema, this takes time\n",
    "\n",
    "%time ratings = spark.read.option(\"header\",\"true\").option(\"inferSchema\", \"true\").csv(PATH + \"ratings1.csv.gz\")\n",
    "\n",
    "# note ratings*.csv.gz will read rating1.csv.gz and ratings2.csv.gz, more data, however slower to run\n",
    "# spark.read.option(\"header\",\"true\").option(\"inferSchema\", \"true\").csv(PATH + \"ratings*.csv.gz\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ratings.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# movielens dataset\n",
    "movies = spark.read.option(\"header\",\"true\").option(\"inferSchema\", \"true\").csv(PATH + \"movies.csv.gz\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "movies.show(5, False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "movies.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tags = spark.read.option(\"header\",\"true\").option(\"inferSchema\", \"true\").csv(PATH + \"tags.csv.gz\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "tags.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Register the dataframes as Spark Temporary Views"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "ratings.createOrReplaceTempView(\"ratings\")\n",
    "movies.createOrReplaceTempView(\"movies\")\n",
    "tags.createOrReplaceTempView(\"tags\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# note what happens when we query a table in a csv file with a filter\n",
    "spark.sql(\"select movieId, title from movies where movieId=1\").explain()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# cache the tables, to improve the performance of the rest of the queries in the notebook\n",
    "# note: default caching level is MEMORY_AND_DISK (i.e. caching in memory if enough heap is available)\n",
    "# note: caching is lazily executed, so a count() action is added to make the operation happen\n",
    "# this operation may take a couple of minutes\n",
    "\n",
    "r = ratings.cache().count()\n",
    "m = movies.cache().count()\n",
    "t = tags.cache().count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(f\"Num ratings = {r}\\nNum tags = {t}\\nNum movies = {m}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### SQL Queries on the loaded tables\n",
    "#### 1) How many movies produced per year?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Add the column Year to \"movies\"\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# retrieve the number of movies per year\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%matplotlib notebook\n",
    "import matplotlib.pyplot as plt\n",
    "import pandas as pd\n",
    "plt.style.use('seaborn-darkgrid')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# prolt the results\n",
    "m_yr.plot(x='year',y='count',kind='line', title='Movies per year');"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 2) Top movies by number of ratings"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# A query to perform a join operation between movies and ratings\n",
    "# Find the highest rated movies\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### What happened in the background? How did the join happen?\n",
    "#### How is the query executed? Can we get more information?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 3) Highly rated movies\n",
    "Find the top 5 highly rated movies"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Drill down on the top entries:\n",
    "   - How many reviews contributed to this rating?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Lets only take in account movies that have more than 100 reviews"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "#### 4) Find the top rated movie of every year since 2000"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# End the Spark application\n",
    "spark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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": 1
}