03-Spark/003.01-Harder-Problems-in-PySpark(Questions-only).ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "da26dd6e-480d-43f1-8564-2125e3d60a5a",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "source": [
    "# Data Analysis using __PySpark__  \n",
    "*Fun with the __MovieLens__ dataset*  \n",
    "\n",
    "**Harder Problems, Moar Fun**"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1924908a-2790-4088-9087-3e2b1e2d12f4",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "source": [
    "<font color='green'>__Support for Google Colab__  </font>\n",
    "\n",
    "open this notebook in Colab using the following button:  \n",
    "  \n",
    "<a href=\"https://colab.research.google.com/github/shauryashaurya/learn-data-munging/blob/main/03-Spark/003.01-Harder-Problems-in-PySpark(Questions-only).ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>  \n",
    "\n",
    "<font color='green'>uncomment and execute the cell below to setup and run this Spark notebook on Google Colab.</font>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3d7b62ce-bc23-4091-9605-e8bff9b18a00",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true
   },
   "source": [
    "# Setup, Loading Data, yada yada... \n",
    "\n",
    "we've done this before..."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "eb71efec-ab7e-4995-8658-6710707a6291",
   "metadata": {},
   "source": [
    "## update Jan 2025 \n",
    "It has been very frustrating to run spark 3.5.x with Python 3.12  \n",
    "While the official documentation says it works, the python workers for pySpark have been crashing randomly.   \n",
    "Not cool!  \n",
    "\n",
    "So for the time being we'll stick with Spark 3.5.4 (ditto for pySpark - v3.5.4) with Python 3.11.xSo for the time being we'll stick with Spark 3.5.4 (ditto for pySpark - v3.5.4) with Python 3.11.x\n",
    "See [this](https://www.reddit.com/r/dataengineering/comments/1dupogi/wasted_45_hours_to_install_pyspark_locally_pain/) reddit post that shares my frustration.  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "3ac6adf3-5ba1-4b8e-930a-af0e9d98e6ee",
   "metadata": {
    "editable": true,
    "slideshow": {
     "slide_type": ""
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "# # SETUP FOR COLAB: select all the lines below and uncomment (CTRL+/ on windows)\n",
    "\n",
    "# # grab spark\n",
    "# # as of Jan 2025, the *working* version is 3.5.4, get the link from Apache Spark's website\n",
    "# ! wget -q https://dlcdn.apache.org/spark/spark-3.5.4/spark-3.5.4-bin-hadoop3.tgz\n",
    "# # unzip spark\n",
    "# !tar xf spark-3.5.4-bin-hadoop3.tgz\n",
    "# # install findspark package\n",
    "# !pip install -q findspark\n",
    "# # Let's download and unzip the MovieLens 25M Dataset as well.\n",
    "# ! mkdir ./../data\n",
    "# ! wget -q https://files.grouplens.org/datasets/movielens/ml-25m.zip\n",
    "# ! unzip ./ml-25m.zip -d ./../data/\n",
    "\n",
    "# # got to provide JAVA_HOME and SPARK_HOME vairables\n",
    "# import os\n",
    "# # IMPORTANT - check the version of java, use 11 or 17, code not tested on 21 yet\n",
    "# os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-11-openjdk-amd64\"\n",
    "# # IMPORTANT - UPDATE THE SPARK_HOME PATH BASED ON THE PACKAGE YOU DOWNLOAD\n",
    "# os.environ[\"SPARK_HOME\"] = \"/content/spark-3.5.4-bin-hadoop3\"\n",
    "# ! echo \"DONE\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8d0b04da-4558-4b16-895a-a4210b4e2309",
   "metadata": {},
   "source": [
    "**Citation**:  \n",
    "*F. Maxwell Harper and Joseph A. Konstan.* 2015.  \n",
    "The MovieLens Datasets: History and Context.  \n",
    "ACM Transactions on Interactive Intelligent Systems (TiiS) 5, 4: 19:1–19:19. <https://doi.org/10.1145/2827872>  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5519b306-210b-408c-af70-2bf1cdfcd55e",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "id": "22f15c88-eef6-4c8c-b5d0-bbe73496db3e",
   "metadata": {},
   "source": [
    "## Setup the Spark Cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "f9e79f7c-0e7e-4df9-9ab7-1dfda672315e",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Step 1: initialize findspark\n",
    "import findspark\n",
    "\n",
    "findspark.init()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "f3c4ca5c-7717-4930-86be-2e9d6b88e6ed",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'3.5.4'"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Step 2: import pyspark\n",
    "import pyspark\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "pyspark.__version__"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "5dfdf716-15cd-4a5f-a7ff-d402e65b7c64",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Step 3: Create a spark session\n",
    "\n",
    "# using local[*] to use as many logical cores as available, use 1 when in doubt\n",
    "# 'local[1]' indicates spark on 1 core on the local machine or specify the number of cores needed\n",
    "# use .config(\"spark.some.config.option\", \"some-value\") for additional configuration\n",
    "\n",
    "spark = (\n",
    "    SparkSession.builder.master(\"local[1]\")\n",
    "    .appName(\"Analyzing Movielens Data\")\n",
    "    .getOrCreate()\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "e8bdd519-7056-4543-9666-625dd06c43f6",
   "metadata": {},
   "outputs": [],
   "source": [
    "# spark"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e8a21403-7733-4a3f-8f5e-20da68519b67",
   "metadata": {},
   "source": [
    "## Load data, carefull with the schema"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "99c7131a-c398-432e-baf3-77803150cbc2",
   "metadata": {},
   "source": [
    "### Schema Spec"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bdd6a14f-9bd2-4ead-a66e-feb199eb52b5",
   "metadata": {},
   "source": [
    "Here's the list of files (as of Aug 2022) that you get when you unzip the dataset:\n",
    "1. **movies**.csv - list of movies with at least one rating.  \n",
    "    Header: ```movieId,title,genres```  \n",
    "1. **links**.csv - IDs to generate links to the movie listing on imdb.com and themoviedb.org  \n",
    "    Header: ```movieId,imdbId,tmdbId```  \n",
    "1. **ratings**.csv - Each line of this file after the header row represents one rating of one movie by one user.  \n",
    "    Header: ```userId,movieId,rating,timestamp```  \n",
    "1. **tags**.csv - Each line of this file after the header row represents one tag applied to one movie by one user.  \n",
    "    Header: ```userId,movieId,tag,timestamp```  \n",
    "1. Tag Genome: The tag genome contains tag relevance scores for movies. See [this](http://files.grouplens.org/papers/tag_genome.pdf)  \n",
    "\t1. **genome-tags**.csv - A list of tags  \n",
    "    Header: ```tagId,tag```  \n",
    "\t1. **genome-scores**.csv - Each movie in the genome has a relevance score value for every tag in the genome  \n",
    "    Header: ```movieId,tagId,relevance```  \n",
    "1. README.txt - Check out the README.txt for more details about the files.  "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "57e53160-8e6e-483b-9098-d6ee250630ef",
   "metadata": {},
   "source": [
    "### Data encoding details"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6340bbdd-b448-405d-895b-d4313151739d",
   "metadata": {},
   "source": [
    "From the Readme file, we have the following observations about the data:\n",
    "1. Each file is a CSV with a single header row\n",
    "1. Separator char is ```,```\n",
    "1. Escape char is ```\"```\n",
    "1. Encoding is UTF-8\n",
    "\n",
    "Let's set these options when reading the CSV files."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d2366035-81c6-4cd5-9bbb-48d85d4ec72a",
   "metadata": {},
   "source": [
    "### Specify the schema for Spark  \n",
    "  \n",
    "Avoid ```inferSchema``` as much as possible, just cleaner"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "fd6d7594-3185-4207-a9a7-bb6a00e86148",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import functions as F\n",
    "from pyspark.sql.window import Window\n",
    "from pyspark.sql.types import *\n",
    "from pyspark.rdd import RDD"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "f2c89daf-690c-4e90-86bf-e95644ca4800",
   "metadata": {},
   "outputs": [],
   "source": [
    "#\n",
    "schema_movies = StructType(\n",
    "    [\n",
    "        StructField(\"movieId\", StringType(), False),\n",
    "        StructField(\"title\", StringType(), False),\n",
    "        StructField(\"genres\", StringType(), True),\n",
    "    ]\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "2a773db7-bc73-4ce8-beac-d6a8032e8f2b",
   "metadata": {},
   "outputs": [],
   "source": [
    "#\n",
    "schema_links = StructType(\n",
    "    [\n",
    "        StructField(\"movieId\", StringType(), False),\n",
    "        StructField(\"imdbId\", StringType(), True),\n",
    "        StructField(\"tmdbId\", StringType(), True),\n",
    "    ]\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "4b418a4e-8971-4879-a616-bdf80f88b123",
   "metadata": {},
   "outputs": [],
   "source": [
    "#\n",
    "schema_ratings = StructType(\n",
    "    [\n",
    "        StructField(\"userId\", StringType(), False),\n",
    "        StructField(\"movieId\", StringType(), False),\n",
    "        StructField(\"rating\", FloatType(), True),\n",
    "        StructField(\"timestamp\", StringType(), True),\n",
    "    ]\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "f2e3e663-46a5-47ad-84db-1136bfef65cd",
   "metadata": {},
   "outputs": [],
   "source": [
    "#\n",
    "schema_tags = StructType(\n",
    "    [\n",
    "        StructField(\"userId\", StringType(), False),\n",
    "        StructField(\"movieId\", StringType(), False),\n",
    "        StructField(\"tag\", StringType(), True),\n",
    "        StructField(\"timestamp\", StringType(), True),\n",
    "    ]\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "5fcd5476-2c53-40be-beb9-19d70dd6f2c1",
   "metadata": {},
   "outputs": [],
   "source": [
    "#\n",
    "schema_genome_tags = StructType(\n",
    "    [\n",
    "\t\tStructField(\"tagId\", StringType(), False), \n",
    "\t\tStructField(\"tag\", StringType(), False)\n",
    "\t]\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "50a9fd1e-1090-4e0c-97fa-52089484c1ed",
   "metadata": {},
   "outputs": [],
   "source": [
    "#\n",
    "# using arbitrary precision signed decimals (java.math.BigDecimal) for relevance scores\n",
    "schema_genome_scores = StructType(\n",
    "    [\n",
    "        StructField(\"movieId\", StringType(), False),\n",
    "        StructField(\"tagId\", StringType(), False),\n",
    "        StructField(\"relevance\", DecimalType(), False),\n",
    "    ]\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8b94a9a8-b62c-42ac-bca8-0097807e7c5a",
   "metadata": {},
   "source": [
    "### Specify the location of your data  "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "95e861cb-4461-488c-8842-44612b39ee03",
   "metadata": {},
   "source": [
    "Change this folder if you are saving the data at the different place"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "e9e0a195-bbd9-4822-ad93-581683470f9c",
   "metadata": {},
   "outputs": [],
   "source": [
    "datalocation = \"../data/ml-25m/\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "bea41d49-427f-4678-b7b0-55fc727eef1c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# specify file names\n",
    "file_path_movies = datalocation + \"movies.csv\"\n",
    "file_path_links = datalocation + \"links.csv\"\n",
    "file_path_ratings = datalocation + \"ratings.csv\"\n",
    "file_path_tags = datalocation + \"tags.csv\"\n",
    "file_path_genome_tags = datalocation + \"genome-tags.csv\"\n",
    "file_path_genome_scores = datalocation + \"genome-scores.csv\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "31b47393-14bf-4fe4-bd56-f9cf75d1289d",
   "metadata": {},
   "source": [
    "### Load the data and review"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f5cdfd15-b22d-4c35-98ac-4f70c9c971c0",
   "metadata": {},
   "source": [
    "Let's load each file in turn and observe, just to get a sense of familiarity with the data.  "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3af0b221-9372-40a7-9fa1-472c789f24a7",
   "metadata": {},
   "source": [
    "#### Movies"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "201f52d6-a5fe-4f4f-9b35-165fa4b14ff2",
   "metadata": {},
   "outputs": [],
   "source": [
    "movies_raw = (\n",
    "    spark.read.format(\"csv\")\n",
    "    .option(\"encoding\", \"UTF-8\")\n",
    "    .option(\"header\", True)\n",
    "    .option(\"sep\", \",\")\n",
    "    .option(\"escape\", '\"')\n",
    "    .schema(schema_movies)\n",
    "    .load(file_path_movies)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "fd07ca00-b0a3-4fea-a5d0-ae742fe3887c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----------------------------------+-------------------------------------------+\n",
      "|movieId|title                             |genres                                     |\n",
      "+-------+----------------------------------+-------------------------------------------+\n",
      "|1      |Toy Story (1995)                  |Adventure|Animation|Children|Comedy|Fantasy|\n",
      "|2      |Jumanji (1995)                    |Adventure|Children|Fantasy                 |\n",
      "|3      |Grumpier Old Men (1995)           |Comedy|Romance                             |\n",
      "|4      |Waiting to Exhale (1995)          |Comedy|Drama|Romance                       |\n",
      "|5      |Father of the Bride Part II (1995)|Comedy                                     |\n",
      "|6      |Heat (1995)                       |Action|Crime|Thriller                      |\n",
      "|7      |Sabrina (1995)                    |Comedy|Romance                             |\n",
      "|8      |Tom and Huck (1995)               |Adventure|Children                         |\n",
      "|9      |Sudden Death (1995)               |Action                                     |\n",
      "|10     |GoldenEye (1995)                  |Action|Adventure|Thriller                  |\n",
      "+-------+----------------------------------+-------------------------------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Spark collects all transformations needed\n",
    "# and execution doesn't begin until an \"action\" is triggered\n",
    "# \n",
    "# 'show' triggers a partial execution \n",
    "#  'show' - limiting computation (where relevant) to the number of rows you want to display\n",
    "movies_raw.show(10, False)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "790b82b2-e8db-442f-a6b4-37e4f1639856",
   "metadata": {},
   "source": [
    "#### Links"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "555f8570-a427-4d8a-bd66-a3ed2d30f5c9",
   "metadata": {},
   "outputs": [],
   "source": [
    "links_raw = (\n",
    "    spark.read.format(\"csv\")\n",
    "    .option(\"encoding\", \"UTF-8\")\n",
    "    .option(\"header\", True)\n",
    "    .option(\"sep\", \",\")\n",
    "    .option(\"escape\", '\"')\n",
    "    .schema(schema_links)\n",
    "    .load(file_path_links)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "420a215d-5a5a-474f-8d3e-9d1544d01268",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------+------+\n",
      "|movieId|imdbId |tmdbId|\n",
      "+-------+-------+------+\n",
      "|1      |0114709|862   |\n",
      "|2      |0113497|8844  |\n",
      "|3      |0113228|15602 |\n",
      "|4      |0114885|31357 |\n",
      "|5      |0113041|11862 |\n",
      "|6      |0113277|949   |\n",
      "|7      |0114319|11860 |\n",
      "|8      |0112302|45325 |\n",
      "|9      |0114576|9091  |\n",
      "|10     |0113189|710   |\n",
      "+-------+-------+------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "links_raw.show(10, False)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3315d161-a790-44c2-89c4-eb3e9e7ad4be",
   "metadata": {},
   "source": [
    "#### Ratings"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "53e44596-6ccf-4013-9603-e275b77ae23b",
   "metadata": {},
   "outputs": [],
   "source": [
    "ratings_raw = (\n",
    "    spark.read.format(\"csv\")\n",
    "    .option(\"encoding\", \"UTF-8\")\n",
    "    .option(\"header\", True)\n",
    "    .option(\"sep\", \",\")\n",
    "    .option(\"escape\", '\"')\n",
    "    .schema(schema_ratings)\n",
    "    .load(file_path_ratings)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "id": "396aac23-6eea-4f62-9cef-0fe9c5b1ed9c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+------+----------+\n",
      "|userId|movieId|rating|timestamp |\n",
      "+------+-------+------+----------+\n",
      "|1     |296    |5.0   |1147880044|\n",
      "|1     |306    |3.5   |1147868817|\n",
      "|1     |307    |5.0   |1147868828|\n",
      "|1     |665    |5.0   |1147878820|\n",
      "|1     |899    |3.5   |1147868510|\n",
      "|1     |1088   |4.0   |1147868495|\n",
      "|1     |1175   |3.5   |1147868826|\n",
      "|1     |1217   |3.5   |1147878326|\n",
      "|1     |1237   |5.0   |1147868839|\n",
      "|1     |1250   |4.0   |1147868414|\n",
      "+------+-------+------+----------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ratings_raw.show(10, False)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "11d47752-39f9-4e99-a842-8ac712001695",
   "metadata": {},
   "source": [
    "#### Tags"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "c8309cd0-4407-4b30-b1ce-d0205d086dcc",
   "metadata": {},
   "outputs": [],
   "source": [
    "tags_raw = (\n",
    "    spark.read.format(\"csv\")\n",
    "    .option(\"encoding\", \"UTF-8\")\n",
    "    .option(\"header\", True)\n",
    "    .option(\"sep\", \",\")\n",
    "    .option(\"escape\", '\"')\n",
    "    .schema(schema_tags)\n",
    "    .load(file_path_tags)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "e5b77c0e-6c7e-4dba-8332-77d982167f6c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+-----------------------+----------+\n",
      "|userId|movieId|tag                    |timestamp |\n",
      "+------+-------+-----------------------+----------+\n",
      "|3     |260    |classic                |1439472355|\n",
      "|3     |260    |sci-fi                 |1439472256|\n",
      "|4     |1732   |dark comedy            |1573943598|\n",
      "|4     |1732   |great dialogue         |1573943604|\n",
      "|4     |7569   |so bad it's good       |1573943455|\n",
      "|4     |44665  |unreliable narrators   |1573943619|\n",
      "|4     |115569 |tense                  |1573943077|\n",
      "|4     |115713 |artificial intelligence|1573942979|\n",
      "|4     |115713 |philosophical          |1573943033|\n",
      "|4     |115713 |tense                  |1573943042|\n",
      "+------+-------+-----------------------+----------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "tags_raw.show(10, False)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a3a07ac0-cf79-446f-9c25-79462e549282",
   "metadata": {},
   "source": [
    "#### Tag Genome"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "adbb00d2-9d6c-47f2-9f7b-51dffe98794e",
   "metadata": {},
   "outputs": [],
   "source": [
    "genome_tags_raw = (\n",
    "    spark.read.format(\"csv\")\n",
    "    .option(\"encoding\", \"UTF-8\")\n",
    "    .option(\"header\", True)\n",
    "    .option(\"sep\", \",\")\n",
    "    .option(\"escape\", '\"')\n",
    "    .schema(schema_genome_tags)\n",
    "    .load(file_path_genome_tags)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "5c83b814-cfc9-450c-9979-8d46b0f6a3a1",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----+------------+\n",
      "|tagId|tag         |\n",
      "+-----+------------+\n",
      "|1    |007         |\n",
      "|2    |007 (series)|\n",
      "|3    |18th century|\n",
      "|4    |1920s       |\n",
      "|5    |1930s       |\n",
      "|6    |1950s       |\n",
      "|7    |1960s       |\n",
      "|8    |1970s       |\n",
      "|9    |1980s       |\n",
      "|10   |19th century|\n",
      "+-----+------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "genome_tags_raw.show(10, False)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c7797a50-7a36-47c3-aba6-b08842b44246",
   "metadata": {},
   "source": [
    "#### Tag Genome Scores"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "c0e0c3ec-8109-45ed-9100-4d53e1d59486",
   "metadata": {},
   "outputs": [],
   "source": [
    "genome_scores_raw = (\n",
    "    spark.read.format(\"csv\")\n",
    "    .option(\"encoding\", \"UTF-8\")\n",
    "    .option(\"header\", True)\n",
    "    .option(\"sep\", \",\")\n",
    "    .option(\"escape\", '\"')\n",
    "    .schema(schema_genome_scores)\n",
    "    .load(file_path_genome_scores)\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "05208d72-80f8-47c8-b3cd-f3fd6f68d576",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----+---------+\n",
      "|movieId|tagId|relevance|\n",
      "+-------+-----+---------+\n",
      "|1      |1    |0        |\n",
      "|1      |2    |0        |\n",
      "|1      |3    |0        |\n",
      "|1      |4    |0        |\n",
      "|1      |5    |0        |\n",
      "|1      |6    |0        |\n",
      "|1      |7    |0        |\n",
      "|1      |8    |0        |\n",
      "|1      |9    |0        |\n",
      "|1      |10   |0        |\n",
      "+-------+-----+---------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "genome_scores_raw.show(10, False)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1a99576b-590b-4356-898a-cab4b12a6ca1",
   "metadata": {},
   "source": [
    "# Harder questions on `MovieLens` using `PySpark`"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0eae5bd9-da87-40a9-aee8-eecf13c27705",
   "metadata": {},
   "source": [
    "## Question Set\n",
    "\n",
    "For each question below, I provide a 'theme' in brackets - helps one think of the possible approaches and classify the 'kind' of query or analysis we are trying to crack."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7b470fe7-5956-4b72-9c92-9791d321a372",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true
   },
   "source": [
    "### **1.  Transitive Genre Preference Propagation (Graph-like + Optimization)**    \n",
    "    \n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   A user directly rates movies. We define *indirect* genre preference as follows: If User A rates 5 movies of Genre X highly (>=4 stars), and User A also rates 5 movies of Genre Y highly, then all users who rated at least 3 movies of Genre Y highly *also* indirectly prefer Genre X. Propagate this preference transitively.  Output the top 5 *indirectly* preferred genres for each user that are NOT directly preferred.    \n",
    "---\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e568bbd1-cc92-4cc2-acf9-7924fa9c147c",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true
   },
   "source": [
    "### **2.  Time-Decayed Movie Recommendation Similarity (Windowing + Complex-ish Aggregation)**    \n",
    "    \n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   Calculate a time-decayed similarity score between all pairs of movies.  The similarity is based on users who have rated both movies.  For each user who rated both, the contribution to similarity decays exponentially with the time difference between their ratings of the two movies (half-life of 30 days).  Output the top 10 most similar movies for each movie.    \n",
    "---\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3a34973f-7903-4f7b-8f94-6c1822365854",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true
   },
   "source": [
    "### **3.  Genre Co-occurrence Network Analysis (Graph-like + Aggregation)**    \n",
    "    \n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   Construct a weighted, undirected graph where nodes are genres, and an edge exists between two genres if at least 100 users have rated at least 5 movies of *each* genre with an average rating of 4 or higher.  The edge weight is the Jaccard similarity of the user sets who contributed to each genre's qualification. Output the top 3 most \"central\" genres based on degree centrality (weighted).    \n",
    "---\n",
    " "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7ac06397-2ba4-4883-be40-843877d7daf9",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true
   },
   "source": [
    "### **4.  Rating Pattern Anomaly Detection (Windowing + Custom Logic)**    \n",
    "    \n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   Identify users whose rating behavior shows significant *sudden* shifts.  A \"shift\" is defined as a change in the average rating of at least 1.5 stars (up or down) over a rolling 7-day window, compared to the previous 30-day average, occurring in at least 3 distinct weeks within the dataset.    \n",
    "---\n",
    "   "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6d1b212f-07fb-4301-a70a-bf88fbe37264",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true
   },
   "source": [
    "### **5.  Movie Recommendation Cold Start (Complex Filtering + Aggregation)**    \n",
    "    \n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   For each movie with *fewer* than 50 ratings, identify the top 5 \"similar\" movies based on a combination of shared genres (weighted by the number of shared genres) and the average rating difference of users who have rated both (penalize large differences).  Handle the case where no users have rated both.    \n",
    "---\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0b8600b2-117c-45b8-8906-a89a945c146d",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true
   },
   "source": [
    "### **6.  Transitive User Similarity (Graph-like + Optimization)**    \n",
    "    \n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   Calculate a transitive user similarity score.  If User A and User B have rated at least 5 movies in common with an average rating difference of less than 1, they have a direct similarity.  Transitive similarity extends this: If A is similar to B, and B is similar to C, then A and C have a transitive similarity score (decayed by a factor of 0.5 for each hop).  Output the 5 most *transitively* similar users for each user.    \n",
    "---\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "deb2f515-6b8a-455b-8bdb-a5c1ec0b720d",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true
   },
   "source": [
    "### **7.  Time-Based Genre Evolution (Windowing + Complex Aggregation)**    \n",
    "    \n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   Track the popularity of each genre over time (monthly windows).  Define \"popularity\" as the average rating of movies of that genre, weighted by the number of ratings in that month.  Identify genres that show a statistically significant (e.g., using a z-score) increase or decrease in popularity over any 6-month period.    \n",
    "---\n",
    "    \n",
    "      "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "44c5a204-5d03-48c1-8f64-695f6fa01ea8",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true
   },
   "source": [
    "### **8.  User Rating Bias Detection (Aggregation + Statistical Analysis)**    \n",
    "    \n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   Identify users who exhibit a significant rating bias (e.g., consistently rating higher or lower than the average).  Calculate each user's average rating deviation from the movie's average rating, and then identify outliers (e.g., users whose average deviation is more than 2 standard deviations from the mean deviation).    \n",
    "---\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b3d348f8-ac44-4a40-80d9-eba99ecd7cbf",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true
   },
   "source": [
    "### **9.  Movie Recommendation Diversification (Complex Filtering + Optimization)**    \n",
    "    \n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   Given a user's top 10 recommended movies (based on a simple collaborative filtering model - assume this is pre-computed), re-rank these movies to increase genre diversity.  Penalize movies that share genres with already-recommended movies.    \n",
    "---\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "974811f2-836e-42e4-ad3f-f9d635944119",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true
   },
   "source": [
    "### **10. Long-Tail Movie Recommendation (Filtering + Aggregation)**    \n",
    "    \n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   Identify \"long-tail\" movies (movies with few ratings but high average ratings).  For each user, recommend the top 3 long-tail movies that *do not* belong to genres the user has frequently rated.    \n",
    "---\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0cd4ae26-e088-4dc1-a3aa-9ea70de04322",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true
   },
   "source": [
    "### **11.  Movie Release Year Influence (Windowing + Aggregation)**    \n",
    "    \n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   Analyze if the release year of a movie influences its ratings, considering user age.  Group users by age brackets (e.g., 18-25, 26-35, etc.) and movies by release year brackets. Calculate the average rating for each user-age/movie-year combination. Identify statistically significant correlations.    \n",
    "---\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "412e9c4b-4897-4755-9c45-2e2dac35477c",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true
   },
   "source": [
    "### **12.  Genre Combination Preference (Aggregation + Complex Logic)**    \n",
    "    \n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   Identify combinations of *two* genres that, when present together in a movie, result in significantly higher average ratings than movies with only one of those genres.    \n",
    "---\n",
    "    \n",
    "     "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7c9b9c74-1d66-4353-bc59-9627c620d29f",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true
   },
   "source": [
    "### **13.  User Segmentation by Rating Patterns (Clustering - Conceptual, more like a discussion, *ignore this one in the first run*)**    \n",
    "\n",
    "    I put the question here, as there's a separate module I am planning on clustering, MLLib and GraphX, this question whets your appetite.\n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   ***Describe* how you would use PySpark to segment users into distinct clusters based on their rating patterns (e.g., users who mostly give high ratings, users who are very critical, users with bimodal distributions, etc.).**\n",
    "*   NOTE: You don't need to implement a full clustering algorithm, but outline the features you would extract and the PySpark operations you would use.   \n",
    "---\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "99163483-68f5-4662-a6cb-7febeb2897c8",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true
   },
   "source": [
    "### **14.  Movie Similarity via Rating Vector Cosine Similarity (Aggregation + Optimization)**    \n",
    "    \n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   Calculate the cosine similarity between all pairs of movies based on their *user rating vectors*.  Handle cases where users have rated one movie but not the other (impute a default rating or use a smoothing technique).  Output the 10 most similar movies for each movie.    \n",
    "---\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4aa9eb92-d01e-4577-b676-21247b8eb716",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true
   },
   "source": [
    "### **15.  Impact of \"Super-Raters\" (Filtering + Aggregation)**    \n",
    "    \n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   Identify \"super-raters\" (users who have rated a significantly higher number of movies than average).  Analyze the impact of removing their ratings on the overall average rating of movies.  Are there specific genres or movies that are disproportionately affected?    \n",
    "---\n",
    "    \n",
    "       "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5715b40c-89bf-4bb2-ac90-77bc3ea87744",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true
   },
   "source": [
    "### **16.  Predictive Rating Decay (Windowing + Regression - Conceptual, ummm...)**    \n",
    "    ...again just embedding thought experiments here, these would be too large or costly to build as a part of this assessment\n",
    "    \n",
    "---  \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   *Describe* how you would use PySpark to build a model that predicts how a movie's average rating will change over time (e.g., will it increase, decrease, or remain stable?). Consider factors like initial ratings, genre, and the ratings of similar movies.  Outline the features and PySpark operations.    \n",
    "---\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1108c1c3-24e4-471c-a4db-66144c41439a",
   "metadata": {
    "jp-MarkdownHeadingCollapsed": true
   },
   "source": [
    "### **17.  Transitive Genre Dislike Propagation (Graph-like + Optimization, Negative Preference)**    \n",
    "    \n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   Similar to Question 1, but propagate *dislike*.  If a user consistently rates movies of Genre X poorly (<=2 stars), and they also rate movies of Genre Y poorly, then users who dislike Genre Y also indirectly dislike Genre X.  Output the top 5 indirectly *disliked* genres for each user.    \n",
    "---\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3790c9dc-6c49-443e-92fe-4796cfe53265",
   "metadata": {},
   "source": [
    "### **18.  Genre Influence Over Time (Windowing + Correlation)**    \n",
    "    \n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*    Investigate the changing influence of genres on each other over time.  For each pair of genres, calculate the correlation between their average monthly ratings over a rolling 12-month window.  Identify genre pairs with increasingly strong positive or negative correlations.    \n",
    "---\n",
    "    \n",
    "        "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8c917a63-8dd8-4e37-9556-b7c9a6253864",
   "metadata": {},
   "source": [
    "### **19.  User Rating Consistency Across Genres (Aggregation + Statistical Analysis)**    \n",
    "    \n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   Analyze whether users maintain consistent rating behavior across different genres.  For each user and genre, calculate the standard deviation of their ratings.  Identify users whose rating consistency varies significantly across genres.    \n",
    "---\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "86f03ab3-7060-4151-8370-d31343419b1a",
   "metadata": {},
   "source": [
    "### **20.  Movie Recommendation \"Serendipity\" (Complex Filtering + Optimization)**    \n",
    "    \n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   Design a metric to measure the \"serendipity\" of a movie recommendation.  A serendipitous recommendation is one that is both relevant (high predicted rating) and *unexpected* (not belonging to the user's frequently rated genres or similar to movies they've already seen).  Implement this metric.    \n",
    "---\n",
    "    \n",
    "        "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "789de971-4b4e-4a5a-9603-89a1c3021dd8",
   "metadata": {},
   "source": [
    "### **21.  Temporal Rating Bias (Windowing + Statistical Analysis)**    \n",
    "    \n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   Investigate if there's a temporal bias in user ratings (e.g., do users tend to rate higher or lower on weekends vs. weekdays, or during certain times of the year?).    \n",
    "---\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "749356b7-0254-4c94-b564-9bf615902429",
   "metadata": {},
   "source": [
    "### **22.  Influence of Early Ratings (Windowing + Aggregation)**    \n",
    "    \n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   Analyze the influence of a movie's *early* ratings (e.g., the first 100 ratings) on its long-term average rating.  Do movies with high initial ratings tend to maintain high ratings, or is there a regression to the mean?    \n",
    "---\n",
    "    \n",
    "        "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e4479be6-9c41-4f20-bc63-5d49a2302ce2",
   "metadata": {},
   "source": [
    "### **23.  Genre Hybridity and Rating (Complex Filtering + Aggregation)**    \n",
    "    \n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   Define a measure of \"genre hybridity\" for a movie (e.g., the number of distinct genres it belongs to, or a more sophisticated measure based on genre co-occurrence).  Investigate the relationship between genre hybridity and average rating.    \n",
    "---\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cdeec994-4d2b-4d16-bdb2-3b491aa50453",
   "metadata": {},
   "source": [
    "### **24.  User \"Criticality\" Evolution (Windowing + Aggregation)**    \n",
    "    \n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   Track how a user's \"criticality\" changes over time.  Define \"criticality\" as the difference between a user's average rating and the average rating of all users for the movies they've rated.  Analyze if users tend to become more or less critical over time.    \n",
    "---\n",
    "    \n",
    "        "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6ef4b0fa-a9fc-4e05-8e3d-ce6fe3e214dc",
   "metadata": {},
   "source": [
    "### **25.  Network Effects in Ratings (Graph-like + Optimization)**    \n",
    "    \n",
    "---    \n",
    "    \n",
    "#### Question    \n",
    "       \n",
    "        \n",
    "*   Construct a user-movie bipartite graph.  Investigate if there are \"network effects\" in ratings – that is, if a user is connected to many users who have rated a movie highly, are they more likely to rate that movie highly as well, even after controlling for the movie's overall average rating?    \n",
    "---\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4952bbf6-60ec-4e65-be7f-1565e18132de",
   "metadata": {},
   "source": [
    "# Clear cache and stop the spark cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e897743f-de29-4c8e-b94d-ce7d0c57a967",
   "metadata": {},
   "outputs": [],
   "source": [
    "# clear cache\n",
    "spark.catalog.clearCache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bfa2c361-2e31-4f21-8941-7b725da366c5",
   "metadata": {},
   "outputs": [],
   "source": [
    "# stop spark\n",
    "spark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3a699f3c-affb-4ae5-9455-23183fdf6077",
   "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.11.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}