{ "cells": [ { "cell_type": "code", "execution_count": 1, "id": "e87ee965-719c-4163-9f2e-a6235818f562", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Collecting graphframes\n", " Downloading graphframes-0.6-py2.py3-none-any.whl.metadata (934 bytes)\n", "Requirement already satisfied: numpy in ./anaconda3/lib/python3.12/site-packages (from graphframes) (1.26.4)\n", "Collecting nose (from graphframes)\n", " Downloading nose-1.3.7-py3-none-any.whl.metadata (1.7 kB)\n", "Downloading graphframes-0.6-py2.py3-none-any.whl (18 kB)\n", "Downloading nose-1.3.7-py3-none-any.whl (154 kB)\n", "Installing collected packages: nose, graphframes\n", "Successfully installed graphframes-0.6 nose-1.3.7\n", "Note: you may need to restart the kernel to use updated packages.\n" ] } ], "source": [ "pip install graphframes " ] }, { "cell_type": "code", "execution_count": 1, "id": "b03e30ab-a455-4ecd-8975-234a5459beb9", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ ":: loading settings :: url = jar:file:/shared/spark-3.4.4-bin-hadoop3/jars/ivy-2.5.1.jar!/org/apache/ivy/core/settings/ivysettings.xml\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ "Ivy Default Cache set to: /home/exouser/.ivy2/cache\n", "The jars for the packages stored in: /home/exouser/.ivy2/jars\n", "graphframes#graphframes added as a dependency\n", ":: resolving dependencies :: org.apache.spark#spark-submit-parent-ff4d2751-eecb-47ca-9cfb-20b888ea92c6;1.0\n", "\tconfs: [default]\n", "\tfound graphframes#graphframes;0.8.3-spark3.4-s_2.12 in spark-packages\n", "\tfound org.slf4j#slf4j-api;1.7.16 in central\n", ":: resolution report :: resolve 113ms :: artifacts dl 4ms\n", "\t:: modules in use:\n", "\tgraphframes#graphframes;0.8.3-spark3.4-s_2.12 from spark-packages in [default]\n", "\torg.slf4j#slf4j-api;1.7.16 from central in [default]\n", "\t---------------------------------------------------------------------\n", "\t| | modules || artifacts |\n", "\t| conf | number| search|dwnlded|evicted|| number|dwnlded|\n", "\t---------------------------------------------------------------------\n", "\t| default | 2 | 0 | 0 | 0 || 2 | 0 |\n", "\t---------------------------------------------------------------------\n", ":: retrieving :: org.apache.spark#spark-submit-parent-ff4d2751-eecb-47ca-9cfb-20b888ea92c6\n", "\tconfs: [default]\n", "\t0 artifacts copied, 2 already retrieved (0kB/4ms)\n", "Setting default log level to \"WARN\".\n", "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n", "26/02/18 04:33:06 WARN SparkConf: Note that spark.local.dir will be overridden by the value set by the cluster manager (via SPARK_LOCAL_DIRS in mesos/standalone/kubernetes and LOCAL_DIRS in YARN).\n" ] } ], "source": [ "from pyspark.sql import SparkSession\n", "\n", "# HALF of your cluster cores: 224 (7 workers * 64 cores = 448 total)\n", "cores_per_app = \"224\"\n", "\n", "spark = (\n", " SparkSession.builder\n", " .master(\"spark://master:7077\")\n", " .appName(\"movielens\")\n", " .config(\"spark.cores.max\", cores_per_app) # cap total cores for this app\n", " .config(\"spark.executor.instances\", \"56\") # 56 executors\n", " .config(\"spark.executor.cores\", \"4\") # 4 cores per executor -> 56*4 = 224\n", " .config(\"spark.executor.memory\", \"6g\") # per-executor memory\n", " .config(\"spark.driver.memory\", \"6g\")\n", " .config(\"spark.sql.shuffle.partitions\", \"200\")\n", " .getOrCreate()\n", ")" ] }, { "cell_type": "code", "execution_count": 2, "id": "aecec18e-9a8e-4f48-ae87-bad1f1725169", "metadata": {}, "outputs": [], "source": [ "base_hdfs = \"hdfs:///movielens_genome\"\n", "\n", "meta_path = f\"{base_hdfs}/metadata.json\"\n", "ratings_path = f\"{base_hdfs}/ratings.json\"\n", "tags_path = f\"{base_hdfs}/tags.json\"\n", "tagcount_path = f\"{base_hdfs}/tag_count.json\"\n", "reviews_path = f\"{base_hdfs}/review_with_id.json\" \n", "surveys_path = f\"{base_hdfs}/surveys_id.json\" \n", "actor_edges_p = f\"{base_hdfs}/actor_movie_edge.json\"\n", "actor_vertices_p= f\"{base_hdfs}/actor_vertices.json\"\n", "director_edges_p= f\"{base_hdfs}/director_movie_edges.json\"\n", "director_vertices_p = f\"{base_hdfs}/director_vertices.json\"\n" ] }, { "cell_type": "code", "execution_count": 3, "id": "3de7b3f3-ef1f-4c7e-9d9d-7c2805acf085", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "df_meta = spark.read.json(meta_path)\n", "df_ratings = spark.read.json(ratings_path)\n", "df_tags = spark.read.json(tags_path)\n", "df_tagcount= spark.read.json(tagcount_path)\n", "df_reviews = spark.read.json(reviews_path)\n", "df_surveys = spark.read.json(surveys_path)\n", "\n", "df_actor_vertices = spark.read.json(actor_vertices_p)\n", "df_actor_edges = spark.read.json(actor_edges_p)\n", "df_director_vertices = spark.read.json(director_vertices_p)\n", "df_director_edges = spark.read.json(director_edges_p)\n" ] }, { "cell_type": "markdown", "id": "1aff120d-dc9d-4a33-83ff-fea729f6d9bc", "metadata": {}, "source": [ "# Vertices" ] }, { "cell_type": "code", "execution_count": 4, "id": "cdc9740f-5144-4497-a4fe-679bd339528b", "metadata": {}, "outputs": [], "source": [ "movie_vertices = df_meta.selectExpr(\n", " \"item_id as id\",\n", " \"'movie' as type\",\n", " \"title\",\n", " \"avgRating\",\n", " \"imdbId\",\n", " \"directedBy\",\n", " \"starring\"\n", ")" ] }, { "cell_type": "code", "execution_count": 5, "id": "b9c4d64b-684d-4961-9905-673a56e0be6e", "metadata": {}, "outputs": [], "source": [ "user_vertices = df_ratings.selectExpr(\n", " \"user_id as id\",\n", " \"'user' as type\"\n", ").distinct()\n" ] }, { "cell_type": "code", "execution_count": 6, "id": "aee6a39d-59bd-4769-948e-bf58acc91efa", "metadata": {}, "outputs": [], "source": [ "tag_vertices = df_tags.selectExpr(\n", " \"id as id\",\n", " \"'tag' as type\",\n", " \"tag\"\n", ")\n" ] }, { "cell_type": "code", "execution_count": 7, "id": "38cc9d52-ee1f-4c4a-a327-d67f206c2bcf", "metadata": {}, "outputs": [], "source": [ "review_vertices = df_reviews.selectExpr(\n", " \"review_id as id\",\n", " \"'review' as type\",\n", " \"txt as text\",\n", " \"item_id\"\n", ")\n" ] }, { "cell_type": "code", "execution_count": 8, "id": "e3ebac77-9bcd-4a21-a00f-a074d78d0559", "metadata": {}, "outputs": [], "source": [ "survey_vertices = df_surveys.selectExpr(\n", " \"survey_id as id\",\n", " \"'survey' as type\",\n", " \"score\",\n", " \"user_id\",\n", " \"item_id\",\n", " \"tag_id\"\n", ")" ] }, { "cell_type": "code", "execution_count": 9, "id": "f5399ef9-65c9-4504-986e-b62e4feef0e3", "metadata": {}, "outputs": [], "source": [ "director_vertices = df_director_vertices.selectExpr(\n", " \"id as id\",\n", " \"'director' as type\",\n", " \"id as name\"\n", ")\n" ] }, { "cell_type": "code", "execution_count": 10, "id": "8305110a-7ae2-4e7f-bf42-cfb4d042999e", "metadata": {}, "outputs": [], "source": [ "actor_vertices = df_actor_vertices.selectExpr(\n", " \"id as id\",\n", " \"'actor' as type\",\n", " \"id as name\"\n", ")\n" ] }, { "cell_type": "markdown", "id": "d5d71721-6705-43dd-943d-9d71c06c4696", "metadata": {}, "source": [ "# Edges" ] }, { "cell_type": "code", "execution_count": 11, "id": "7073fe3e-5585-418f-9eef-351a90bd2045", "metadata": {}, "outputs": [], "source": [ "# RATED: user -> movie\n", "rated_edges = df_ratings.selectExpr(\n", " \"user_id as src\",\n", " \"item_id as dst\",\n", " \"rating\",\n", " \"'rated' as etype\"\n", ").dropDuplicates()\n", "\n", "# HAS_TAG: movie -> tag\n", "has_tag_edges = df_tagcount.selectExpr(\n", " \"item_id as src\",\n", " \"tag_id as dst\",\n", " \"num as count\",\n", " \"'has_tag' as etype\"\n", ").dropDuplicates()\n", "\n", "# HAS_REVIEW: review -> movie\n", "has_review_edges = df_reviews.selectExpr(\n", " \"review_id as src\",\n", " \"item_id as dst\",\n", " \"'has_review' as etype\"\n", ").dropDuplicates()\n", "\n", "# ANSWERED: user -> survey\n", "answered_edges = df_surveys.selectExpr(\n", " \"user_id as src\",\n", " \"survey_id as dst\",\n", " \"'answered' as etype\"\n", ").dropDuplicates()\n", "\n", "# ABOUT_MOVIE: survey -> movie\n", "about_movie_edges = df_surveys.selectExpr(\n", " \"survey_id as src\",\n", " \"item_id as dst\",\n", " \"'about_movie' as etype\"\n", ").dropDuplicates()\n", "\n", "# ABOUT_TAG: survey -> tag\n", "about_tag_edges = df_surveys.selectExpr(\n", " \"survey_id as src\",\n", " \"tag_id as dst\",\n", " \"'about_tag' as etype\"\n", ").dropDuplicates()\n", "\n", "# DIRECTED: director (name) -> movie\n", "directed_edges = df_director_edges.selectExpr(\n", " \"src as src\", # director name (matches director vertex id)\n", " \"dst as dst\", # movie item_id (matches movie vertex id)\n", " \"'directed' as etype\"\n", ").dropDuplicates()\n", "\n", "# ACTED_IN: actor (name) -> movie\n", "acted_in_edges = df_actor_edges.selectExpr(\n", " \"src as src\", # actor name (matches actor vertex id)\n", " \"dst as dst\", # movie item_id (matches movie vertex id)\n", " \"'acted_in' as etype\"\n", ").dropDuplicates()\n" ] }, { "cell_type": "code", "execution_count": 12, "id": "f4d07d86-3ec3-4912-8592-74de1d819237", "metadata": {}, "outputs": [], "source": [ "# make sure each per-type DF has \"id\" and \"type\" columns (you already created these)\n", "vertices = movie_vertices \\\n", " .unionByName(user_vertices, allowMissingColumns=True) \\\n", " .unionByName(tag_vertices, allowMissingColumns=True) \\\n", " .unionByName(review_vertices, allowMissingColumns=True) \\\n", " .unionByName(survey_vertices, allowMissingColumns=True) \\\n", " .unionByName(actor_vertices, allowMissingColumns=True) \\\n", " .unionByName(director_vertices,allowMissingColumns=True)\n", "\n", "edges = rated_edges \\\n", " .unionByName(has_tag_edges, allowMissingColumns=True) \\\n", " .unionByName(has_review_edges, allowMissingColumns=True) \\\n", " .unionByName(answered_edges, allowMissingColumns=True) \\\n", " .unionByName(about_movie_edges,allowMissingColumns=True) \\\n", " .unionByName(about_tag_edges, allowMissingColumns=True) \\\n", " .unionByName(directed_edges, allowMissingColumns=True) \\\n", " .unionByName(acted_in_edges, allowMissingColumns=True) \n", "\n", "from graphframes import GraphFrames\n", "g = GraphFrames(vertices, edges)\n" ] }, { "cell_type": "code", "execution_count": 13, "id": "84b244b7-96e9-4aa7-837a-e0567a251208", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "[Stage 12:====================================================> (235 + 5) / 240]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+--------+-------+\n", "| type| count|\n", "+--------+-------+\n", "| review|2624608|\n", "| movie| 84661|\n", "| user| 247383|\n", "| actor| 123049|\n", "|director| 33604|\n", "| survey| 58903|\n", "| tag| 1094|\n", "+--------+-------+\n", "\n", "CPU times: user 9.25 ms, sys: 3.08 ms, total: 12.3 ms\n", "Wall time: 4.33 s\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "%%time\n", "vertices.groupBy(\"type\").count().show()" ] }, { "cell_type": "code", "execution_count": 14, "id": "00e0a6b4-fe90-4d65-b03b-6696b2e43aa9", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "[Stage 32:=====================================================>(227 + 2) / 229]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+-----------+--------+\n", "| etype| count|\n", "+-----------+--------+\n", "| rated|28438449|\n", "| has_review| 2624608|\n", "|about_movie| 58903|\n", "| has_tag| 212704|\n", "| about_tag| 58903|\n", "| acted_in| 371629|\n", "| directed| 84294|\n", "| answered| 58903|\n", "+-----------+--------+\n", "\n", "CPU times: user 9.29 ms, sys: 7.33 ms, total: 16.6 ms\n", "Wall time: 4.68 s\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "%%time\n", "edges.groupBy(\"etype\").count().show()" ] }, { "cell_type": "markdown", "id": "9933bac3-3b9d-4b9b-9ed9-f1dfe153e2ae", "metadata": {}, "source": [ "## QUERIES" ] }, { "attachments": {}, "cell_type": "markdown", "id": "1779060a-b1de-4e26-8814-cb21fc315a6e", "metadata": {}, "source": [ "# === DIRECTOR, ACTOR, MOVIE===" ] }, { "cell_type": "code", "execution_count": 15, "id": "dee09cdf-905a-4334-9e9b-aac1d708b9ab", "metadata": {}, "outputs": [], "source": [ "import time\n", "\n", "def measure(df):\n", " start = time.time()\n", " df.count() # triggers the Spark job\n", " end = time.time()\n", " print(f\"Time taken: {end - start:.3f} seconds\")\n", " return df # return the DataFrame itself\n" ] }, { "attachments": {}, "cell_type": "markdown", "id": "dc8b0297-a65f-4091-a414-415e35fdc056", "metadata": {}, "source": [ "### Top 10 Directors by Average Movie Rating" ] }, { "cell_type": "code", "execution_count": 19, "id": "1f978aa0-bf49-409a-be7d-d48918fa858d", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Top 10 Directors by Average Movie Rating (Movies ≥ 3)\n", "\n", "Time taken: 1.505 seconds\n", "+----------------+---------+------+\n", "|Director |AvgRating|Movies|\n", "+----------------+---------+------+\n", "|Darren Wilson |4.8 |5 |\n", "|Alan Lowery |4.45 |5 |\n", "|B. S. Johnson |4.38 |4 |\n", "|Robin Anderson |4.33 |3 |\n", "|Juanita Wilson |4.33 |3 |\n", "|Ladislav Smoljak|4.3 |19 |\n", "|Thierry Chabert |4.28 |3 |\n", "|Joël Séria |4.25 |3 |\n", "|Daniel McCabe |4.24 |3 |\n", "|John Downer |4.22 |3 |\n", "+----------------+---------+------+\n", "\n" ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# --------------------------------------------------\n", "# Motif: (director)-[:directed]->(movie)\n", "# --------------------------------------------------\n", "\n", "directed_motif = g.find(\"\"\"\n", " (director)-[e]->(movie)\n", "\"\"\").filter(\n", " (F.col(\"e.etype\") == \"directed\") & # edge type\n", " (F.col(\"director.type\") == \"director\") &\n", " (F.col(\"movie.type\") == \"movie\") &\n", " (F.col(\"movie.avgRating\").isNotNull())\n", ")\n", "\n", "# --------------------------------------------------\n", "# Aggregate per director\n", "# --------------------------------------------------\n", "\n", "director_stats = directed_motif.groupBy(\n", " F.col(\"director.id\").alias(\"DirectorID\"),\n", " F.col(\"director.name\").alias(\"Director\")\n", ").agg(\n", " F.round(F.avg(\"movie.avgRating\"), 2).alias(\"AvgRating\"),\n", " F.count(\"movie.id\").alias(\"Movies\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Filter to avoid 1-hit wonders and rank\n", "# --------------------------------------------------\n", "\n", "final_df = director_stats.filter(\n", " F.col(\"Movies\") >= 3\n", ").orderBy(\n", " F.desc(\"AvgRating\")\n", ").limit(10).select(\n", " F.col(\"Director\"),\n", " F.col(\"AvgRating\"),\n", " F.col(\"Movies\")\n", ")\n", "\n", "print(\"\\nTop 10 Directors by Average Movie Rating (Movies ≥ 3)\\n\")\n", "measure(final_df).show(truncate=False)\n" ] }, { "attachments": {}, "cell_type": "markdown", "id": "2bf338e7-255e-4f84-9f2b-291567439fe4", "metadata": {}, "source": [ "### Directors with the Most Movies Above 4.0 Rating (Not just average, how many great movies have they made? Considered minimum of 2 movies)" ] }, { "cell_type": "code", "execution_count": 20, "id": "4089129c-dd3d-4299-b8a0-e501390cb733", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Directors with Most Movies Rated ≥ 4.0 (Minimum 2)\n", "\n", "Time taken: 1.034 seconds\n", "+----------------+---------------+\n", "|Director |HighRatedMovies|\n", "+----------------+---------------+\n", "|Ladislav Smoljak|18 |\n", "|Akira Kurosawa |11 |\n", "|Alfred Hitchcock|10 |\n", "|Chris Marker |8 |\n", "|Chuck Jones |8 |\n", "|Adam Curtis |8 |\n", "|Hayao Miyazaki |7 |\n", "|Igor Maslennikov|6 |\n", "|Billy Wilder |6 |\n", "|Masaki Kobayashi|6 |\n", "+----------------+---------------+\n", "\n" ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# --------------------------------------------------\n", "# Motif: (director)-[:directed]->(movie)\n", "# --------------------------------------------------\n", "\n", "directed_motif = g.find(\"\"\"\n", " (director)-[e]->(movie)\n", "\"\"\").filter(\n", " (F.col(\"e.etype\") == \"directed\") & # edge type\n", " (F.col(\"director.type\") == \"director\") &\n", " (F.col(\"movie.type\") == \"movie\") &\n", " (F.col(\"movie.avgRating\") >= 4.0)\n", ")\n", "\n", "# --------------------------------------------------\n", "# Count high-rated movies per director\n", "# --------------------------------------------------\n", "\n", "director_stats = directed_motif.groupBy(\n", " F.col(\"director.id\").alias(\"DirectorID\"),\n", " F.col(\"director.name\").alias(\"Director\")\n", ").agg(\n", " F.count(\"movie.id\").alias(\"HighRatedMovies\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Filter and rank\n", "# --------------------------------------------------\n", "\n", "final_df = director_stats.filter(\n", " F.col(\"HighRatedMovies\") >= 2\n", ").orderBy(\n", " F.desc(\"HighRatedMovies\")\n", ").limit(10).select(\n", " F.col(\"Director\"),\n", " F.col(\"HighRatedMovies\")\n", ")\n", "\n", "print(\"\\nDirectors with Most Movies Rated ≥ 4.0 (Minimum 2)\\n\")\n", "measure(final_df).show(truncate=False)" ] }, { "cell_type": "markdown", "id": "78ebead8-ab23-40e1-828b-6a74a7a778de", "metadata": {}, "source": [ "### Top 10 Actors by Number of Movies Acted In" ] }, { "cell_type": "code", "execution_count": 21, "id": "b52c2585-5cd5-4ee2-a948-4be095068423", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Top 10 Actors by Number of Movies Acted In\n", "\n", "Time taken: 1.255 seconds\n", "+-----------------+-----------+\n", "|Actor |TotalMovies|\n", "+-----------------+-----------+\n", "|Eric Roberts |171 |\n", "|Christopher Lee |166 |\n", "|Mel Blanc |144 |\n", "|Donald Sutherland|141 |\n", "|Clarence Nash |133 |\n", "|Jr. |132 |\n", "|John Wayne |127 |\n", "|Gérard Depardieu |123 |\n", "|Ray Milland |121 |\n", "|Michael Ironside |120 |\n", "+-----------------+-----------+\n", "\n" ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# --------------------------------------------------\n", "# Motif: (actor)-[:acted_in]->(movie)\n", "# --------------------------------------------------\n", "\n", "acted_motif = g.find(\"\"\"\n", " (actor)-[e]->(movie)\n", "\"\"\").filter(\n", " (F.col(\"e.etype\") == \"acted_in\") & # adjust if using 'relationship'\n", " (F.col(\"actor.type\") == \"actor\") &\n", " (F.col(\"movie.type\") == \"movie\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Count movies per actor\n", "# --------------------------------------------------\n", "\n", "actor_stats = acted_motif.groupBy(\n", " F.col(\"actor.id\").alias(\"ActorID\"),\n", " F.col(\"actor.name\").alias(\"Actor\")\n", ").agg(\n", " F.count(\"movie.id\").alias(\"TotalMovies\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Order and limit\n", "# --------------------------------------------------\n", "\n", "final_df = actor_stats.orderBy(\n", " F.desc(\"TotalMovies\")\n", ").limit(10).select(\n", " F.col(\"Actor\"),\n", " F.col(\"TotalMovies\")\n", ")\n", "\n", "print(\"\\nTop 10 Actors by Number of Movies Acted In\\n\")\n", "measure(final_df).show(truncate=False)\n" ] }, { "cell_type": "markdown", "id": "18395909-aeeb-4f54-b408-76e41bfd7163", "metadata": {}, "source": [ "### Top Actors by Average Rating of Movies (Which actors are most associated with successful movies?)" ] }, { "cell_type": "code", "execution_count": 22, "id": "46726693-9c50-415f-8f61-7ebd5f427d1f", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Top Actors by Average Movie Rating (Minimum 3 Movies)\n", "\n", "Time taken: 1.505 seconds\n", "+-----------------+---------+-----------+\n", "|Actor |AvgRating|TotalMovies|\n", "+-----------------+---------+-----------+\n", "|Dewshane Williams|5.0 |3 |\n", "|Scott Freiman |4.67 |3 |\n", "|William Hoyland |4.5 |3 |\n", "|Jang So-yeon |4.41 |3 |\n", "|Genadij Rumlena |4.4 |6 |\n", "|Ion Besoiu |4.4 |6 |\n", "|Jean Constantin |4.39 |8 |\n", "|Bořivoj Penc |4.38 |6 |\n", "|Toma Caragiu |4.36 |3 |\n", "|Puiu Calinescu |4.36 |3 |\n", "+-----------------+---------+-----------+\n", "\n" ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# --------------------------------------------------\n", "# Motif: (actor)-[:acted_in]->(movie)\n", "# --------------------------------------------------\n", "\n", "acted_motif = g.find(\"\"\"\n", " (actor)-[e]->(movie)\n", "\"\"\").filter(\n", " (F.col(\"e.etype\") == \"acted_in\") & # adjust if using 'relationship'\n", " (F.col(\"actor.type\") == \"actor\") &\n", " (F.col(\"movie.type\") == \"movie\") &\n", " (F.col(\"movie.avgRating\").isNotNull())\n", ")\n", "\n", "# --------------------------------------------------\n", "# Aggregate per actor\n", "# --------------------------------------------------\n", "\n", "actor_stats = acted_motif.groupBy(\n", " F.col(\"actor.id\").alias(\"ActorID\"),\n", " F.col(\"actor.name\").alias(\"Actor\")\n", ").agg(\n", " F.round(F.avg(\"movie.avgRating\"), 2).alias(\"AvgRating\"),\n", " F.count(\"movie.id\").alias(\"TotalMovies\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Filter and rank\n", "# --------------------------------------------------\n", "\n", "final_df = actor_stats.filter(\n", " F.col(\"TotalMovies\") >= 3\n", ").orderBy(\n", " F.desc(\"AvgRating\")\n", ").limit(10).select(\n", " F.col(\"Actor\"),\n", " F.col(\"AvgRating\"),\n", " F.col(\"TotalMovies\")\n", ")\n", "\n", "print(\"\\nTop Actors by Average Movie Rating (Minimum 3 Movies)\\n\")\n", "measure(final_df).show(truncate=False)\n" ] }, { "attachments": {}, "cell_type": "markdown", "id": "20ec0d2a-24b5-49ff-a208-c786a668529b", "metadata": {}, "source": [ "### Top Actor-Director Pairs by Number of Collaborations" ] }, { "cell_type": "code", "execution_count": 23, "id": "55fa69a8-0c51-40a0-bf28-a062385994cc", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Top Actor-Director Pairs by Number of Collaborations\n", "\n", "Time taken: 1.817 seconds\n", "+-----------------+---------------+--------------+\n", "|Director |Actor |Collaborations|\n", "+-----------------+---------------+--------------+\n", "|Chuck Jones |Mel Blanc |66 |\n", "|Jack Hannah |Clarence Nash |47 |\n", "|Georges Méliès |Georges Méliès |44 |\n", "|Charles Chaplin |Charles Chaplin|41 |\n", "|Jack King |Clarence Nash |39 |\n", "|Buster Keaton |Buster Keaton |32 |\n", "|Friz Freleng |Mel Blanc |30 |\n", "|Woody Allen |Woody Allen |29 |\n", "|Charles Chaplin |Edna Purviance |28 |\n", "|Miguel M. Delgado|Cantinflas |27 |\n", "+-----------------+---------------+--------------+\n", "\n" ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# --------------------------------------------------\n", "# 2-hop motif:\n", "# (director)-[:directed]->(movie)<-[:acted_in]-(actor)\n", "# --------------------------------------------------\n", "\n", "collab_motif = g.find(\"\"\"\n", " (director)-[e1]->(movie);\n", " (actor)-[e2]->(movie)\n", "\"\"\").filter(\n", " (F.col(\"e1.etype\") == \"directed\") &\n", " (F.col(\"e2.etype\") == \"acted_in\") &\n", " (F.col(\"director.type\") == \"director\") &\n", " (F.col(\"actor.type\") == \"actor\") &\n", " (F.col(\"movie.type\") == \"movie\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Count collaborations per (director, actor) pair\n", "# --------------------------------------------------\n", "\n", "pair_stats = collab_motif.groupBy(\n", " F.col(\"director.id\").alias(\"DirectorID\"),\n", " F.col(\"director.name\").alias(\"Director\"),\n", " F.col(\"actor.id\").alias(\"ActorID\"),\n", " F.col(\"actor.name\").alias(\"Actor\")\n", ").agg(\n", " F.count(\"movie.id\").alias(\"Collaborations\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Rank pairs\n", "# --------------------------------------------------\n", "\n", "final_df = pair_stats.orderBy(\n", " F.desc(\"Collaborations\")\n", ").limit(10).select(\n", " F.col(\"Director\"),\n", " F.col(\"Actor\"),\n", " F.col(\"Collaborations\")\n", ")\n", "\n", "print(\"\\nTop Actor-Director Pairs by Number of Collaborations\\n\")\n", "measure(final_df).show(truncate=False)\n" ] }, { "attachments": {}, "cell_type": "markdown", "id": "f5a62743-1e75-42bb-97d6-4870d86e21c7", "metadata": {}, "source": [ "### Actor-Director Collaboration Quality (Which pairs consistently produce great movies?)" ] }, { "cell_type": "code", "execution_count": 24, "id": "1525f7a2-a1f3-445c-8bb0-6c0f55b70e97", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Actor-Director Collaboration Quality (Min 3 Movies)\n", "\n", "Time taken: 1.646 seconds\n", "+-----------------+----------------+-------+---------+\n", "|Director |Actor |Collabs|AvgRating|\n", "+-----------------+----------------+-------+---------+\n", "|John Stimpson |Tom Kemp |3 |4.79 |\n", "|Alan Lowery |John Pilger |3 |4.58 |\n", "|B. S. Johnson |William Hoyland |3 |4.5 |\n", "|Ladislav Smoljak |Genadij Rumlena |6 |4.4 |\n", "|Ladislav Smoljak |Ladislav Smoljak|12 |4.38 |\n", "|Sergiu Nicolaescu|Ion Besoiu |5 |4.38 |\n", "|Ladislav Smoljak |Bořivoj Penc |6 |4.38 |\n", "|Mircea Drăgan |Dem Rădulescu |3 |4.36 |\n", "|Mircea Drăgan |Toma Caragiu |3 |4.36 |\n", "|Mircea Drăgan |Puiu Calinescu |3 |4.36 |\n", "+-----------------+----------------+-------+---------+\n", "\n" ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# --------------------------------------------------\n", "# 2-hop motif:\n", "# (director)-[:directed]->(movie)<-[:acted_in]-(actor)\n", "# --------------------------------------------------\n", "\n", "collab_motif = g.find(\"\"\"\n", " (director)-[e1]->(movie);\n", " (actor)-[e2]->(movie)\n", "\"\"\").filter(\n", " (F.col(\"e1.etype\") == \"directed\") &\n", " (F.col(\"e2.etype\") == \"acted_in\") &\n", " (F.col(\"director.type\") == \"director\") &\n", " (F.col(\"actor.type\") == \"actor\") &\n", " (F.col(\"movie.type\") == \"movie\") &\n", " (F.col(\"movie.avgRating\").isNotNull())\n", ")\n", "\n", "# --------------------------------------------------\n", "# Aggregate collaboration stats\n", "# --------------------------------------------------\n", "\n", "pair_stats = collab_motif.groupBy(\n", " F.col(\"director.id\").alias(\"DirectorID\"),\n", " F.col(\"director.name\").alias(\"Director\"),\n", " F.col(\"actor.id\").alias(\"ActorID\"),\n", " F.col(\"actor.name\").alias(\"Actor\")\n", ").agg(\n", " F.count(\"movie.id\").alias(\"Collabs\"),\n", " F.round(F.avg(\"movie.avgRating\"), 2).alias(\"AvgRating\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Filter to serious collaborations and rank\n", "# --------------------------------------------------\n", "\n", "final_df = pair_stats.filter(\n", " F.col(\"Collabs\") >= 3\n", ").orderBy(\n", " F.desc(\"AvgRating\")\n", ").limit(10).select(\n", " F.col(\"Director\"),\n", " F.col(\"Actor\"),\n", " F.col(\"Collabs\"),\n", " F.col(\"AvgRating\")\n", ")\n", "\n", "print(\"\\nActor-Director Collaboration Quality (Min 3 Movies)\\n\")\n", "measure(final_df).show(truncate=False)\n" ] }, { "attachments": {}, "cell_type": "markdown", "id": "e762362d-8acd-4e64-a4c9-3abf90c14859", "metadata": {}, "source": [ "### Top Actor Groups That Consistently Produce High-Rated Movies" ] }, { "cell_type": "code", "execution_count": 25, "id": "82a60527-ab8d-40c8-b55a-445df7b40f83", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Top Actor Triplets Producing High-Rated Movies\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] }, { "name": "stdout", "output_type": "stream", "text": [ "Time taken: 5.455 seconds\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ "[Stage 241:=======> (27 + 173) / 200]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+----------------+----------------+-----------------+----------+---------+--------------------------------------------------------------------------------------------------------------------------------------------+\n", "|Actor1 |Actor2 |Actor3 |MovieCount|AvgRating|Movies |\n", "+----------------+----------------+-----------------+----------+---------+--------------------------------------------------------------------------------------------------------------------------------------------+\n", "|Amza Pellea |Jean Constantin |Sergiu Nicolaescu|4 |4.75 |[Revenge (1978), Nemuritorii (1974), A Police Inspector Calls (1974), The Last Bullet (1973)] |\n", "|Ion Besoiu |Jean Constantin |Sergiu Nicolaescu|3 |4.67 |[Revenge (1978), Nemuritorii (1974), A Police Inspector Calls (1974)] |\n", "|Amza Pellea |Ion Besoiu |Jean Constantin |3 |4.67 |[Revenge (1978), Nemuritorii (1974), A Police Inspector Calls (1974)] |\n", "|Amza Pellea |Ion Besoiu |Sergiu Nicolaescu|3 |4.67 |[Revenge (1978), Nemuritorii (1974), A Police Inspector Calls (1974)] |\n", "|Genadij Rumlena |Ladislav Smoljak|Zdeněk Svěrák |3 |4.46 |[Dobytí severního pólu (1985), Dumb Bobes, or Czech Tarzan (1971), An Uncertain Season (1988)] |\n", "|Ladislav Smoljak|Miloň Čepelka |Zdeněk Svěrák |4 |4.45 |[Akt (1997), Vyšetřování ztráty třídní knihy (1997), Hospoda Na mýtince (1997), An Uncertain Season (1988)] |\n", "|Dem Rădulescu |Puiu Calinescu |Toma Caragiu |3 |4.36 |[Brigade Miscellaneous on Alert (1971), Brigade Miscellaneous Steps In (1970), Brigade Miscellaneous in the Mountains and at the Sea (1971)]|\n", "|Jaroslav Weigel |Ladislav Smoljak|Miloň Čepelka |4 |4.35 |[Posel z Liptákova (1977), Akt (1997), Vyšetřování ztráty třídní knihy (1997), Murder in a Parlor Car Compartment (1997)] |\n", "|Lex Barker |Pierre Brice |Ralf Wolter |4 |4.34 |[Winnetou: The Last Shot (1965), Winnetou and the Crossbreed (1966), In the Valley of Death (1968), Old Shatterhand (1964)] |\n", "|Jan Hraběta |Ladislav Smoljak|Zdeněk Svěrák |3 |4.33 |[Dumb Bobes, or Czech Tarzan (1971), Akt (1997), Lijavec (1982)] |\n", "+----------------+----------------+-----------------+----------+---------+--------------------------------------------------------------------------------------------------------------------------------------------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# --------------------------------------------------\n", "# 3-actor motif sharing the same movie\n", "# (a1)-[:acted_in]->(m)<-[:acted_in]-(a2)\n", "# (a3)-[:acted_in]->(m)\n", "# --------------------------------------------------\n", "\n", "triplet_motif = g.find(\"\"\"\n", " (a1)-[e1]->(m);\n", " (a2)-[e2]->(m);\n", " (a3)-[e3]->(m)\n", "\"\"\").filter(\n", " (F.col(\"e1.etype\") == \"acted_in\") &\n", " (F.col(\"e2.etype\") == \"acted_in\") &\n", " (F.col(\"e3.etype\") == \"acted_in\") &\n", " (F.col(\"a1.type\") == \"actor\") &\n", " (F.col(\"a2.type\") == \"actor\") &\n", " (F.col(\"a3.type\") == \"actor\") &\n", " (F.col(\"m.type\") == \"movie\") &\n", " (F.col(\"m.avgRating\").isNotNull()) &\n", " # enforce unique ordered triplets\n", " (F.col(\"a1.name\") < F.col(\"a2.name\")) &\n", " (F.col(\"a2.name\") < F.col(\"a3.name\"))\n", ")\n", "\n", "# --------------------------------------------------\n", "# Aggregate stats per actor triplet\n", "# --------------------------------------------------\n", "\n", "triplet_stats = triplet_motif.groupBy(\n", " F.col(\"a1.name\").alias(\"Actor1\"),\n", " F.col(\"a2.name\").alias(\"Actor2\"),\n", " F.col(\"a3.name\").alias(\"Actor3\")\n", ").agg(\n", " F.countDistinct(\"m.id\").alias(\"MovieCount\"),\n", " F.round(F.avg(\"m.avgRating\"), 2).alias(\"AvgRating\"),\n", " F.collect_set(\"m.title\").alias(\"Movies\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Filter serious collaborations and rank\n", "# --------------------------------------------------\n", "\n", "final_df = triplet_stats.filter(\n", " F.col(\"MovieCount\") >= 3\n", ").orderBy(\n", " F.desc(\"AvgRating\")\n", ").limit(10)\n", "\n", "print(\"\\nTop Actor Triplets Producing High-Rated Movies\\n\")\n", "measure(final_df).show(truncate=False)\n" ] }, { "cell_type": "markdown", "id": "b17b27b9-07ea-409d-bcf9-59317c811135", "metadata": {}, "source": [ "### Actors Frequently Appearing in High-Rated Movies with Different Directors (Highlights high-performing, collaborative actors trusted by many top directors.)" ] }, { "cell_type": "code", "execution_count": 26, "id": "ff42a85a-434b-49b8-9356-614bb6a6a5f4", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Actors Frequently Appearing in High-Rated Movies with Different Directors\n", "\n", "Time taken: 1.371 seconds\n", "+------------------+------------------+--------------+\n", "|Actor |UniqueTopDirectors|TopRatedMovies|\n", "+------------------+------------------+--------------+\n", "|Tom Hanks |15 |5 |\n", "|Stephen Root |13 |3 |\n", "|Harry Dean Stanton|12 |12 |\n", "|Eric Roberts |11 |11 |\n", "|Donald Pleasence |11 |8 |\n", "|Franco Nero |9 |9 |\n", "|Alberto Sordi |8 |11 |\n", "|David Attenborough|8 |9 |\n", "|Peter Coyote |8 |8 |\n", "|Malcolm McDowell |8 |8 |\n", "+------------------+------------------+--------------+\n", "\n" ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# --------------------------------------------------\n", "# Motif:\n", "# (actor)-[:acted_in]->(movie)<-[:directed]-(director)\n", "# --------------------------------------------------\n", "\n", "actor_director_motif = g.find(\"\"\"\n", " (actor)-[e1]->(movie);\n", " (director)-[e2]->(movie)\n", "\"\"\").filter(\n", " (F.col(\"e1.etype\") == \"acted_in\") &\n", " (F.col(\"e2.etype\") == \"directed\") &\n", " (F.col(\"actor.type\") == \"actor\") &\n", " (F.col(\"director.type\") == \"director\") &\n", " (F.col(\"movie.type\") == \"movie\") &\n", " (F.col(\"movie.avgRating\") >= 4.0)\n", ")\n", "\n", "# --------------------------------------------------\n", "# Aggregate stats per actor\n", "# --------------------------------------------------\n", "\n", "actor_stats = actor_director_motif.groupBy(\n", " F.col(\"actor.id\").alias(\"ActorID\"),\n", " F.col(\"actor.name\").alias(\"Actor\")\n", ").agg(\n", " F.countDistinct(\"director.id\").alias(\"UniqueTopDirectors\"),\n", " F.countDistinct(\"movie.id\").alias(\"TopRatedMovies\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Filter and rank\n", "# --------------------------------------------------\n", "\n", "final_df = actor_stats.filter(\n", " F.col(\"TopRatedMovies\") >= 3\n", ").orderBy(\n", " F.desc(\"UniqueTopDirectors\"),\n", " F.desc(\"TopRatedMovies\")\n", ").limit(10).select(\n", " F.col(\"Actor\"),\n", " F.col(\"UniqueTopDirectors\"),\n", " F.col(\"TopRatedMovies\")\n", ")\n", "\n", "print(\"\\nActors Frequently Appearing in High-Rated Movies with Different Directors\\n\")\n", "measure(final_df).show(truncate=False)\n" ] }, { "attachments": {}, "cell_type": "markdown", "id": "57f7ae65-da66-4899-b2a2-6fe218efec76", "metadata": {}, "source": [ "### ===USER + RATING====\n", "### Top 10 Most Active Users by Rating Count" ] }, { "cell_type": "code", "execution_count": 27, "id": "9caae66c-a2c0-430a-b637-e5823304de6c", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Top 10 Most Active Users by Rating Count\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] }, { "name": "stdout", "output_type": "stream", "text": [ "Time taken: 7.872 seconds\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ "[Stage 294:=========================================> (160 + 40) / 200]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+------+-----------+\n", "|UserID|RatedMovies|\n", "+------+-----------+\n", "|347565|34377 |\n", "|414405|9327 |\n", "|741204|9279 |\n", "|632423|8828 |\n", "|144113|8569 |\n", "|928461|7515 |\n", "|19163 |7069 |\n", "|801871|7046 |\n", "|540572|6990 |\n", "|334733|6071 |\n", "+------+-----------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# --------------------------------------------------\n", "# Motif: (user)-[:rated]->(movie)\n", "# --------------------------------------------------\n", "\n", "rated_motif = g.find(\"\"\"\n", " (user)-[e]->(movie)\n", "\"\"\").filter(\n", " (F.col(\"e.etype\") == \"rated\") & # adjust if using 'relationship'\n", " (F.col(\"user.type\") == \"user\") &\n", " (F.col(\"movie.type\") == \"movie\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Count ratings per user\n", "# --------------------------------------------------\n", "\n", "user_stats = rated_motif.groupBy(\n", " F.col(\"user.id\").alias(\"UserID\")\n", ").agg(\n", " F.count(\"movie.id\").alias(\"RatedMovies\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Rank users\n", "# --------------------------------------------------\n", "\n", "final_df = user_stats.orderBy(\n", " F.desc(\"RatedMovies\")\n", ").limit(10)\n", "\n", "print(\"\\nTop 10 Most Active Users by Rating Count\\n\")\n", "measure(final_df).show(truncate=False)\n" ] }, { "attachments": {}, "cell_type": "markdown", "id": "248b88b4-26d4-414e-a0a1-ecb20dbe18bd", "metadata": {}, "source": [ "### Users Who Rate Harshly (Lowest Average Rating)" ] }, { "cell_type": "code", "execution_count": 28, "id": "71995c26-32bd-43fa-b091-400278bcc516", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Users Who Rate Harshly (Lowest Average Rating, ≥ 5 Ratings)\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] }, { "name": "stdout", "output_type": "stream", "text": [ "Time taken: 7.177 seconds\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ "[Stage 315:============================================> (171 + 29) / 200]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+------+------------+---------+\n", "|UserID|TotalRatings|AvgRating|\n", "+------+------------+---------+\n", "|156696|15 |0.5 |\n", "|150876|9 |0.5 |\n", "|518407|15 |0.5 |\n", "|491966|15 |0.5 |\n", "|520972|20 |0.5 |\n", "|225736|41 |0.5 |\n", "|15630 |15 |0.5 |\n", "|930000|16 |0.5 |\n", "|271154|7 |0.5 |\n", "|857442|8 |0.5 |\n", "+------+------------+---------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# --------------------------------------------------\n", "# Motif: (user)-[rated]->(movie)\n", "# --------------------------------------------------\n", "\n", "rated_motif = g.find(\"\"\"\n", " (user)-[e]->(movie)\n", "\"\"\").filter(\n", " (F.col(\"e.etype\") == \"rated\") & # adjust if using 'relationship'\n", " (F.col(\"user.type\") == \"user\") &\n", " (F.col(\"movie.type\") == \"movie\") &\n", " (F.col(\"e.rating\").isNotNull())\n", ")\n", "\n", "# --------------------------------------------------\n", "# Aggregate per user\n", "# --------------------------------------------------\n", "\n", "user_stats = rated_motif.groupBy(\n", " F.col(\"user.id\").alias(\"UserID\")\n", ").agg(\n", " F.count(\"movie.id\").alias(\"TotalRatings\"),\n", " F.round(F.avg(\"e.rating\"), 2).alias(\"AvgRating\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Filter serious raters and rank (harsh first)\n", "# --------------------------------------------------\n", "\n", "final_df = user_stats.filter(\n", " F.col(\"TotalRatings\") >= 5\n", ").orderBy(\n", " F.asc(\"AvgRating\") # lowest average first\n", ").limit(10)\n", "\n", "print(\"\\nUsers Who Rate Harshly (Lowest Average Rating, ≥ 5 Ratings)\\n\")\n", "measure(final_df).show(truncate=False)\n" ] }, { "attachments": {}, "cell_type": "markdown", "id": "05eee6be-96f0-484d-b3d5-8d9654251947", "metadata": {}, "source": [ "### Most Similar Users Based on Ratings for the Same Movies (Find pairs of users with overlapping movie preferences. (Collaborative filtering foundation))" ] }, { "cell_type": "code", "execution_count": 18, "id": "baf8cdca-770f-451f-9ad1-803426d9105b", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "ERROR:root:KeyboardInterrupt while sending command. (0 + 200) / 200]]\n", "Traceback (most recent call last):\n", " File \"/shared/spark-3.4.4-bin-hadoop3/python/lib/py4j-0.10.9.7-src.zip/py4j/java_gateway.py\", line 1038, in send_command\n", " response = connection.send_command(command)\n", " File \"/shared/spark-3.4.4-bin-hadoop3/python/lib/py4j-0.10.9.7-src.zip/py4j/clientserver.py\", line 511, in send_command\n", " answer = smart_decode(self.stream.readline()[:-1])\n", " File \"/shared/anaconda3/lib/python3.10/socket.py\", line 717, in readinto\n", " return self._sock.recv_into(b)\n", "KeyboardInterrupt\n" ] }, { "ename": "KeyboardInterrupt", "evalue": "", "output_type": "error", "traceback": [ "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m", "\u001b[0;31mKeyboardInterrupt\u001b[0m Traceback (most recent call last)", "Cell \u001b[0;32mIn[18], line 42\u001b[0m\n\u001b[1;32m 31\u001b[0m user_pair_stats \u001b[38;5;241m=\u001b[39m similar_users_motif\u001b[38;5;241m.\u001b[39mgroupBy(\n\u001b[1;32m 32\u001b[0m F\u001b[38;5;241m.\u001b[39mcol(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mu1.id\u001b[39m\u001b[38;5;124m\"\u001b[39m)\u001b[38;5;241m.\u001b[39malias(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mUser1\u001b[39m\u001b[38;5;124m\"\u001b[39m),\n\u001b[1;32m 33\u001b[0m F\u001b[38;5;241m.\u001b[39mcol(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mu2.id\u001b[39m\u001b[38;5;124m\"\u001b[39m)\u001b[38;5;241m.\u001b[39malias(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mUser2\u001b[39m\u001b[38;5;124m\"\u001b[39m)\n\u001b[1;32m 34\u001b[0m )\u001b[38;5;241m.\u001b[39magg(\n\u001b[1;32m 35\u001b[0m F\u001b[38;5;241m.\u001b[39mcountDistinct(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mm.id\u001b[39m\u001b[38;5;124m\"\u001b[39m)\u001b[38;5;241m.\u001b[39malias(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mSharedMovies\u001b[39m\u001b[38;5;124m\"\u001b[39m)\n\u001b[1;32m 36\u001b[0m )\n\u001b[1;32m 38\u001b[0m final_df \u001b[38;5;241m=\u001b[39m user_pair_stats\u001b[38;5;241m.\u001b[39morderBy(\n\u001b[1;32m 39\u001b[0m F\u001b[38;5;241m.\u001b[39mdesc(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mSharedMovies\u001b[39m\u001b[38;5;124m\"\u001b[39m)\n\u001b[1;32m 40\u001b[0m )\u001b[38;5;241m.\u001b[39mlimit(\u001b[38;5;241m10\u001b[39m)\n\u001b[0;32m---> 42\u001b[0m \u001b[43mmeasure\u001b[49m\u001b[43m(\u001b[49m\u001b[43mfinal_df\u001b[49m\u001b[43m)\u001b[49m\u001b[38;5;241m.\u001b[39mshow(truncate\u001b[38;5;241m=\u001b[39m\u001b[38;5;28;01mFalse\u001b[39;00m)\n", "Cell \u001b[0;32mIn[15], line 5\u001b[0m, in \u001b[0;36mmeasure\u001b[0;34m(df)\u001b[0m\n\u001b[1;32m 3\u001b[0m \u001b[38;5;28;01mdef\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[38;5;21mmeasure\u001b[39m(df):\n\u001b[1;32m 4\u001b[0m start \u001b[38;5;241m=\u001b[39m time\u001b[38;5;241m.\u001b[39mtime()\n\u001b[0;32m----> 5\u001b[0m \u001b[43mdf\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mcount\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m \u001b[38;5;66;03m# triggers the Spark job\u001b[39;00m\n\u001b[1;32m 6\u001b[0m end \u001b[38;5;241m=\u001b[39m time\u001b[38;5;241m.\u001b[39mtime()\n\u001b[1;32m 7\u001b[0m \u001b[38;5;28mprint\u001b[39m(\u001b[38;5;124mf\u001b[39m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mTime taken: \u001b[39m\u001b[38;5;132;01m{\u001b[39;00mend\u001b[38;5;250m \u001b[39m\u001b[38;5;241m-\u001b[39m\u001b[38;5;250m \u001b[39mstart\u001b[38;5;132;01m:\u001b[39;00m\u001b[38;5;124m.3f\u001b[39m\u001b[38;5;132;01m}\u001b[39;00m\u001b[38;5;124m seconds\u001b[39m\u001b[38;5;124m\"\u001b[39m)\n", "File \u001b[0;32m/shared/spark-3.4.4-bin-hadoop3/python/pyspark/sql/dataframe.py:1195\u001b[0m, in \u001b[0;36mDataFrame.count\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m 1172\u001b[0m \u001b[38;5;28;01mdef\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[38;5;21mcount\u001b[39m(\u001b[38;5;28mself\u001b[39m) \u001b[38;5;241m-\u001b[39m\u001b[38;5;241m>\u001b[39m \u001b[38;5;28mint\u001b[39m:\n\u001b[1;32m 1173\u001b[0m \u001b[38;5;250m \u001b[39m\u001b[38;5;124;03m\"\"\"Returns the number of rows in this :class:`DataFrame`.\u001b[39;00m\n\u001b[1;32m 1174\u001b[0m \n\u001b[1;32m 1175\u001b[0m \u001b[38;5;124;03m .. versionadded:: 1.3.0\u001b[39;00m\n\u001b[0;32m (...)\u001b[0m\n\u001b[1;32m 1193\u001b[0m \u001b[38;5;124;03m 3\u001b[39;00m\n\u001b[1;32m 1194\u001b[0m \u001b[38;5;124;03m \"\"\"\u001b[39;00m\n\u001b[0;32m-> 1195\u001b[0m \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28mint\u001b[39m(\u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_jdf\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mcount\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m)\n", "File \u001b[0;32m/shared/spark-3.4.4-bin-hadoop3/python/lib/py4j-0.10.9.7-src.zip/py4j/java_gateway.py:1321\u001b[0m, in \u001b[0;36mJavaMember.__call__\u001b[0;34m(self, *args)\u001b[0m\n\u001b[1;32m 1314\u001b[0m args_command, temp_args \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_build_args(\u001b[38;5;241m*\u001b[39margs)\n\u001b[1;32m 1316\u001b[0m command \u001b[38;5;241m=\u001b[39m proto\u001b[38;5;241m.\u001b[39mCALL_COMMAND_NAME \u001b[38;5;241m+\u001b[39m\\\n\u001b[1;32m 1317\u001b[0m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mcommand_header \u001b[38;5;241m+\u001b[39m\\\n\u001b[1;32m 1318\u001b[0m args_command \u001b[38;5;241m+\u001b[39m\\\n\u001b[1;32m 1319\u001b[0m proto\u001b[38;5;241m.\u001b[39mEND_COMMAND_PART\n\u001b[0;32m-> 1321\u001b[0m answer \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mgateway_client\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43msend_command\u001b[49m\u001b[43m(\u001b[49m\u001b[43mcommand\u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m 1322\u001b[0m return_value \u001b[38;5;241m=\u001b[39m get_return_value(\n\u001b[1;32m 1323\u001b[0m answer, \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mgateway_client, \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mtarget_id, \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mname)\n\u001b[1;32m 1325\u001b[0m \u001b[38;5;28;01mfor\u001b[39;00m temp_arg \u001b[38;5;129;01min\u001b[39;00m temp_args:\n", "File \u001b[0;32m/shared/spark-3.4.4-bin-hadoop3/python/lib/py4j-0.10.9.7-src.zip/py4j/java_gateway.py:1038\u001b[0m, in \u001b[0;36mGatewayClient.send_command\u001b[0;34m(self, command, retry, binary)\u001b[0m\n\u001b[1;32m 1036\u001b[0m connection \u001b[38;5;241m=\u001b[39m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_get_connection()\n\u001b[1;32m 1037\u001b[0m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[0;32m-> 1038\u001b[0m response \u001b[38;5;241m=\u001b[39m \u001b[43mconnection\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43msend_command\u001b[49m\u001b[43m(\u001b[49m\u001b[43mcommand\u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m 1039\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m binary:\n\u001b[1;32m 1040\u001b[0m \u001b[38;5;28;01mreturn\u001b[39;00m response, \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_create_connection_guard(connection)\n", "File \u001b[0;32m/shared/spark-3.4.4-bin-hadoop3/python/lib/py4j-0.10.9.7-src.zip/py4j/clientserver.py:511\u001b[0m, in \u001b[0;36mClientServerConnection.send_command\u001b[0;34m(self, command)\u001b[0m\n\u001b[1;32m 509\u001b[0m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[1;32m 510\u001b[0m \u001b[38;5;28;01mwhile\u001b[39;00m \u001b[38;5;28;01mTrue\u001b[39;00m:\n\u001b[0;32m--> 511\u001b[0m answer \u001b[38;5;241m=\u001b[39m smart_decode(\u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mstream\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mreadline\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m[:\u001b[38;5;241m-\u001b[39m\u001b[38;5;241m1\u001b[39m])\n\u001b[1;32m 512\u001b[0m logger\u001b[38;5;241m.\u001b[39mdebug(\u001b[38;5;124m\"\u001b[39m\u001b[38;5;124mAnswer received: \u001b[39m\u001b[38;5;132;01m{0}\u001b[39;00m\u001b[38;5;124m\"\u001b[39m\u001b[38;5;241m.\u001b[39mformat(answer))\n\u001b[1;32m 513\u001b[0m \u001b[38;5;66;03m# Happens when a the other end is dead. There might be an empty\u001b[39;00m\n\u001b[1;32m 514\u001b[0m \u001b[38;5;66;03m# answer before the socket raises an error.\u001b[39;00m\n", "File \u001b[0;32m/shared/anaconda3/lib/python3.10/socket.py:717\u001b[0m, in \u001b[0;36mSocketIO.readinto\u001b[0;34m(self, b)\u001b[0m\n\u001b[1;32m 715\u001b[0m \u001b[38;5;28;01mwhile\u001b[39;00m \u001b[38;5;28;01mTrue\u001b[39;00m:\n\u001b[1;32m 716\u001b[0m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[0;32m--> 717\u001b[0m \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_sock\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mrecv_into\u001b[49m\u001b[43m(\u001b[49m\u001b[43mb\u001b[49m\u001b[43m)\u001b[49m\n\u001b[1;32m 718\u001b[0m \u001b[38;5;28;01mexcept\u001b[39;00m timeout:\n\u001b[1;32m 719\u001b[0m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_timeout_occurred \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;01mTrue\u001b[39;00m\n", "\u001b[0;31mKeyboardInterrupt\u001b[0m: " ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# Step 1: Active users\n", "active_users = g.find(\"(u)-[e]->(m)\") \\\n", " .filter((F.col(\"e.etype\") == \"rated\") & (F.col(\"u.type\") == \"user\")) \\\n", " .groupBy(F.col(\"u.id\").alias(\"user_id\")) \\\n", " .agg(F.count(\"m.id\").alias(\"RatingCount\")) \\\n", " .filter(F.col(\"RatingCount\") >= 5) \\\n", " .select(\"user_id\")\n", "\n", "# Step 2: Similar ratings motif\n", "similar_users_motif = g.find(\"\"\"\n", " (u1)-[r1]->(m);\n", " (u2)-[r2]->(m)\n", "\"\"\").filter(\n", " (F.col(\"r1.etype\") == \"rated\") &\n", " (F.col(\"r2.etype\") == \"rated\") &\n", " (F.col(\"u1.type\") == \"user\") &\n", " (F.col(\"u2.type\") == \"user\") &\n", " (F.col(\"m.type\") == \"movie\") &\n", " (F.abs(F.col(\"r1.rating\") - F.col(\"r2.rating\")) <= 1)\n", ")\n", "\n", "# Step 3: Restrict to active users only\n", "similar_users_motif = similar_users_motif \\\n", " .join(active_users.alias(\"a1\"), F.col(\"u1.id\") == F.col(\"a1.user_id\")) \\\n", " .join(active_users.alias(\"a2\"), F.col(\"u2.id\") == F.col(\"a2.user_id\")) \\\n", " .filter(F.col(\"u1.id\") < F.col(\"u2.id\"))\n", "\n", "# Step 4: Aggregate early\n", "user_pair_stats = similar_users_motif.groupBy(\n", " F.col(\"u1.id\").alias(\"User1\"),\n", " F.col(\"u2.id\").alias(\"User2\")\n", ").agg(\n", " F.countDistinct(\"m.id\").alias(\"SharedMovies\")\n", ")\n", "\n", "final_df = user_pair_stats.orderBy(\n", " F.desc(\"SharedMovies\")\n", ").limit(10)\n", "\n", "measure(final_df).show(truncate=False)\n" ] }, { "attachments": {}, "cell_type": "markdown", "id": "392794aa-72bf-414a-82f3-1e26aa2b7004", "metadata": {}, "source": [ "### Users Whose Ratings Most Closely Match Movie Averages" ] }, { "cell_type": "code", "execution_count": 17, "id": "a3b24a2c-bbe0-49fb-819e-7b5845f34ffa", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Users Whose Ratings Most Closely Match Movie Averages\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] }, { "name": "stdout", "output_type": "stream", "text": [ "Time taken: 3.994 seconds\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ "[Stage 77:========================> (91 + 109) / 200]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+------+-----------+------------+\n", "|UserID|RatingCount|AvgDeviation|\n", "+------+-----------+------------+\n", "|960231|5 |0.051 |\n", "|224898|5 |0.066 |\n", "|944024|50 |0.071 |\n", "|593034|6 |0.081 |\n", "|313380|5 |0.093 |\n", "|51532 |18 |0.103 |\n", "|746601|5 |0.104 |\n", "|690265|8 |0.108 |\n", "|10700 |6 |0.113 |\n", "|748740|30 |0.12 |\n", "+------+-----------+------------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# --------------------------------------------------\n", "# Motif: (user)-[rated]->(movie)\n", "# --------------------------------------------------\n", "\n", "rated_motif = g.find(\"\"\"\n", " (user)-[e]->(movie)\n", "\"\"\").filter(\n", " (F.col(\"e.etype\") == \"rated\") & # adjust if using 'relationship'\n", " (F.col(\"user.type\") == \"user\") &\n", " (F.col(\"movie.type\") == \"movie\") &\n", " (F.col(\"movie.avgRating\").isNotNull()) &\n", " (F.col(\"e.rating\").isNotNull())\n", ")\n", "\n", "# --------------------------------------------------\n", "# Compute deviation per user\n", "# --------------------------------------------------\n", "\n", "user_stats = rated_motif.groupBy(\n", " F.col(\"user.id\").alias(\"UserID\")\n", ").agg(\n", " F.count(\"movie.id\").alias(\"RatingCount\"),\n", " F.round(\n", " F.avg(F.abs(F.col(\"e.rating\") - F.col(\"movie.avgRating\"))),\n", " 3\n", " ).alias(\"AvgDeviation\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Filter meaningful users and rank (closest first)\n", "# --------------------------------------------------\n", "\n", "final_df = user_stats.filter(\n", " F.col(\"RatingCount\") >= 5\n", ").orderBy(\n", " F.asc(\"AvgDeviation\")\n", ").limit(10)\n", "\n", "print(\"\\nUsers Whose Ratings Most Closely Match Movie Averages\\n\")\n", "measure(final_df).show(truncate=False)\n" ] }, { "attachments": {}, "cell_type": "markdown", "id": "06207784-5b48-407a-8719-5bcb9faff9b1", "metadata": {}, "source": [ "### ===REVIEW===\n", "### Movies with the Most Reviews\n" ] }, { "cell_type": "code", "execution_count": 16, "id": "60192cae-2a93-437a-8f13-4116c75ed9e1", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Movies with the Most Reviews\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] }, { "name": "stdout", "output_type": "stream", "text": [ "Time taken: 4.541 seconds\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ "[Stage 66:=====================================================> (48 + 2) / 50]]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+---------------------------------------------------------+-----------+\n", "|title |ReviewCount|\n", "+---------------------------------------------------------+-----------+\n", "|Lord of the Rings: The Fellowship of the Ring, The (2001)|5122 |\n", "|Dark Knight, The (2008) |5115 |\n", "|Shawshank Redemption, The (1994) |4913 |\n", "|Star Wars: Episode VII - The Force Awakens (2015) |4384 |\n", "|Matrix, The (1999) |3765 |\n", "|Star Wars: Episode I - The Phantom Menace (1999) |3636 |\n", "|Star Wars: Episode II - Attack of the Clones (2002) |3556 |\n", "|Blair Witch Project, The (1999) |3443 |\n", "|Batman v Superman: Dawn of Justice (2016) |3358 |\n", "|Star Wars: Episode III - Revenge of the Sith (2005) |3331 |\n", "+---------------------------------------------------------+-----------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# --------------------------------------------------\n", "# Correct motif: (review)-[:has_review]->(movie)\n", "# --------------------------------------------------\n", "\n", "review_motif = g.find(\"\"\"\n", " (review)-[e]->(movie)\n", "\"\"\").filter(\n", " (F.col(\"e.etype\") == \"has_review\") &\n", " (F.col(\"review.type\") == \"review\") &\n", " (F.col(\"movie.type\") == \"movie\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Count reviews per movie\n", "# --------------------------------------------------\n", "\n", "movie_stats = review_motif.groupBy(\n", " F.col(\"movie.id\").alias(\"movie_id\"),\n", " F.col(\"movie.title\").alias(\"title\")\n", ").agg(\n", " F.count(\"review.id\").alias(\"ReviewCount\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Rank\n", "# --------------------------------------------------\n", "\n", "final_df = movie_stats.orderBy(\n", " F.desc(\"ReviewCount\")\n", ").limit(10).select(\n", " F.col(\"title\"),\n", " F.col(\"ReviewCount\")\n", ")\n", "\n", "print(\"\\nMovies with the Most Reviews\\n\")\n", "measure(final_df).show(truncate=False)\n" ] }, { "attachments": {}, "cell_type": "markdown", "id": "fba5e1fd-6c3e-4593-aeed-d320cecbbdd1", "metadata": {}, "source": [ "### High-Rated Movies That Also Have Many Reviews" ] }, { "cell_type": "code", "execution_count": 16, "id": "dbb6535d-13b6-4c1b-901b-e61d51edee10", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "High-Rated Movies with Many Reviews\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " ]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Time taken: 4.115 seconds\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ "[Stage 60:=====================================================>(221 + 3) / 224]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+---------------------------------------------------------+-------+-----------+\n", "|Movie |Rating |ReviewCount|\n", "+---------------------------------------------------------+-------+-----------+\n", "|Lord of the Rings: The Fellowship of the Ring, The (2001)|4.09616|5122 |\n", "|Dark Knight, The (2008) |4.18379|5115 |\n", "|Shawshank Redemption, The (1994) |4.41985|4913 |\n", "|Matrix, The (1999) |4.15952|3765 |\n", "|Lord of the Rings: The Return of the King, The (2003) |4.10393|3275 |\n", "|Fight Club (1999) |4.23547|3082 |\n", "|Inception (2010) |4.17404|2967 |\n", "|Interstellar (2014) |4.12967|2962 |\n", "|American Beauty (1999) |4.10904|2757 |\n", "|Godfather, The (1972) |4.33078|2541 |\n", "+---------------------------------------------------------+-------+-----------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# --------------------------------------------------\n", "# Correct motif: (review)-[:has_review]->(movie)\n", "# --------------------------------------------------\n", "\n", "review_motif = g.find(\"\"\"\n", " (review)-[e]->(movie)\n", "\"\"\").filter(\n", " (F.col(\"e.etype\") == \"has_review\") &\n", " (F.col(\"review.type\") == \"review\") &\n", " (F.col(\"movie.type\") == \"movie\") &\n", " (F.col(\"movie.avgRating\").isNotNull())\n", ")\n", "\n", "# --------------------------------------------------\n", "# Aggregate review count per movie\n", "# --------------------------------------------------\n", "\n", "movie_stats = review_motif.groupBy(\n", " F.col(\"movie.id\").alias(\"MovieID\"),\n", " F.col(\"movie.title\").alias(\"Movie\"),\n", " F.col(\"movie.avgRating\").alias(\"Rating\")\n", ").agg(\n", " F.count(\"review.id\").alias(\"ReviewCount\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Apply thresholds and rank\n", "# --------------------------------------------------\n", "\n", "final_df = movie_stats.filter(\n", " (F.col(\"ReviewCount\") >= 10) &\n", " (F.col(\"Rating\") >= 4.0)\n", ").orderBy(\n", " F.desc(\"ReviewCount\")\n", ").limit(10).select(\n", " F.col(\"Movie\"),\n", " F.col(\"Rating\"),\n", " F.col(\"ReviewCount\")\n", ")\n", "\n", "print(\"\\nHigh-Rated Movies with Many Reviews\\n\")\n", "measure(final_df).show(truncate=False)\n" ] }, { "attachments": {}, "cell_type": "markdown", "id": "46ae0d00-8883-4a13-86fb-c410cfbbca7f", "metadata": {}, "source": [ "### Directors Whose Movies Collect the Most Reviews" ] }, { "cell_type": "code", "execution_count": 17, "id": "92cfe15f-91c2-4e3a-94bd-d54c023ab333", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Directors Whose Movies Collect the Most Reviews\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " ]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Time taken: 3.702 seconds\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] }, { "name": "stdout", "output_type": "stream", "text": [ "+------------------+------------+\n", "|Director |TotalReviews|\n", "+------------------+------------+\n", "|Steven Spielberg |23879 |\n", "|Christopher Nolan |21986 |\n", "|Peter Jackson |18849 |\n", "|Ridley Scott |17754 |\n", "|M. Night Shyamalan|13581 |\n", "|Martin Scorsese |13080 |\n", "|George Lucas |12557 |\n", "|Quentin Tarantino |12232 |\n", "|Michael Bay |11540 |\n", "|Zack Snyder |11508 |\n", "+------------------+------------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# --------------------------------------------------\n", "# Motif:\n", "# (director)-[:directed]->(movie)\n", "# (review)-[:has_review]->(movie)\n", "# --------------------------------------------------\n", "\n", "director_review_motif = g.find(\"\"\"\n", " (director)-[e1]->(movie);\n", " (review)-[e2]->(movie)\n", "\"\"\").filter(\n", " (F.col(\"e1.etype\") == \"directed\") &\n", " (F.col(\"e2.etype\") == \"has_review\") &\n", " (F.col(\"director.type\") == \"director\") &\n", " (F.col(\"movie.type\") == \"movie\") &\n", " (F.col(\"review.type\") == \"review\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Count reviews per director\n", "# --------------------------------------------------\n", "\n", "director_stats = director_review_motif.groupBy(\n", " F.col(\"director.id\").alias(\"DirectorID\"),\n", " F.col(\"director.name\").alias(\"Director\")\n", ").agg(\n", " F.count(\"review.id\").alias(\"TotalReviews\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Rank directors\n", "# --------------------------------------------------\n", "\n", "final_df = director_stats.orderBy(\n", " F.desc(\"TotalReviews\")\n", ").limit(10).select(\n", " F.col(\"Director\"),\n", " F.col(\"TotalReviews\")\n", ")\n", "\n", "print(\"\\nDirectors Whose Movies Collect the Most Reviews\\n\")\n", "measure(final_df).show(truncate=False)\n" ] }, { "attachments": {}, "cell_type": "markdown", "id": "63f15a2e-ddc1-4849-a3ed-66f59fa6b481", "metadata": {}, "source": [ "### Most frequently used words in reviews:" ] }, { "cell_type": "code", "execution_count": 18, "id": "7d577992-69f3-49f8-a01d-4daeebe1728a", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Most Frequently Used Words in Reviews\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] }, { "name": "stdout", "output_type": "stream", "text": [ "Time taken: 3.372 seconds\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ "[Stage 119:===================================================> (219 + 5) / 224]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+-----+-------+\n", "|word |freq |\n", "+-----+-------+\n", "|movie|5953237|\n", "|film |5080608|\n", "|like |2340766|\n", "|good |2005894|\n", "|about|1991607|\n", "|more |1865802|\n", "|some |1809667|\n", "|very |1700259|\n", "|story|1618946|\n", "|time |1508047|\n", "|well |1400216|\n", "|great|1386259|\n", "|their|1354898|\n", "|even |1290162|\n", "|will |1251124|\n", "|much |1234845|\n", "|only |1207987|\n", "|than |1202424|\n", "|also |1142701|\n", "|first|1115268|\n", "+-----+-------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# --------------------------------------------------\n", "# 1️⃣ Get review text\n", "# --------------------------------------------------\n", "\n", "reviews_df = g.vertices.filter(\n", " (F.col(\"type\") == \"review\") &\n", " (F.col(\"text\").isNotNull()) # change to reviewText if needed\n", ").select(\n", " F.lower(F.col(\"text\")).alias(\"text\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# 2️⃣ Split text into words\n", "# --------------------------------------------------\n", "\n", "words_df = reviews_df.select(\n", " F.explode(\n", " F.split(F.col(\"text\"), \"\\\\W+\")\n", " ).alias(\"word\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# 3️⃣ Remove short words and stopwords\n", "# --------------------------------------------------\n", "\n", "stopwords = [\n", " \"this\", \"that\", \"with\", \"from\", \"have\", \"your\",\n", " \"just\", \"what\", \"when\", \"they\", \"were\", \"then\",\n", " \"there\", \"would\", \"which\", \"could\", \"really\"\n", "]\n", "\n", "filtered_words = words_df.filter(\n", " (F.length(\"word\") > 3) &\n", " (~F.col(\"word\").isin(stopwords))\n", ")\n", "\n", "# --------------------------------------------------\n", "# 4️⃣ Count frequency\n", "# --------------------------------------------------\n", "\n", "result_df = filtered_words.groupBy(\"word\").agg(\n", " F.count(\"*\").alias(\"freq\")\n", ").orderBy(\n", " F.desc(\"freq\")\n", ").limit(20)\n", "\n", "print(\"\\nMost Frequently Used Words in Reviews\\n\")\n", "measure(result_df).show(truncate=False)\n" ] }, { "attachments": {}, "cell_type": "markdown", "id": "c269e77e-331a-4bd4-97c4-a7071f68617f", "metadata": {}, "source": [ "### Average Review Length per Movie (Which movies received the most detailed (or terse) reviews?)" ] }, { "cell_type": "code", "execution_count": 19, "id": "4c860502-303f-44dc-9e81-b5fbb435070f", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Average Review Length per Movie (Min 5 Reviews)\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] }, { "name": "stdout", "output_type": "stream", "text": [ "Time taken: 4.475 seconds\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ "[Stage 145:===============================================> (184 + 16) / 200]5]]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+-----------------------------------------------------------------+---------------+-----------+\n", "|Movie |AvgReviewLength|ReviewCount|\n", "+-----------------------------------------------------------------+---------------+-----------+\n", "|Garbage Warrior (2007) |4239.0 |9 |\n", "|The Crossing (2014) |4222.0 |6 |\n", "|Heaven's Heart (2008) |4111.77 |13 |\n", "|Helter Skelter (2012) |4100.0 |5 |\n", "|Making Plans for Lena (Non ma fille, tu n'iras pas danser) (2009)|3916.2 |5 |\n", "|Prescription Thugs (2016) |3702.2 |5 |\n", "|Kony 2012 (2012) |3674.4 |5 |\n", "|L'assedio dell'Alcazar (1940) |3640.2 |5 |\n", "|The Plague (2006) |3583.2 |5 |\n", "|Go for Sisters (2013) |3538.17 |6 |\n", "+-----------------------------------------------------------------+---------------+-----------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# --------------------------------------------------\n", "# Motif: (review)-[:has_review]->(movie)\n", "# --------------------------------------------------\n", "\n", "review_motif = g.find(\"\"\"\n", " (review)-[e]->(movie)\n", "\"\"\").filter(\n", " (F.col(\"e.etype\") == \"has_review\") &\n", " (F.col(\"review.type\") == \"review\") &\n", " (F.col(\"movie.type\") == \"movie\") &\n", " (F.col(\"review.text\").isNotNull()) # change to reviewText if needed\n", ")\n", "\n", "# --------------------------------------------------\n", "# Aggregate average review length per movie\n", "# --------------------------------------------------\n", "\n", "movie_stats = review_motif.groupBy(\n", " F.col(\"movie.id\").alias(\"MovieID\"),\n", " F.col(\"movie.title\").alias(\"Movie\")\n", ").agg(\n", " F.avg(F.length(F.col(\"review.text\"))).alias(\"AvgLength\"),\n", " F.count(\"review.id\").alias(\"ReviewCount\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Apply threshold and rank\n", "# --------------------------------------------------\n", "\n", "final_df = movie_stats.filter(\n", " F.col(\"ReviewCount\") >= 5\n", ").select(\n", " F.col(\"Movie\"),\n", " F.round(F.col(\"AvgLength\"), 2).alias(\"AvgReviewLength\"),\n", " F.col(\"ReviewCount\")\n", ").orderBy(\n", " F.desc(\"AvgReviewLength\")\n", ").limit(10)\n", "\n", "print(\"\\nAverage Review Length per Movie (Min 5 Reviews)\\n\")\n", "measure(final_df).show(truncate=False)\n" ] }, { "attachments": {}, "cell_type": "markdown", "id": "b178a793-6a57-4db6-8096-cbb9ab408538", "metadata": {}, "source": [ "### Movies Most Frequently Called \"Boring\", \"Disgusting\", etc." ] }, { "cell_type": "code", "execution_count": 20, "id": "90f1b20f-7699-428f-8a64-4af56b325973", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Movies Most Frequently Called Negative Terms\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] }, { "name": "stdout", "output_type": "stream", "text": [ "Time taken: 2.705 seconds\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ "[Stage 168:===========================================> (188 + 36) / 224]]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+---------------------------------------------------------+----------------+\n", "|Movie |NegativeMentions|\n", "+---------------------------------------------------------+----------------+\n", "|Star Wars: Episode VII - The Force Awakens (2015) |2118 |\n", "|Batman v Superman: Dawn of Justice (2016) |1703 |\n", "|Star Wars: Episode II - Attack of the Clones (2002) |1578 |\n", "|Star Wars: Episode I - The Phantom Menace (1999) |1534 |\n", "|Lord of the Rings: The Fellowship of the Ring, The (2001)|1491 |\n", "|Star Wars: Episode III - Revenge of the Sith (2005) |1390 |\n", "|Dark Knight, The (2008) |1382 |\n", "|Suicide Squad (2016) |1219 |\n", "|Avatar (2009) |1109 |\n", "|Spider-Man 3 (2007) |1084 |\n", "+---------------------------------------------------------+----------------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# --------------------------------------------------\n", "# Motif: (review)-[:has_review]->(movie)\n", "# --------------------------------------------------\n", "\n", "negative_motif = g.find(\"\"\"\n", " (review)-[e]->(movie)\n", "\"\"\").filter(\n", " (F.col(\"e.etype\") == \"has_review\") &\n", " (F.col(\"review.type\") == \"review\") &\n", " (F.col(\"movie.type\") == \"movie\") &\n", " (F.col(\"review.text\").isNotNull())\n", ")\n", "\n", "# --------------------------------------------------\n", "# Filter reviews containing negative keywords\n", "# --------------------------------------------------\n", "\n", "negative_words = [\"boring\", \"disgusting\", \"waste\", \"terrible\", \"bad\"]\n", "\n", "filtered_reviews = negative_motif.filter(\n", " F.lower(F.col(\"review.text\")).rlike(\n", " \"|\".join(negative_words)\n", " )\n", ")\n", "\n", "# --------------------------------------------------\n", "# Count negative mentions per movie\n", "# --------------------------------------------------\n", "\n", "movie_stats = filtered_reviews.groupBy(\n", " F.col(\"movie.id\").alias(\"MovieID\"),\n", " F.col(\"movie.title\").alias(\"Movie\")\n", ").agg(\n", " F.count(\"review.id\").alias(\"NegativeMentions\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Rank\n", "# --------------------------------------------------\n", "\n", "final_df = movie_stats.orderBy(\n", " F.desc(\"NegativeMentions\")\n", ").limit(10).select(\n", " F.col(\"Movie\"),\n", " F.col(\"NegativeMentions\")\n", ")\n", "\n", "print(\"\\nMovies Most Frequently Called Negative Terms\\n\")\n", "measure(final_df).show(truncate=False)\n" ] }, { "attachments": {}, "cell_type": "markdown", "id": "dd8286b2-6db2-4540-8803-d735d1ce726d", "metadata": {}, "source": [ "### ===SURVEY + TAG===\n", "\n", "### Top Tags by Frequency in Surveys (Which tags are evaluated most often by users?)" ] }, { "cell_type": "code", "execution_count": 21, "id": "ef2a711a-bf77-4fbd-b0c3-a56dfc349b9d", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Top Tags by Frequency in Surveys\n", "\n", "Time taken: 0.606 seconds\n", "+-----------+-----------+\n", "|Tag |SurveyCount|\n", "+-----------+-----------+\n", "|comedy |616 |\n", "|sci-fi |579 |\n", "|classic |549 |\n", "|action |510 |\n", "|time travel|388 |\n", "|surreal |384 |\n", "|fantasy |383 |\n", "|romance |381 |\n", "|animation |369 |\n", "|drugs |340 |\n", "+-----------+-----------+\n", "\n" ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# --------------------------------------------------\n", "# Motif: (survey)-[:about_tag]->(tag)\n", "# --------------------------------------------------\n", "\n", "survey_tag_motif = g.find(\"\"\"\n", " (survey)-[e]->(tag)\n", "\"\"\").filter(\n", " (F.col(\"e.etype\") == \"about_tag\") & # adjust if using 'relationship'\n", " (F.col(\"survey.type\") == \"survey\") &\n", " (F.col(\"tag.type\") == \"tag\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Count surveys per tag\n", "# --------------------------------------------------\n", "\n", "tag_stats = survey_tag_motif.groupBy(\n", " F.col(\"tag.id\").alias(\"TagID\"),\n", " F.col(\"tag.tag\").alias(\"Tag\")\n", ").agg(\n", " F.count(\"*\").alias(\"SurveyCount\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Rank tags\n", "# --------------------------------------------------\n", "\n", "final_df = tag_stats.orderBy(\n", " F.desc(\"SurveyCount\")\n", ").limit(10).select(\n", " F.col(\"Tag\"),\n", " F.col(\"SurveyCount\")\n", ")\n", "\n", "print(\"\\nTop Tags by Frequency in Surveys\\n\")\n", "measure(final_df).show(truncate=False)\n" ] }, { "attachments": {}, "cell_type": "markdown", "id": "10d03fde-6fe1-44e1-adac-80f51fe2c3a9", "metadata": {}, "source": [ "### Movies With Most Tagged Survey Feedback (Which movies have been analyzed most often in surveys?)" ] }, { "cell_type": "code", "execution_count": 22, "id": "17187632-b743-4f05-ab77-21e9606d8a9c", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Tags Most Frequently Scored ≥ 4 in Surveys\n", "\n", "Time taken: 0.618 seconds\n", "+-----------+--------------+\n", "|Tag |HighScoreCount|\n", "+-----------+--------------+\n", "|comedy |279 |\n", "|sci-fi |259 |\n", "|action |244 |\n", "|classic |241 |\n", "|fantasy |171 |\n", "|time travel|163 |\n", "|animation |157 |\n", "|romance |154 |\n", "|surreal |149 |\n", "|aliens |140 |\n", "+-----------+--------------+\n", "\n" ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# --------------------------------------------------\n", "# Motif: (survey)-[:about_tag]->(tag)\n", "# --------------------------------------------------\n", "\n", "survey_tag_motif = g.find(\"\"\"\n", " (survey)-[e]->(tag)\n", "\"\"\").filter(\n", " (F.col(\"e.etype\") == \"about_tag\") & # adjust if using 'relationship'\n", " (F.col(\"survey.type\") == \"survey\") &\n", " (F.col(\"tag.type\") == \"tag\") &\n", " (F.col(\"survey.score\") >= 4)\n", ")\n", "\n", "# --------------------------------------------------\n", "# Count high-score surveys per tag\n", "# --------------------------------------------------\n", "\n", "tag_stats = survey_tag_motif.groupBy(\n", " F.col(\"tag.id\").alias(\"TagID\"),\n", " F.col(\"tag.tag\").alias(\"Tag\")\n", ").agg(\n", " F.count(\"*\").alias(\"HighScoreCount\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Rank tags\n", "# --------------------------------------------------\n", "\n", "final_df = tag_stats.orderBy(\n", " F.desc(\"HighScoreCount\")\n", ").limit(10).select(\n", " F.col(\"Tag\"),\n", " F.col(\"HighScoreCount\")\n", ")\n", "\n", "print(\"\\nTags Most Frequently Scored ≥ 4 in Surveys\\n\")\n", "measure(final_df).show(truncate=False)\n" ] }, { "attachments": {}, "cell_type": "markdown", "id": "9cf73c6a-1893-4569-bb25-baed39192c71", "metadata": {}, "source": [ "### Best Tags per Movie by Survey and Tag_Count" ] }, { "cell_type": "code", "execution_count": 23, "id": "122d708d-727f-416b-8364-7b306b57ab60", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Best Tags per Movie by Survey and Tag_Count\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] }, { "name": "stdout", "output_type": "stream", "text": [ "Time taken: 1.871 seconds\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ "[Stage 238:> (0 + 2) / 2]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+-------------------------------------------------------------+---------------+---------------+--------------+--------------+\n", "|Movie |Tag |TagApplications|AvgSurveyScore|SurveyMentions|\n", "+-------------------------------------------------------------+---------------+---------------+--------------+--------------+\n", "|Back to the Future (1985) |time travel |3360 |5.0 |21 |\n", "|Lives of Others, The (Das leben der Anderen) (2006) |east germany |804 |5.0 |12 |\n", "|Pirates of the Caribbean: The Curse of the Black Pearl (2003)|pirates |1628 |5.0 |11 |\n", "|Schindler's List (1993) |holocaust |1903 |5.0 |11 |\n", "|Yellow Submarine (1968) |beatles |121 |5.0 |11 |\n", "|Pulp Fiction (1994) |tarantino |1190 |5.0 |10 |\n", "|Slumdog Millionaire (2008) |india |760 |5.0 |10 |\n", "|Godfather, The (1972) |mafia |2142 |5.0 |9 |\n", "|Fahrenheit 9/11 (2004) |documentary |279 |5.0 |9 |\n", "|Monty Python's Life of Brian (1979) |british comedy |324 |5.0 |9 |\n", "|Monty Python and the Holy Grail (1975) |british comedy |945 |5.0 |9 |\n", "|Godfather: Part II, The (1974) |mafia |774 |5.0 |9 |\n", "|Hard Day's Night, A (1964) |beatles |180 |5.0 |9 |\n", "|Star Trek II: The Wrath of Khan (1982) |star trek |264 |5.0 |8 |\n", "|Goodfellas (1990) |mafia |1120 |5.0 |8 |\n", "|League of Their Own, A (1992) |baseball |128 |5.0 |8 |\n", "|Lord of the Rings: The Return of the King, The (2003) |fantasy |1656 |5.0 |8 |\n", "|Amelie (Fabuleux destin d'Amélie Poulain, Le) (2001) |france |384 |5.0 |8 |\n", "|Hannibal (2001) |hannibal lecter|128 |5.0 |8 |\n", "|Back to the Future Part III (1990) |time travel |512 |5.0 |8 |\n", "+-------------------------------------------------------------+---------------+---------------+--------------+--------------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# --------------------------------------------------\n", "# 3-edge motif:\n", "# (movie)-[:has_tag]->(tag)\n", "# (survey)-[:about_movie]->(movie)\n", "# (survey)-[:about_tag]->(tag)\n", "# --------------------------------------------------\n", "\n", "motif = g.find(\"\"\"\n", " (movie)-[ht]->(tag);\n", " (survey)-[am]->(movie);\n", " (survey)-[at]->(tag)\n", "\"\"\").filter(\n", " (F.col(\"ht.etype\") == \"has_tag\") &\n", " (F.col(\"am.etype\") == \"about_movie\") &\n", " (F.col(\"at.etype\") == \"about_tag\") &\n", " (F.col(\"movie.type\") == \"movie\") &\n", " (F.col(\"tag.type\") == \"tag\") &\n", " (F.col(\"survey.type\") == \"survey\") &\n", " (F.col(\"survey.score\").isNotNull())\n", ")\n", "\n", "# --------------------------------------------------\n", "# Aggregate metrics per (movie, tag)\n", "# --------------------------------------------------\n", "\n", "stats = motif.groupBy(\n", " F.col(\"movie.id\").alias(\"MovieID\"),\n", " F.col(\"movie.title\").alias(\"Movie\"),\n", " F.col(\"tag.id\").alias(\"TagID\"),\n", " F.col(\"tag.tag\").alias(\"Tag\")\n", ").agg(\n", " F.sum(\"ht.count\").alias(\"TagApplications\"),\n", " F.avg(\"survey.score\").alias(\"AvgSurveyScore\"),\n", " F.count(\"survey.id\").alias(\"SurveyMentions\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Apply thresholds and rank\n", "# --------------------------------------------------\n", "\n", "final_df = stats.filter(\n", " (F.col(\"TagApplications\") > 5) &\n", " (F.col(\"SurveyMentions\") >= 5)\n", ").select(\n", " F.col(\"Movie\"),\n", " F.col(\"Tag\"),\n", " F.col(\"TagApplications\"),\n", " F.round(F.col(\"AvgSurveyScore\"), 2).alias(\"AvgSurveyScore\"),\n", " F.col(\"SurveyMentions\")\n", ").orderBy(\n", " F.desc(\"AvgSurveyScore\"),\n", " F.desc(\"SurveyMentions\")\n", ").limit(20)\n", "\n", "print(\"\\nBest Tags per Movie by Survey and Tag_Count\\n\")\n", "measure(final_df).show(truncate=False)\n" ] }, { "attachments": {}, "cell_type": "markdown", "id": "b6497dca-95bf-487c-9c9e-36fb65524f41", "metadata": {}, "source": [ "### Users Who Participated in the Most Surveys" ] }, { "cell_type": "code", "execution_count": 24, "id": "8cb915cc-8370-48a6-8f80-89ae5f465960", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Users Who Participated in the Most Surveys\n", "\n", "Time taken: 0.958 seconds\n", "+------+------------+\n", "|User |TotalSurveys|\n", "+------+------------+\n", "|627711|246 |\n", "|978707|246 |\n", "|185112|246 |\n", "|184449|246 |\n", "|704376|246 |\n", "|641649|246 |\n", "|754076|246 |\n", "|939084|246 |\n", "|445909|246 |\n", "|570456|246 |\n", "+------+------------+\n", "\n" ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# --------------------------------------------------\n", "# Motif: (user)-[:answered]->(survey)\n", "# --------------------------------------------------\n", "\n", "answered_motif = g.find(\"\"\"\n", " (user)-[e]->(survey)\n", "\"\"\").filter(\n", " (F.col(\"e.etype\") == \"answered\") & # adjust if using 'relationship'\n", " (F.col(\"user.type\") == \"user\") &\n", " (F.col(\"survey.type\") == \"survey\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Count surveys per user\n", "# --------------------------------------------------\n", "\n", "user_stats = answered_motif.groupBy(\n", " F.col(\"user.id\").alias(\"User\")\n", ").agg(\n", " F.count(\"survey.id\").alias(\"TotalSurveys\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Rank users\n", "# --------------------------------------------------\n", "\n", "final_df = user_stats.orderBy(\n", " F.desc(\"TotalSurveys\")\n", ").limit(10)\n", "\n", "print(\"\\nUsers Who Participated in the Most Surveys\\n\")\n", "measure(final_df).show(truncate=False)\n" ] }, { "attachments": {}, "cell_type": "markdown", "id": "f359dd88-c7a2-4d5a-a5a7-93c54bde95f1", "metadata": {}, "source": [ "### Tag-Movie Associations with High Confidence (Which movie-tag pairs were most confidently agreed upon by users?)" ] }, { "cell_type": "code", "execution_count": 25, "id": "d9cf607a-33e9-4cf6-969a-74e5d135e194", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Tag-Movie Associations with High Confidence\n", "\n", "Time taken: 0.951 seconds\n", "+---------------------------------------------------+------------+--------------+\n", "|Movie |Tag |HighScoreCount|\n", "+---------------------------------------------------+------------+--------------+\n", "|Back to the Future (1985) |time travel |21 |\n", "|Serenity (2005) |firefly |16 |\n", "|Monty Python and the Holy Grail (1975) |monty python|14 |\n", "|This Is Spinal Tap (1984) |mockumentary|14 |\n", "|Monty Python's Life of Brian (1979) |monty python|14 |\n", "|Terminator 2: Judgment Day (1991) |time travel |13 |\n", "|Shawshank Redemption, The (1994) |prison |13 |\n", "|E.T. the Extra-Terrestrial (1982) |sci-fi |13 |\n", "|Hotel Rwanda (2004) |genocide |12 |\n", "|Lives of Others, The (Das leben der Anderen) (2006)|east germany|12 |\n", "+---------------------------------------------------+------------+--------------+\n", "\n" ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# --------------------------------------------------\n", "# Motif:\n", "# (survey)-[:about_movie]->(movie)\n", "# (survey)-[:about_tag]->(tag)\n", "# --------------------------------------------------\n", "\n", "confidence_motif = g.find(\"\"\"\n", " (survey)-[am]->(movie);\n", " (survey)-[at]->(tag)\n", "\"\"\").filter(\n", " (F.col(\"am.etype\") == \"about_movie\") &\n", " (F.col(\"at.etype\") == \"about_tag\") &\n", " (F.col(\"survey.type\") == \"survey\") &\n", " (F.col(\"movie.type\") == \"movie\") &\n", " (F.col(\"tag.type\") == \"tag\") &\n", " (F.col(\"survey.score\") >= 4)\n", ")\n", "\n", "# --------------------------------------------------\n", "# Count high-score associations per (movie, tag)\n", "# --------------------------------------------------\n", "\n", "stats = confidence_motif.groupBy(\n", " F.col(\"movie.id\").alias(\"MovieID\"),\n", " F.col(\"movie.title\").alias(\"Movie\"),\n", " F.col(\"tag.id\").alias(\"TagID\"),\n", " F.col(\"tag.tag\").alias(\"Tag\")\n", ").agg(\n", " F.count(\"survey.id\").alias(\"HighScoreCount\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# Rank associations\n", "# --------------------------------------------------\n", "\n", "final_df = stats.orderBy(\n", " F.desc(\"HighScoreCount\")\n", ").limit(10).select(\n", " F.col(\"Movie\"),\n", " F.col(\"Tag\"),\n", " F.col(\"HighScoreCount\")\n", ")\n", "\n", "print(\"\\nTag-Movie Associations with High Confidence\\n\")\n", "measure(final_df).show(truncate=False)\n" ] }, { "attachments": {}, "cell_type": "markdown", "id": "f8b9c9bf-478d-4bf0-922b-7df8e70e5151", "metadata": {}, "source": [ "### Users Who Did NOT Participate in Any Survey" ] }, { "cell_type": "code", "execution_count": 26, "id": "c5a3abfd-a383-4dd6-a070-fd22c94b5182", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Users Who Did NOT Participate in Any Survey\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] }, { "name": "stdout", "output_type": "stream", "text": [ "Time taken: 2.514 seconds\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ "[Stage 309:===================================================> (217 + 7) / 224]" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+------+\n", "|UserID|\n", "+------+\n", "|2904 |\n", "|768779|\n", "|536596|\n", "|234938|\n", "|767839|\n", "|213151|\n", "|721429|\n", "|327044|\n", "|311799|\n", "|651662|\n", "|819769|\n", "|609272|\n", "|687192|\n", "|764131|\n", "|218214|\n", "|772621|\n", "|867736|\n", "|836024|\n", "|635130|\n", "|435697|\n", "+------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " " ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# --------------------------------------------------\n", "# 1️⃣ All users\n", "# --------------------------------------------------\n", "\n", "users_df = g.vertices.filter(\n", " F.col(\"type\") == \"user\"\n", ").select(\n", " F.col(\"id\").alias(\"UserID\")\n", ")\n", "\n", "# --------------------------------------------------\n", "# 2️⃣ Users who answered at least one survey\n", "# --------------------------------------------------\n", "\n", "answered_motif = g.find(\"\"\"\n", " (user)-[e]->(survey)\n", "\"\"\").filter(\n", " (F.col(\"e.etype\") == \"answered\") &\n", " (F.col(\"user.type\") == \"user\") &\n", " (F.col(\"survey.type\") == \"survey\")\n", ")\n", "\n", "answered_users = answered_motif.select(\n", " F.col(\"user.id\").alias(\"UserID\")\n", ").distinct()\n", "\n", "# --------------------------------------------------\n", "# 3️⃣ Anti-join to find users with NO survey participation\n", "# --------------------------------------------------\n", "\n", "inactive_users = users_df.join(\n", " answered_users,\n", " \"UserID\",\n", " \"left_anti\"\n", ").limit(20)\n", "\n", "print(\"\\nUsers Who Did NOT Participate in Any Survey\\n\")\n", "measure(inactive_users).show(truncate=False)\n" ] }, { "attachments": {}, "cell_type": "markdown", "id": "0cce99fc-aa96-4457-944c-2cca3bfacb90", "metadata": {}, "source": [ "### Users With Highest Survey Agreement Accuracy (Measures users whose survey scores closely align with consensus (like those who gave 5 when others did too))." ] }, { "cell_type": "code", "execution_count": 27, "id": "ce4942c0-f76d-4a9e-9a6a-fb5668fc2f83", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", "Users With Highest Survey Agreement Accuracy\n", "\n", "Time taken: 1.515 seconds\n", "+------+------------------+-----+\n", "|User |AvgTagDisagreement|Count|\n", "+------+------------------+-----+\n", "|229012|0.81 |11 |\n", "|160799|0.95 |12 |\n", "|24783 |1.01 |60 |\n", "|31073 |1.03 |24 |\n", "|786743|1.05 |48 |\n", "|101793|1.05 |102 |\n", "|573078|1.08 |47 |\n", "|487921|1.08 |240 |\n", "|525534|1.09 |48 |\n", "|649965|1.09 |48 |\n", "+------+------------------+-----+\n", "\n" ] } ], "source": [ "from pyspark.sql import functions as F\n", "\n", "# Tag consensus: average score per tag\n", "tag_consensus_motif = g.find(\"\"\"\n", " (survey)-[e]->(tag)\n", "\"\"\").filter(\n", " (F.col(\"e.etype\") == \"about_tag\") &\n", " (F.col(\"survey.type\") == \"survey\") &\n", " (F.col(\"tag.type\") == \"tag\") &\n", " (F.col(\"survey.score\").isNotNull())\n", ")\n", "\n", "tag_consensus = tag_consensus_motif.groupBy(\n", " F.col(\"tag.id\").alias(\"TagID\")\n", ").agg(\n", " F.avg(\"survey.score\").alias(\"Consensus\")\n", ")\n", "\n", "user_tag_motif = g.find(\"\"\"\n", " (user)-[a]->(survey);\n", " (survey)-[b]->(tag)\n", "\"\"\").filter(\n", " (F.col(\"a.etype\") == \"answered\") &\n", " (F.col(\"b.etype\") == \"about_tag\") &\n", " (F.col(\"user.type\") == \"user\") &\n", " (F.col(\"survey.type\") == \"survey\") &\n", " (F.col(\"tag.type\") == \"tag\") &\n", " (F.col(\"survey.score\").isNotNull())\n", ")\n", "\n", "user_tag_with_consensus = user_tag_motif.join(\n", " tag_consensus,\n", " F.col(\"tag.id\") == F.col(\"TagID\"),\n", " \"inner\"\n", ")\n", "\n", "user_stats = user_tag_with_consensus.groupBy(\n", " F.col(\"user.id\").alias(\"User\")\n", ").agg(\n", " F.avg(F.abs(F.col(\"survey.score\") - F.col(\"Consensus\"))).alias(\"AvgDeviation\"),\n", " F.count(\"*\").alias(\"Count\")\n", ")\n", "\n", "final_df = user_stats.filter(\n", " F.col(\"Count\") >= 10\n", ").select(\n", " F.col(\"User\"),\n", " F.round(F.col(\"AvgDeviation\"), 2).alias(\"AvgTagDisagreement\"),\n", " F.col(\"Count\")\n", ").orderBy(\n", " F.asc(\"AvgTagDisagreement\")\n", ").limit(10)\n", "\n", "print(\"\\nUsers With Highest Survey Agreement Accuracy\\n\")\n", "measure(final_df).show(truncate=False)\n", "\n" ] }, { "cell_type": "code", "execution_count": null, "id": "6e49643a-6b2a-4f63-9c96-1a74e4a55dc5", "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.10.19" } }, "nbformat": 4, "nbformat_minor": 5 }