03-Spark/003.02-Harder-Problems-in-PySpark(Questions-with-Hints).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, *with Hints***"
   ]
  },
  {
   "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.02-Harder-Problems-in-PySpark(Questions-with-Hints).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 problems on `MovieLens` using `PySpark`\n",
    "  \n",
    "Look, if we've done the previous notebooks (002.xx) many of these would feel a continuation of the same theme or a repetition even.  \n",
    "In many cases these should really not feel \"hard\".   \n",
    "  \n",
    "I have put these together to say: \"here's all the ones that appear hard, also some patterns that I have seen across them\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b2ebc5d7-3faf-4133-a42d-58467d4a40dc",
   "metadata": {},
   "source": [
    "## **Patterns** I have frequently seen in some of the \"harder\"-ish-real-world-adjacent problems"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "32693e9d-6497-4728-a5f4-637698da4b8a",
   "metadata": {},
   "source": [
    "### **A. Figure out a Sequence or do some Pattern Detection** \n",
    "analyzing sequences of events/facts, not just individual events or simple aggregations.   \n",
    "This demands more sophisticated use of window functions, date/time manipulations, and potentially iterative processing. Iterative nearly always blows up in production. :)  "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "06704fb1-e4ce-4844-bf89-04933812afb9",
   "metadata": {},
   "source": [
    "### **B. Implicit Graph-Like Relationships** \n",
    "There are situations where you are dealing with an implicit Graph-Like structure (hierarchies, relationships across entities, business rules, reachability etc. etc.).   \n",
    "This is where you've got to start thinking about how to represent and traverse relationships that are not explicitly defined as foreign keys.  "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "894f7c4f-8ee6-47f3-bf60-2fb250cf287b",
   "metadata": {},
   "source": [
    "### **C. Complex Windowing and Grouping** \n",
    "combinations of window functions, aggregations, and filtering...  \n",
    "just the perfect recipe for a drawn-out struggle with your spark jobs, late night on a Saturday as your family slowly forgets what you look like and your existence hurtles towards decrepitude. Just chef's kiss.   \n",
    "The fix? A deep (deeper?) understanding of how these operations interact, review early, warn often."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ec6b994e-04f0-4f3d-a511-a6c2d9ed0527",
   "metadata": {},
   "source": [
    "### **D. Transitive Closure** \n",
    "Fancy word, (just to prove I am not a philistine) but really means that if A is connected to B and B is connected to C, then A and C may have a connection.   \n",
    "This sort of stuff is useful for in all kinds of queries, esp. in a business situation, when it's like 5PM on a Friday and you have plans for the weekend.   \n",
    "No seriously, T.C. adds a significant layer of complexity that is difficult to handle efficiently in a distributed environment like Spark.    \n",
    "Think of transitive closure like **friend-of-a-friend** situation on a social network.    \n",
    "    \n",
    "- Suppose **Alice knows Bob** and **Bob knows Charlie**.    \n",
    "- Even though Alice and Charlie don’t directly know each other, we can *infer* that Alice can be connected to Charlie through Bob.    \n",
    "    \n",
    "In a **transitive closure**, we keep expanding these connections until we capture all possible indirect relationships, or your spark cluster has burned down (like obvs...) or Chapter 11 rumours start going around due to all the cloud costs.     \n",
    "Here's some common scenarios:     \n",
    "- **Flights between cities**: If you can fly from **New York → London** and from **London → Paris**, then logically, you can reach **New York → Paris** via London.    \n",
    "- **Supply Chains**: If a **raw material supplier** delivers to **Factory A**, and Factory A supplies to **Warehouse B**, then the raw material can reach **Warehouse B** indirectly.    \n",
    "     \n",
    "Or in the case of our `MovieLens` dataset context, let’s say two users have *similar taste* if they rate the same movies (that same friend-of-a-friend idea).     \n",
    "- **User A rates Movie X**    \n",
    "- **User B also rates Movie X** → So User A and B are connected.    \n",
    "- **User B rates Movie Y** → This means User A could also be connected to Movie Y (indirectly).    \n",
    "       \n",
    "Use **transitive closure** to expand these connections, so we can **recommend movies based on indirect user interactions**.    \n",
    "Other interesting queries come to mind:   \n",
    "- **Find Hidden Similar Users**: If User A indirectly connects to User C via B, see we can recommend movies watched by C to A.     \n",
    "- **Detect User Communities**: find groups of users with similar movie preferences based on extended relationships.     "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "aa2bab94-b3bc-4c6c-b6e2-9285da463a11",
   "metadata": {},
   "source": [
    "### **E. Non-Standard Aggregations**  \n",
    "Stuff like \"average path length\" and \"longest consecutive sequence\" type calculations are not standard aggregations.   \n",
    "They require custom logic within the Spark framework and hair, lots of hair on your head, so you can pull some out while solving for these...    "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "40ced220-bfa7-4dcc-9080-d950fe5c0cc2",
   "metadata": {},
   "source": [
    "### **Optimization Techniques** I have often used in these cases (esp. our friend the Transitive Closure)\n",
    "* Use **broadcast joins** for small subsets - Instead of costly shuffles, we **broadcast** one side of the join when the dataset is small.  \n",
    "* Use **temporary tables** for better query performance - yeah right (snigger)    \n",
    "* Leverage **SQL recursion** type approach for efficiency - this is iffy, use as a last resort, like all your bullets are spent, your knife blade is rusty, there's a pack of lions at the door, your mushrooms just kicked in and there's not much else you can do. "
   ]
  },
  {
   "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.\n",
    "*   We define *indirect* genre preference as follows:\n",
    "    *   If User A rates 5 movies of Genre X highly (>=4 stars),\n",
    "    *   and User A also rates 5 movies of Genre Y highly,\n",
    "    *   then all users who rated at least 3 movies of Genre Y highly *also* indirectly prefer Genre X.     \n",
    ".          \n",
    "*   Propagate this preference transitively:\n",
    "    *   **Output the top 5 *indirectly* preferred genres for each user that are NOT directly preferred.**    \n",
    "---\n",
    "    \n",
    "        \n",
    "*   <font color='lightgray'> <b>Why this could be interesting:</b>  Combines transitive closure (graph-like propagation) with filtering based on direct preferences.  Requires iterative processing and careful handling of already-propagated preferences to avoid infinite loops.  The \"indirect, not direct\" condition is crucial and easily missed.</font>    \n",
    "*   <font color='lightgray'> <b>Optimization Hint:</b>  Consider using a combination of iterative DataFrame joins and filtering, with careful use of caching and broadcasting smaller DataFrames (like direct genre preferences per user) to minimize shuffle operations.  Explore accumulator usage for tracking convergence. Explore converting to RDDs for finer-grained control over partitioning if DataFrames become a bottleneck.</font>         \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.\n",
    "*   The similarity is based on users who have rated both movies.\n",
    "*   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).\n",
    "*   **Output the top 10 most similar movies for each movie.**    \n",
    "---\n",
    "    \n",
    "        \n",
    "*   <font color='lightgray'> <b>Why this could be interesting:</b> Requires joining `ratings` with itself, calculating time differences, applying exponential decay, and then aggregating per movie pair.  The time-decay calculation and aggregation within a window are complex.</font>    \n",
    "*   <font color='lightgray'> <b>Optimization Hint:</b>  Pre-calculate the exponential decay factor based on time difference buckets (e.g., 0-1 day, 1-2 days, etc.) to avoid redundant calculations.  Consider using a broadcast join if one of the movie DataFrames is significantly smaller.  Window functions might seem like a fit, but custom aggregation is likely necessary.</font>         \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.**\n",
    "*   The edge weight is the Jaccard similarity of the user sets who contributed to each genre's qualification.\n",
    "*   **Output the top 3 most \"central\" genres based on degree centrality (weighted).**      \n",
    "---\n",
    "    \n",
    "        \n",
    "*   <font color='lightgray'> <b>Why this could be interesting:</b>  Requires multiple joins and aggregations to determine qualifying genres and users.  Calculating Jaccard similarity efficiently is a challenge.  The concept of \"centrality\" adds a graph analysis layer.</font>    \n",
    "*   <font color='lightgray'> <b>Optimization Hint:</b>  Use broadcast joins to distribute smaller DataFrames (like user sets per genre).  Consider pre-computing user-genre affinities to speed up subsequent calculations. Explore efficient set intersection and union operations (e.g., using bitsets within a UDF if necessary).</font>         \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.**\n",
    "*   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",
    "    \n",
    "        \n",
    "*   <font color='lightgray'> <b>Why this could be interesting:</b>  Combines rolling window calculations with comparison to a longer-term average.  The \"3 distinct weeks\" condition adds complexity to avoid false positives.</font>    \n",
    "*   <font color='lightgray'> <b>Optimization Hint:</b>  Use window functions with custom aggregations to calculate both the 7-day and 30-day averages.  Efficient date handling and filtering are crucial. Caching intermediate results is vital.</font>         \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, \n",
    "    *   **identify the top 5 \"similar\" movies based on a combination of shared genres** (weighted by the number of shared genres) and\n",
    "    *   **the average rating difference of users who have rated both** (penalize large differences).\n",
    "*   Handle the case where no users have rated both.    \n",
    "---\n",
    "    \n",
    "        \n",
    "*   <font color='lightgray'> <b>Why this could be interesting:</b> Addresses the cold-start problem, requiring careful handling of missing data and a combination of genre-based and rating-based similarity.</font>    \n",
    "*   <font color='lightgray'> <b>Optimization Hint:</b>  Pre-calculate genre overlaps between movies.  Use coalesce and default values to handle cases with no common raters.  Broadcast join the smaller DataFrame (movies with < 50 ratings).</font>         \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.**\n",
    "*   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.\n",
    "*   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).\n",
    "*   **Output the 5 most *transitively* similar users for each user.**    \n",
    "---\n",
    "    \n",
    "        \n",
    "*   <font color='lightgray'> <b>Why this could be interesting:</b>  Combines transitive closure with a decay factor.  Requires iterative calculation and careful management of similarity scores across hops.</font>    \n",
    "*   <font color='lightgray'> <b>Optimization Hint:</b>  Consider an iterative approach using joins and aggregations, caching intermediate similarity DataFrames.  Use a convergence threshold to stop the iteration. Explore graph libraries (like GraphFrames) if performance is critical, but be aware of the overhead.</font>         \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).**\n",
    "*   Define \"popularity\" as the average rating of movies of that genre, weighted by the number of ratings in that month.\n",
    "*   **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",
    "        \n",
    "*   <font color='lightgray'> <b>Why this could be interesting:</b>  Requires windowing by month, calculating weighted averages, and then applying a statistical test across a longer window.</font>    \n",
    "*   <font color='lightgray'> <b>Optimization Hint:</b>  Pre-calculate monthly aggregations (total ratings, sum of ratings) for each genre.  Use window functions to calculate the z-score efficiently.</font>         \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).**\n",
    "*   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",
    "    \n",
    "        \n",
    "*   <font color='lightgray'> <b>Why this could be interesting:</b>  Requires calculating per-movie and per-user averages, then combining them to find deviations.  Outlier detection adds a statistical layer.</font>    \n",
    "*   <font color='lightgray'> <b>Optimization Hint:</b>  Use broadcast joins to efficiently combine movie and user statistics.  Calculate the mean and standard deviation of the deviations in a separate step for efficiency.</font>         \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.**\n",
    "*   Penalize movies that share genres with already-recommended movies.    \n",
    "---\n",
    "    \n",
    "        \n",
    "*   <font color='lightgray'> <b>Why this could be interesting:</b>  Requires reasoning about genre overlap and applying a penalty that discourages redundancy.</font>    \n",
    "*   <font color='lightgray'> <b>Optimization Hint:</b>  Pre-calculate genre sets for each movie.  Use a custom ranking function (UDF) that iteratively selects movies, penalizing based on genre overlap with previously selected movies.</font>         \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).**\n",
    "*   **For each user, recommend the top 3 long-tail movies that *do not* belong to genres the user has frequently rated.**    \n",
    "---\n",
    "    \n",
    "        \n",
    "*   <font color='lightgray'> <b>Why this could be interesting:</b>  Combines filtering based on rating count and average rating with a negative genre preference filter.</font>    \n",
    "*   <font color='lightgray'> <b>Optimization Hint:</b>  Pre-calculate user genre preferences.  Use efficient filtering and joining techniques to avoid unnecessary computations.</font>         \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.**\n",
    "*   **Group users by age brackets (e.g., 18-25, 26-35, etc.) and movies by release year brackets.**\n",
    "*   **Calculate the average rating for each user-age/movie-year combination.**\n",
    "*   **Identify statistically significant correlations.**    \n",
    "---\n",
    "    \n",
    "        \n",
    "*   <font color='lightgray'> <b>Why this could be interesting:</b> Requires grouping by two different time-based attributes and calculating correlations.</font>         \n",
    "* <font color='lightgray'> <b>Optimization Hint:</b>  Pre-calculate age and release year brackets.  Use efficient grouping and aggregation operations.</font>         \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",
    "        \n",
    "*   <font color='lightgray'> <b>Why this could be interesting:</b>  Requires parsing the pipe-separated genres string, generating genre combinations, and comparing average ratings across different groups.</font>    \n",
    "*   <font color='lightgray'> <b>Optimization Hint:</b>  Use UDFs to efficiently parse genres and generate combinations.  Pre-calculate average ratings for single-genre movies.</font>         \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",
    "    \n",
    "        \n",
    "*   <font color='lightgray'> <b>Why this could be interesting:</b> Requires understanding of clustering concepts and how to translate them into PySpark feature engineering and (potentially) MLlib usage.</font>    \n",
    "*   <font color='lightgray'> <b>Optimization Hint:</b>  Consider features like average rating, rating variance, skewness, kurtosis, and percentile distributions.  Explain how you would use window functions, aggregations, and potentially UDFs to extract these features.</font>         \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*.**\n",
    "*   Handle cases where users have rated one movie but not the other (impute a default rating or use a smoothing technique).\n",
    "*   **Output the 10 most similar movies for each movie.**    \n",
    "---\n",
    "    \n",
    "                \n",
    "*    <font color='lightgray'> <b>Why this could be interesting:</b> High memory usage to create rating vectors, dealing with sparsity.</font>         \n",
    "*    <font color='lightgray'> <b>Optimization Hint:</b> Instead of creating large vectors, calculate dot products and norms directly through joins and aggregations to save substantial memory.</font>         \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).**\n",
    "*   Analyze the impact of removing their ratings on the overall average rating of movies.\n",
    "*   **Are there specific genres or movies that are disproportionately affected?**\n",
    "    \n",
    "---\n",
    "    \n",
    "        \n",
    "*   <font color='lightgray'> <b>Why this could be interesting:</b> Requires identifying outliers in rating counts and then comparing aggregations with and without these users.</font>    \n",
    "*   <font color='lightgray'> <b>Optimization Hint:</b> Use efficient filtering and aggregation techniques.  Calculate overall averages before and after removing super-raters.</font>         \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?).**\n",
    "*   Consider factors like initial ratings, genre, and the ratings of similar movies.\n",
    "*   **Outline the features and PySpark operations.**    \n",
    "---\n",
    "    \n",
    "                \n",
    "*    <font color='lightgray'> <b>Why this could be interesting:</b> Needs time-series analysis, feature engineering with genre, and similarity.</font>         \n",
    "*    <font color='lightgray'> <b>Optimization Hint:</b> Create time-series features (average rating in different time windows).  Use genre and similarity (from previous questions) as features.  Explain how you would use window functions and potentially MLlib regression models.</font>         \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*.**\n",
    "*   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.\n",
    "*   **Output the top 5 indirectly *disliked* genres for each user.**    \n",
    "---\n",
    "    \n",
    "        \n",
    "*   <font color='lightgray'> <b>Why this could be interesting:</b>  Mirrors the complexity of Question 1, but with the added nuance of handling negative preferences, which can be less intuitive.</font>    \n",
    "*   <font color='lightgray'> <b>Optimization Hint:</b> Same optimization strategies as Question 1 apply, but carefully consider how to represent and propagate \"dislike\" efficiently.</font>         \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.\n",
    "*    For each pair of genres, calculate the correlation between their average monthly ratings over a rolling 12-month window.\n",
    "*    **Identify genre pairs with increasingly strong positive or negative correlations.**    \n",
    "---\n",
    "    \n",
    "                \n",
    "*    <font color='lightgray'> <b>Why this could be interesting:</b>  Requires nested windowing (monthly ratings within a 12-month rolling window) and correlation calculation.</font>         \n",
    "*    <font color='lightgray'> <b>Optimization Hint:</b> Pre-calculate monthly average ratings for each genre. Use window functions to calculate the rolling 12-month correlation.</font>        \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.\n",
    "*   For each user and genre, calculate the standard deviation of their ratings.\n",
    "*   **Identify users whose rating consistency varies significantly across genres.**    \n",
    "---\n",
    "    \n",
    "        \n",
    "*   <font color='lightgray'> <b>Why this could be interesting:</b> Requires calculating per-user, per-genre statistics and then analyzing the variance of these statistics.</font>    \n",
    "*   <font color='lightgray'> <b>Optimization Hint:</b> Use efficient grouping and aggregation operations. Calculate the standard deviation of standard deviations.</font>        \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**.\n",
    "*   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).\n",
    "*   **Implement** this metric.    \n",
    "---\n",
    "    \n",
    "        \n",
    "*   <font color='lightgray'> <b>Why this could be interesting:</b>  Requires combining relevance (predicted rating) with a measure of novelty or unexpectedness.</font>    \n",
    "*   <font color='lightgray'> <b>Optimization Hint:</b>  Use a combination of predicted ratings (from a pre-computed model), genre dissimilarity, and similarity to previously seen movies.</font>        \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",
    "    \n",
    "        \n",
    "*   <font color='lightgray'> <b>Why this could be interesting:</b> Requires extracting time-based features (day of week, month, etc.) and analyzing rating distributions across these features.</font>    \n",
    "*   <font color='lightgray'> <b>Optimization Hint:</b> Use PySpark's date/time functions to extract relevant features.  Compare rating distributions using statistical tests.</font>        \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.\n",
    "*   **Do movies with high initial ratings tend to maintain high ratings, or is there a regression to the mean?**    \n",
    "---\n",
    "    \n",
    "        \n",
    "        \n",
    "*    <font color='lightgray'> <b>Why this could be interesting:</b> Extracting first n ratings, calculate the moving averages, calculate correlations and handle edge cases.</font>        \n",
    "*    <font color='lightgray'> <b>Optimization Hint:</b> Use window functions with a specific frame (first N rows) to calculate the early average rating.  Correlate this with the overall average rating.</font>        \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).\n",
    "*   **Investigate the relationship between genre hybridity and average rating.**    \n",
    "---\n",
    "    \n",
    "        \n",
    "*   <font color='lightgray'> <b>Why this could be interesting:</b>  Requires defining and calculating a complex metric based on the genre string, then correlating it with ratings.</font>    \n",
    "*   <font color='lightgray'> <b>Optimization Hint:</b> Use UDFs to calculate genre hybridity.  Consider different weighting schemes for genres.</font>        \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.\n",
    "*   Define \"criticality\" as the difference between a user's average rating and the average rating of all users for the movies they've rated.\n",
    "*   **Analyze if users tend to become more or less critical over time.**    \n",
    "---\n",
    "    \n",
    "        \n",
    "        \n",
    "*    <font color='lightgray'> <b>Why this could be interesting:</b> Calculating user criticality relative to the average user.        \n",
    "*    <font color='lightgray'> <b>Optimization Hint:</b> Use window functions to track a user's average rating and the overall average rating over time. Calculate the difference and analyze trends.</font>        \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.\n",
    "*   **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",
    "    \n",
    "        \n",
    "*   <font color='lightgray'> <b>Why this could be interesting:</b>  Requires constructing a bipartite graph representation and analyzing the influence of neighbors' ratings.</font>    \n",
    "*   <font color='lightgray'> <b>Optimization Hint:</b>  Represent the data as an adjacency list (user -> [movies rated]) and use joins to find neighbors' ratings.  Calculate conditional probabilities to assess network effects. Use broadcasting of smaller graphs.</font>        "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4952bbf6-60ec-4e65-be7f-1565e18132de",
   "metadata": {},
   "source": [
    "# Clear cache and stop the spark cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "e897743f-de29-4c8e-b94d-ce7d0c57a967",
   "metadata": {},
   "outputs": [],
   "source": [
    "# clear cache\n",
    "spark.catalog.clearCache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "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
}