{ "cells": [ { "cell_type": "code", "execution_count": 138, "id": "c5dff16d-6e70-43f3-af66-4dafed945509", "metadata": { "scrolled": true }, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Requirement already satisfied: findspark in ./anaconda3/lib/python3.12/site-packages (2.0.1)\n" ] } ], "source": [ "!pip install findspark" ] }, { "cell_type": "code", "execution_count": 139, "id": "980a1569-462c-427e-b098-1f57d2a80614", "metadata": { "scrolled": true }, "outputs": [ { "data": { "text/plain": [ "'/shared/spark-3.4.4-bin-hadoop3'" ] }, "execution_count": 139, "metadata": {}, "output_type": "execute_result" } ], "source": [ "import findspark\n", "findspark.init()\n", "findspark.find()" ] }, { "cell_type": "markdown", "id": "1a46564e-cc4b-42fd-b2f4-4d51766bee9f", "metadata": {}, "source": [ "# Vertex Creation" ] }, { "cell_type": "code", "execution_count": 140, "id": "355fa966-dd7c-4590-912a-2cba16d2c85e", "metadata": {}, "outputs": [], "source": [ "# Import PySpark\n", "from pyspark.sql import SparkSession\n", "\n", "#Create SparkSession\n", "# spark = SparkSession.builder.appName('yelp').config(\"spark.executor.memory\", \"4g\").config(\"spark.driver.memory\", \"4g\").getOrCreate()\n", "\n", "num_executors = \"4\"\n", "spark = SparkSession.builder.appName('movielens').config(\"spark.executor.memory\", \"4g\").config(\"spark.driver.memory\", \"4g\") \\\n", " .config(\"spark.executor.instances\", num_executors).getOrCreate()" ] }, { "cell_type": "code", "execution_count": 198, "id": "a4a108ec-11fe-439b-bf43-37816788a44f", "metadata": {}, "outputs": [], "source": [ "movies_df = spark.read.option(\"header\", True).csv(\"hdfs://master:9000/user/exouser/ml-32m/movies.csv\")\n", "ratings_df = spark.read.option(\"header\", True).csv(\"hdfs://master:9000/user/exouser/ml-32m/ratings.csv\")\n", "tags_df = spark.read.option(\"header\", True).csv(\"hdfs://master:9000/user/exouser/ml-32m/tags.csv\")\n", "links_df = spark.read.option(\"header\", True).csv(\"hdfs://master:9000/user/exouser/ml-32m/links.csv\")\n" ] }, { "cell_type": "code", "execution_count": 199, "id": "cda9b858-6f88-4c5f-ada3-fd36851619ab", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+-------+--------------------+--------------------+\n", "|movieId| title| genres|\n", "+-------+--------------------+--------------------+\n", "| 1| Toy Story (1995)|Adventure|Animati...|\n", "| 2| Jumanji (1995)|Adventure|Childre...|\n", "| 3|Grumpier Old Men ...| Comedy|Romance|\n", "| 4|Waiting to Exhale...|Comedy|Drama|Romance|\n", "| 5|Father of the Bri...| Comedy|\n", "+-------+--------------------+--------------------+\n", "only showing top 5 rows\n", "\n" ] } ], "source": [ "movies_df.show(5)\n" ] }, { "cell_type": "code", "execution_count": 255, "id": "3a22deac-ebe1-4f2a-b116-c7826228f0de", "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", "+-------+-------+------+\n", "only showing top 5 rows\n", "\n" ] } ], "source": [ "links_df.show(5)" ] }, { "cell_type": "code", "execution_count": 256, "id": "62a37b73-2d6c-4651-9272-caaf7b9c6502", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+------+-------+------+---------+\n", "|userId|movieId|rating|timestamp|\n", "+------+-------+------+---------+\n", "| 1| 17| 4.0|944249077|\n", "| 1| 25| 1.0|944250228|\n", "| 1| 29| 2.0|943230976|\n", "| 1| 30| 5.0|944249077|\n", "| 1| 32| 5.0|943228858|\n", "+------+-------+------+---------+\n", "only showing top 5 rows\n", "\n" ] } ], "source": [ "ratings_df.show(5)" ] }, { "cell_type": "code", "execution_count": 254, "id": "52104445-8a30-40a8-ae25-53c06dccc56b", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+------+-------+-----------+----------+\n", "|userId|movieId| tag| timestamp|\n", "+------+-------+-----------+----------+\n", "| 22| 26479|Kevin Kline|1583038886|\n", "| 22| 79592| misogyny|1581476297|\n", "| 22| 247150| acrophobia|1622483469|\n", "| 34| 2174| music|1249808064|\n", "| 34| 2174| weird|1249808102|\n", "+------+-------+-----------+----------+\n", "only showing top 5 rows\n", "\n" ] } ], "source": [ "tags_df.show(5)" ] }, { "cell_type": "code", "execution_count": 200, "id": "2df0eec5-582a-497d-9fa6-c43a633db265", "metadata": {}, "outputs": [ { "data": { "text/plain": [ "87585" ] }, "execution_count": 200, "metadata": {}, "output_type": "execute_result" } ], "source": [ "from pyspark.sql.functions import col\n", "\n", "clean_movies_df = movies_df.filter(col(\"movieId\").isNotNull() & col(\"title\").isNotNull())\n", "clean_movies_df.count()" ] }, { "cell_type": "code", "execution_count": 201, "id": "0e7c377b-027f-448c-aeaa-095d1750c348", "metadata": {}, "outputs": [], "source": [ "## without this parsing was not done properly" ] }, { "cell_type": "code", "execution_count": 202, "id": "3a0dd364-840d-4848-bedb-fd00bf9157a7", "metadata": {}, "outputs": [], "source": [ "movies_df = spark.read.option(\"header\", True) \\\n", " .option(\"quote\", \"\\\"\") \\\n", " .option(\"escape\", \"\\\"\") \\\n", " .csv(\"hdfs://master:9000/user/exouser/ml-32m/movies.csv\")\n" ] }, { "cell_type": "code", "execution_count": 356, "id": "93c2caa9-0e3e-468a-8a95-cb50b1b34a67", "metadata": {}, "outputs": [ { "data": { "text/plain": [ "0" ] }, "execution_count": 356, "metadata": {}, "output_type": "execute_result" } ], "source": [ "## This is to check if all genre entry exist\n", "clean_movies_df = movies_df.filter(col(\"movieId\").isNotNull() & col(\"title\").isNotNull())\n", "\n", "movie_genres_df = clean_movies_df \\\n", " .filter((col(\"genres\").isNull()) | (col(\"genres\") == \"\")) \n", "movie_genres_df.count()" ] }, { "cell_type": "code", "execution_count": 357, "id": "9b67cde7-e7d9-44be-8f2e-b6fcf22fe6ca", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+------------------+\n", "|genre |\n", "+------------------+\n", "|(no genres listed)|\n", "|Action |\n", "|Adventure |\n", "|Animation |\n", "|Children |\n", "|Comedy |\n", "|Crime |\n", "|Documentary |\n", "|Drama |\n", "|Fantasy |\n", "|Film-Noir |\n", "|Horror |\n", "|IMAX |\n", "|Musical |\n", "|Mystery |\n", "|Romance |\n", "|Sci-Fi |\n", "|Thriller |\n", "|War |\n", "|Western |\n", "+------------------+\n", "\n" ] } ], "source": [ "clean_movies_df = movies_df.filter(col(\"movieId\").isNotNull() & col(\"title\").isNotNull())\n", "\n", "movie_genres_df = clean_movies_df \\\n", " .filter((col(\"genres\").isNotNull())) \\\n", " .withColumn(\"genre\", explode(split(col(\"genres\"), \"\\\\|\"))) \\\n", " .withColumn(\"genre\", trim(col(\"genre\"))) \\\n", " .filter((col(\"genre\") != \"\") & (col(\"genre\").isNotNull()))\n", "\n", "movie_genres_df.select(\"genre\").distinct().orderBy(\"genre\").show(50, truncate=False)\n" ] }, { "cell_type": "code", "execution_count": 267, "id": "33e33f17-41dd-432e-890f-1e0e7b88961b", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ " \r" ] }, { "data": { "text/plain": [ "32000204" ] }, "execution_count": 267, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# non null entries of ratings\n", "from pyspark.sql.functions import to_timestamp\n", "\n", "clean_ratings_df = ratings_df \\\n", " .filter(col(\"userId\").isNotNull() & col(\"movieId\").isNotNull() & col(\"rating\").isNotNull()) \\\n", " .withColumn(\"userId\", col(\"userId\").cast(\"int\")) \\\n", " .withColumn(\"movieId\", col(\"movieId\").cast(\"int\")) \\\n", " .withColumn(\"rating\", col(\"rating\").cast(\"float\")) \\\n", " .withColumn(\"timestamp\", to_timestamp(col(\"timestamp\").cast(\"long\")))\n", "\n", "clean_ratings_df.count()\n" ] }, { "cell_type": "code", "execution_count": 268, "id": "ef2ac1d4-0941-487c-844b-b32fcbd8a354", "metadata": {}, "outputs": [ { "data": { "text/plain": [ "2000072" ] }, "execution_count": 268, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# Non null & only valid id of movie, user exits in tags\n", "clean_tags_df = tags_df \\\n", " .filter(col(\"userId\").isNotNull() & col(\"movieId\").isNotNull() & col(\"tag\").isNotNull()) \\\n", " .filter(trim(col(\"tag\")) != \"\") \\\n", " .withColumn(\"userId\", col(\"userId\").cast(\"int\")) \\\n", " .withColumn(\"movieId\", col(\"movieId\").cast(\"int\")) \\\n", " .withColumn(\"timestamp\", to_timestamp(col(\"timestamp\").cast(\"long\")))\n", "\n", "clean_tags_df.count()\n" ] }, { "cell_type": "code", "execution_count": 273, "id": "d60b9f42-170e-4cb9-8875-d14723dae2a7", "metadata": {}, "outputs": [ { "data": { "text/plain": [ "2000072" ] }, "execution_count": 273, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# movieIds must exist in clean_movies_df\n", "valid_movie_ids = clean_movies_df.select(col(\"movieId\").cast(\"int\")).distinct()\n", "valid_user_ids = clean_ratings_df.select(col(\"userId\").cast(\"int\")).distinct()\n", "\n", "tags_matched_movies = clean_tags_df.join(valid_movie_ids, on=\"movieId\", how=\"inner\")\n", "tags_matched_movies.count()\n", "\n", "\n" ] }, { "cell_type": "code", "execution_count": 274, "id": "84a0326e-6518-458c-b0df-fc6cd848c264", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ " \r" ] }, { "data": { "text/plain": [ "2000072" ] }, "execution_count": 274, "metadata": {}, "output_type": "execute_result" } ], "source": [ "tags_matched_users = clean_tags_df.join(valid_user_ids, on=\"userId\", how=\"inner\")\n", "tags_matched_users.count()" ] }, { "cell_type": "code", "execution_count": 389, "id": "4db65631-7fca-47c9-ac55-d5f7260fe18a", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ " \r" ] }, { "data": { "text/plain": [ "2000072" ] }, "execution_count": 389, "metadata": {}, "output_type": "execute_result" } ], "source": [ "#I had to do this because neo4j input was made to have only one entry for (userId, MovieId)\n", "\n", "# MATCH (u:User)-[t:TAGGED]->(m:Movie)\n", "# WITH u.userId AS userId, m.movieId AS movieId, COUNT(*) AS tagCount\n", "# WHERE tagCount > 1\n", "# RETURN userId, movieId, tagCount\n", "# LIMIT 10;\n", "\n", "# Can run above code to check that there will be no entries that exceed 1 in neo4j\n", "\n", "tags_deduped = tags_matched_movies\n", "\n", "tags_deduped.count()\n" ] }, { "cell_type": "code", "execution_count": 278, "id": "e1502178-f0e7-4d93-a44b-6d2a9764e1c9", "metadata": {}, "outputs": [ { "data": { "text/plain": [ "87585" ] }, "execution_count": 278, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# Value matches with neo4j vertices\n", "\n", "# MATCH (m:Movie)\n", "# RETURN COUNT(m.movieId) AS movieCount;\n", "\n", "movie_vertices = clean_movies_df \\\n", " .select(col(\"movieId\").cast(\"string\").alias(\"id\"), \"title\") \\\n", " .distinct()\n", "\n", "movie_vertices.count()\n" ] }, { "cell_type": "code", "execution_count": 283, "id": "a3c8cbcd-0ae1-434b-bd4e-1cfcf56210df", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ " \r" ] }, { "data": { "text/plain": [ "200948" ] }, "execution_count": 283, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# Same Value with neo4j\n", "\n", "# MATCH (u:User)\n", "# RETURN COUNT(u.userId) AS userCount;\n", "\n", "user_ids_from_ratings = clean_ratings_df.select(col(\"userId\").cast(\"string\").alias(\"id\"))\n", "user_ids_from_tags = tags_deduped.select(col(\"userId\").cast(\"string\").alias(\"id\"))\n", "\n", "user_vertices = user_ids_from_ratings.union(user_ids_from_tags).distinct()\n", "\n", "user_vertices.count()\n" ] }, { "cell_type": "code", "execution_count": 293, "id": "18972fd3-5bb5-4dc9-a003-69bbf77cd847", "metadata": {}, "outputs": [ { "data": { "text/plain": [ "20" ] }, "execution_count": 293, "metadata": {}, "output_type": "execute_result" } ], "source": [ "# same value with neo4j\n", "\n", "# MATCH (g:Genre)\n", "# RETURN COUNT(g.name) AS genreCount;\n", "\n", "genre_vertices = movie_genres_df \\\n", " .select(trim(col(\"genre\")).alias(\"id\")) \\\n", " .distinct()\n", "genre_vertices.count()" ] }, { "cell_type": "code", "execution_count": 296, "id": "065b93fc-fd8b-4262-a49d-b4a636ac4105", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "+------------------+\n", "| id|\n", "+------------------+\n", "| Crime|\n", "| Romance|\n", "| Thriller|\n", "| Adventure|\n", "| Drama|\n", "| War|\n", "| Documentary|\n", "| Fantasy|\n", "| Mystery|\n", "| Musical|\n", "| Animation|\n", "| Film-Noir|\n", "|(no genres listed)|\n", "| IMAX|\n", "| Horror|\n", "| Western|\n", "| Comedy|\n", "| Children|\n", "| Action|\n", "| Sci-Fi|\n", "+------------------+\n", "\n" ] } ], "source": [ "genre_vertices.show()" ] }, { "cell_type": "code", "execution_count": null, "id": "9d77157f-8497-454b-a6a1-e954369ea915", "metadata": {}, "outputs": [], "source": [] }, { "cell_type": "code", "execution_count": 390, "id": "19f9c909-d425-4b62-8794-2137e64246fe", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ " \r" ] }, { "data": { "text/plain": [ "0" ] }, "execution_count": 390, "metadata": {}, "output_type": "execute_result" } ], "source": [ "movie_vertices.select(\"id\").intersect(user_vertices.select(\"id\")).count()\n" ] }, { "cell_type": "code", "execution_count": 301, "id": "e9957f65-771d-4c94-aa9f-c90c85d9480e", "metadata": {}, "outputs": [], "source": [ "### This is to make all id's different by adding prefix\n", "\n", "from pyspark.sql.functions import concat, lit, col\n", "\n", "movie_vertices = clean_movies_df.select(\n", " concat(lit(\"m_\"), col(\"movieId\").cast(\"string\")).alias(\"id\"),\n", " col(\"title\")\n", ")\n", "\n", "user_vertices = ratings_df.select(\n", " concat(lit(\"u_\"), col(\"userId\").cast(\"string\")).alias(\"id\")\n", ").distinct()\n", "\n", "genre_vertices = movie_genres_df.select(\n", " concat(lit(\"g_\"), col(\"genre\")).alias(\"id\")\n", ").distinct()\n" ] }, { "cell_type": "code", "execution_count": 391, "id": "e17a778e-beca-4a91-af80-53e4d533b186", "metadata": {}, "outputs": [], "source": [ "vertices = movie_vertices.unionByName(user_vertices, allowMissingColumns=True) \\\n", " .unionByName(genre_vertices, allowMissingColumns=True) \\\n", ".dropDuplicates([\"id\"])\n" ] }, { "cell_type": "code", "execution_count": 392, "id": "b14b8048-7953-4e69-91a4-7dbebc9dcd79", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ " \r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Total vertices: 288553\n", "Movies: 87585\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " \r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Users: 200948\n", "Genres: 20\n" ] } ], "source": [ "print(\"Total vertices:\", vertices.count())\n", "print(\"Movies:\", movie_vertices.count())\n", "print(\"Users:\", user_vertices.count())\n", "print(\"Genres:\", genre_vertices.count())\n" ] }, { "cell_type": "markdown", "id": "b86d6079-a05c-4641-9f99-e51640a72c3e", "metadata": {}, "source": [ "# Create Edges" ] }, { "cell_type": "code", "execution_count": 304, "id": "d02df970-af69-4e7d-83f5-e16ec6f0b0e2", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ " \r" ] }, { "data": { "text/plain": [ "32000204" ] }, "execution_count": 304, "metadata": {}, "output_type": "execute_result" } ], "source": [ "from pyspark.sql.functions import lit, col, concat\n", "\n", "rated_edges = clean_ratings_df \\\n", " .select(\n", " concat(lit(\"u_\"), col(\"userId\").cast(\"string\")).alias(\"src\"),\n", " concat(lit(\"m_\"), col(\"movieId\").cast(\"string\")).alias(\"dst\"),\n", " \"rating\", \"timestamp\"\n", " ) \\\n", " .withColumn(\"relationship\", lit(\"RATED\"))\n", "\n", "rated_edges.count()\n" ] }, { "cell_type": "code", "execution_count": 305, "id": "7e8b14d0-b1f0-493d-8d37-873a23b19034", "metadata": {}, "outputs": [ { "data": { "text/plain": [ "350042" ] }, "execution_count": 305, "metadata": {}, "output_type": "execute_result" } ], "source": [ "\n", "tagged_edges = tags_deduped \\\n", " .select(\n", " concat(lit(\"u_\"), col(\"userId\").cast(\"string\")).alias(\"src\"),\n", " concat(lit(\"m_\"), col(\"movieId\").cast(\"string\")).alias(\"dst\"),\n", " \"tag\", \"timestamp\"\n", " ) \\\n", " .withColumn(\"relationship\", lit(\"TAGGED\"))\n", "\n", "tagged_edges.count()" ] }, { "cell_type": "code", "execution_count": 393, "id": "0c4976a1-b644-4ced-b007-20df6f6abfec", "metadata": {}, "outputs": [ { "data": { "text/plain": [ "2000072" ] }, "execution_count": 393, "metadata": {}, "output_type": "execute_result" } ], "source": [ "## new tagged edges count without dropping\n", "tagged_edges = tags_deduped \\\n", " .select(\n", " concat(lit(\"u_\"), col(\"userId\").cast(\"string\")).alias(\"src\"),\n", " concat(lit(\"m_\"), col(\"movieId\").cast(\"string\")).alias(\"dst\"),\n", " \"tag\", \"timestamp\"\n", " ) \\\n", " .withColumn(\"relationship\", lit(\"TAGGED\"))\n", "\n", "tagged_edges.count()" ] }, { "cell_type": "code", "execution_count": 307, "id": "dc76ac5f-6f89-4d9c-a022-eb8ab23e2dc9", "metadata": {}, "outputs": [ { "data": { "text/plain": [ "154170" ] }, "execution_count": 307, "metadata": {}, "output_type": "execute_result" } ], "source": [ "from pyspark.sql.functions import concat, lit, trim, col\n", "\n", "has_genre_edges = movie_genres_df \\\n", " .select(\n", " concat(lit(\"m_\"), col(\"movieId\").cast(\"string\")).alias(\"src\"),\n", " concat(lit(\"g_\"), trim(col(\"genre\"))).alias(\"dst\")\n", " ) \\\n", " .withColumn(\"relationship\", lit(\"HAS_GENRE\"))\n", "has_genre_edges.count()" ] }, { "cell_type": "code", "execution_count": 309, "id": "807f62e3-d15e-4f3c-959d-ce4f04be2a41", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ " \r" ] }, { "data": { "text/plain": [ "32504416" ] }, "execution_count": 309, "metadata": {}, "output_type": "execute_result" } ], "source": [ "edges = rated_edges.unionByName(tagged_edges, allowMissingColumns=True) \\\n", " .unionByName(has_genre_edges, allowMissingColumns=True)\n", "edges.count()" ] }, { "cell_type": "code", "execution_count": 394, "id": "f784f48a-2fb7-46b0-b502-fb317341476d", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ " \r" ] }, { "data": { "text/plain": [ "34154446" ] }, "execution_count": 394, "metadata": {}, "output_type": "execute_result" } ], "source": [ "## new edges withouth dropping duplicate tags\n", "edges = rated_edges.unionByName(tagged_edges, allowMissingColumns=True) \\\n", " .unionByName(has_genre_edges, allowMissingColumns=True)\n", "edges.count()" ] }, { "cell_type": "code", "execution_count": 395, "id": "220ae5e2-3e1b-4c8c-b5c7-2ffbf73e0779", "metadata": {}, "outputs": [], "source": [ "SparkSession.getActiveSession = staticmethod(lambda: spark)\n" ] }, { "cell_type": "code", "execution_count": 396, "id": "8306ff58-df4d-4465-807c-ce7395314b35", "metadata": {}, "outputs": [], "source": [ "from graphframes import GraphFrames\n", "\n", "g = GraphFrames(vertices, edges)\n" ] }, { "cell_type": "code", "execution_count": 312, "id": "a3901a3d-5476-42bf-951e-3ce04b14c4e3", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ " \r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Total vertices: 288553\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " \r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Total edges: 32504416\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " \r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+------------+--------+\n", "|relationship| count|\n", "+------------+--------+\n", "| RATED|32000204|\n", "| TAGGED| 350042|\n", "| HAS_GENRE| 154170|\n", "+------------+--------+\n", "\n" ] } ], "source": [ "print(\"Total vertices:\", g.vertices.count())\n", "print(\"Total edges:\", g.edges.count())\n", "g.edges.groupBy(\"relationship\").count().show()\n" ] }, { "cell_type": "code", "execution_count": 399, "id": "e689103a-df86-4687-98b3-0438d6dab08b", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ " \r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Total vertices in new: 288553\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " \r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Total edges in new: 34154446\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ "[Stage 5051:====================================================> (51 + 1) / 52]\r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+------------+--------+\n", "|relationship| count|\n", "+------------+--------+\n", "| RATED|32000204|\n", "| TAGGED| 2000072|\n", "| HAS_GENRE| 154170|\n", "+------------+--------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " \r" ] } ], "source": [ "## new graph\n", "print(\"Total vertices in new:\", g.vertices.count())\n", "print(\"Total edges in new:\", g.edges.count())\n", "g.edges.groupBy(\"relationship\").count().show()" ] }, { "cell_type": "code", "execution_count": 314, "id": "cd7d7f1a-aea7-474f-bfba-9ffda398c3ac", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ " \r" ] }, { "data": { "text/plain": [ "0" ] }, "execution_count": 314, "metadata": {}, "output_type": "execute_result" } ], "source": [ "#Now no intersection\n", "movie_vertices.select(\"id\").intersect(user_vertices.select(\"id\")).count()\n" ] }, { "cell_type": "code", "execution_count": 315, "id": "38790c24-66cf-402a-96ac-8535fcc16838", "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Movies: 87585\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " \r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "Users: 200948\n", "Genres: 20\n" ] } ], "source": [ "print(\"Movies:\", movie_vertices.count())\n", "print(\"Users:\", user_vertices.count())\n", "print(\"Genres:\", genre_vertices.count())\n" ] }, { "cell_type": "code", "execution_count": 370, "id": "7664bc14-2ad3-462d-a81e-8e16c278c67e", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "[Stage 4595:==========================================> (3 + 1) / 4]\r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+-----------+--------------------+\n", "| id| title|\n", "+-----------+--------------------+\n", "|g_Adventure| null|\n", "| g_Crime| null|\n", "| g_IMAX| null|\n", "| m_100034| Girl Model (2011)|\n", "| m_100056| Die (2010)|\n", "| m_100060|Sunny (Sseo-ni) (...|\n", "| m_100062|My Way (Mai Wei) ...|\n", "| m_100072|Metsän tarina (2012)|\n", "| m_100101| Jean Gentil (2010)|\n", "| m_100155|Green Wave, The (...|\n", "| m_100194|Jim Jefferies: Fu...|\n", "| m_100203| Sundome (2007)|\n", "| m_100246| Pretty Sweet (2012)|\n", "| m_100257|Confessions from ...|\n", "| m_100270|Love Loves Coinci...|\n", "| m_100275| Return (2011)|\n", "| m_100291|Ballplayer: Pelot...|\n", "| m_100315|2 Become 1 (Tin s...|\n", "| m_100324| LUV (2012)|\n", "| m_100368|No Man of Her Own...|\n", "+-----------+--------------------+\n", "only showing top 20 rows\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " \r" ] } ], "source": [ "vertices.show()" ] }, { "cell_type": "markdown", "id": "65c46f68-e82c-4959-b0ee-c978d0cbeb5f", "metadata": {}, "source": [ "# Queries" ] }, { "cell_type": "code", "execution_count": 368, "id": "d32abe67-d965-41c7-9c3a-92e795a90725", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ " \r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+--------------------+---------+\n", "|genre |frequency|\n", "+--------------------+---------+\n", "|g_Drama |34175 |\n", "|g_Comedy |23124 |\n", "|g_Thriller |11823 |\n", "|g_Romance |10369 |\n", "|g_Action |9668 |\n", "|g_Documentary |9363 |\n", "|g_Horror |8654 |\n", "|g_(no genres listed)|7080 |\n", "|g_Crime |6976 |\n", "|g_Adventure |5402 |\n", "|g_Sci-Fi |4907 |\n", "|g_Animation |4617 |\n", "|g_Children |4520 |\n", "|g_Mystery |4013 |\n", "|g_Fantasy |3851 |\n", "|g_War |2325 |\n", "|g_Western |1696 |\n", "|g_Musical |1059 |\n", "|g_Film-Noir |353 |\n", "|g_IMAX |195 |\n", "+--------------------+---------+\n", "\n" ] } ], "source": [ "### Genre Usage Summary (Validation Query)\n", "\n", "g.find(\"(m)-[e]->(g)\") \\\n", " .filter(\"e.relationship = 'HAS_GENRE'\") \\\n", " .groupBy(\"g.id\") \\\n", " .count() \\\n", " .withColumnRenamed(\"id\", \"genre\") \\\n", " .withColumnRenamed(\"count\", \"frequency\") \\\n", " .orderBy(\"frequency\", ascending=False) \\\n", " .show(20, truncate=False)\n", "\n", "# Da Yan note: .withColumnRenamed(\"g.id\", \"genre\") => .withColumnRenamed(\"id\", \"genre\")" ] }, { "cell_type": "code", "execution_count": 318, "id": "ea015bc3-8c1b-4cc7-abe7-708c3464f5c7", "metadata": {}, "outputs": [], "source": [ "#Query - 1: Check different rating values and their occurrence frequencies" ] }, { "cell_type": "code", "execution_count": 369, "id": "09b96a8d-990d-432d-9f8f-0e8087545318", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "[Stage 4587:===============================================> (28 + 4) / 32]\r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+------+---------+\n", "|rating|frequency|\n", "+------+---------+\n", "| 0.5| 525132|\n", "| 1.0| 946675|\n", "| 1.5| 531063|\n", "| 2.0| 2028622|\n", "| 2.5| 1685386|\n", "| 3.0| 6054990|\n", "| 3.5| 4290105|\n", "| 4.0| 8367654|\n", "| 4.5| 2974000|\n", "| 5.0| 4596577|\n", "+------+---------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " \r" ] } ], "source": [ "g.find(\"(u)-[r]->(m)\") \\\n", " .filter(\"r.relationship = 'RATED'\") \\\n", " .groupBy(\"r.rating\") \\\n", " .count() \\\n", " .withColumnRenamed(\"count\", \"frequency\") \\\n", " .orderBy(\"rating\") \\\n", " .show()\n" ] }, { "cell_type": "code", "execution_count": 320, "id": "31960b74-ab5a-4901-8821-ea326204e1c7", "metadata": {}, "outputs": [], "source": [ "#Query - 2: List users and the movies they rated that belong to the specific genre \"Comedy\", find top-20 results in descending order of ratings:" ] }, { "cell_type": "code", "execution_count": 371, "id": "1565692f-74f8-4bef-bec5-29d1bbcff867", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "[Stage 4625:======================================> (23 + 9) / 32]\r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+--------+--------+---------------------------------------+------+--------+\n", "|userId |movieId |title |rating|genre |\n", "+--------+--------+---------------------------------------+------+--------+\n", "|u_64276 |m_177545|#realityhigh (2017) |5.0 |g_Comedy|\n", "|u_148479|m_143681|'49-'17 (1917) |5.0 |g_Comedy|\n", "|u_100357|m_97757 |'Hellboy': The Seeds of Creation (2004)|5.0 |g_Comedy|\n", "|u_13501 |m_97757 |'Hellboy': The Seeds of Creation (2004)|5.0 |g_Comedy|\n", "|u_155483|m_97757 |'Hellboy': The Seeds of Creation (2004)|5.0 |g_Comedy|\n", "|u_95523 |m_97757 |'Hellboy': The Seeds of Creation (2004)|5.0 |g_Comedy|\n", "|u_28362 |m_97757 |'Hellboy': The Seeds of Creation (2004)|5.0 |g_Comedy|\n", "|u_88196 |m_97757 |'Hellboy': The Seeds of Creation (2004)|5.0 |g_Comedy|\n", "|u_87874 |m_97757 |'Hellboy': The Seeds of Creation (2004)|5.0 |g_Comedy|\n", "|u_41450 |m_268482|'Tis the Season to be Merry (2021) |5.0 |g_Comedy|\n", "|u_25741 |m_2072 |'burbs, The (1989) |5.0 |g_Comedy|\n", "|u_73522 |m_2072 |'burbs, The (1989) |5.0 |g_Comedy|\n", "|u_22718 |m_2072 |'burbs, The (1989) |5.0 |g_Comedy|\n", "|u_150728|m_2072 |'burbs, The (1989) |5.0 |g_Comedy|\n", "|u_24286 |m_2072 |'burbs, The (1989) |5.0 |g_Comedy|\n", "|u_73181 |m_2072 |'burbs, The (1989) |5.0 |g_Comedy|\n", "|u_20841 |m_2072 |'burbs, The (1989) |5.0 |g_Comedy|\n", "|u_45790 |m_2072 |'burbs, The (1989) |5.0 |g_Comedy|\n", "|u_20994 |m_2072 |'burbs, The (1989) |5.0 |g_Comedy|\n", "|u_73427 |m_2072 |'burbs, The (1989) |5.0 |g_Comedy|\n", "+--------+--------+---------------------------------------+------+--------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " \r" ] } ], "source": [ "top_comedy_ratings = g.find(\"(u)-[r]->(m); (m)-[h]->(g)\") \\\n", " .filter(\"r.relationship = 'RATED' AND h.relationship = 'HAS_GENRE' AND g.id = 'g_Comedy'\") \\\n", " .select(\n", " col(\"u.id\").alias(\"userId\"),\n", " col(\"m.id\").alias(\"movieId\"),\n", " col(\"m.title\").alias(\"title\"),\n", " col(\"r.rating\").alias(\"rating\"),\n", " col(\"g.id\").alias(\"genre\")\n", " ) \\\n", " .orderBy(col(\"rating\").desc(), col(\"title\").asc()) \\\n", " .limit(20)\n", "\n", "top_comedy_ratings.show(truncate=False)\n" ] }, { "cell_type": "code", "execution_count": 326, "id": "dd542682-08ad-47a7-b677-8568db93d20e", "metadata": {}, "outputs": [], "source": [ "# Query - 3: Find the title of the movies that belong to multiple genres, e.g., \"Comedy\" and \"Romance\":" ] }, { "cell_type": "code", "execution_count": 376, "id": "b60c6355-1c46-44de-bcc4-41ce8b5d611f", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ " \r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+----------------------------------+\n", "|title |\n", "+----------------------------------+\n", "|#FriendButMarried (2018) |\n", "|#Iamhere (2020) |\n", "|#Roxy (2018) |\n", "|#Selfie69 (2016) |\n", "|#Stuck (2014) |\n", "|#Xmas (2022) |\n", "|'Neath Brooklyn Bridge (1942) |\n", "|'Tis the Season to be Merry (2021)|\n", "|(500) Days of Summer (2009) |\n", "|(Girl)Friend (2018) |\n", "+----------------------------------+\n", "\n" ] } ], "source": [ "# Find movies that are connected to both Comedy and Romance\n", "both_genres = g.find(\"(m)-[e1]->(g1); (m)-[e2]->(g2)\") \\\n", " .filter(\"\"\"\n", " e1.relationship = 'HAS_GENRE' AND g1.id = 'g_Comedy' AND\n", " e2.relationship = 'HAS_GENRE' AND g2.id = 'g_Romance'\n", " \"\"\") \\\n", " .select(col(\"m.title\").alias(\"title\")) \\\n", " .distinct() \\\n", " .orderBy(\"title\") \\\n", " .limit(10)\n", "\n", "both_genres.show(truncate=False)\n" ] }, { "cell_type": "code", "execution_count": 329, "id": "315598cc-459e-4767-87e3-e77d87f42118", "metadata": {}, "outputs": [], "source": [ "# Query - 4: Find users who have rated movies in both \"Action\" and \"Romance\" genres, list 10 userIds:\n", "\n", "# takes at least 20 min!!!" ] }, { "cell_type": "code", "execution_count": 411, "id": "ee05018b-6d56-424b-a90d-d21c75edfa0c", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "[Stage 5224:====================================================> (33 + 1) / 34]\r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+------+\n", "|userId|\n", "+------+\n", "|u_1 |\n", "|u_2 |\n", "|u_3 |\n", "|u_5 |\n", "|u_6 |\n", "|u_7 |\n", "|u_8 |\n", "|u_9 |\n", "|u_10 |\n", "|u_11 |\n", "+------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " \r" ] } ], "source": [ "from pyspark.sql.functions import col\n", "\n", "action_romance_users = g.find(\"(u)-[r1]->(m1); (u)-[r2]->(m2); (m1)-[g1edge]->(g1); (m2)-[g2edge]->(g2)\") \\\n", " .filter(\"\"\"\n", " r1.relationship = 'RATED' AND\n", " r2.relationship = 'RATED' AND\n", " g1edge.relationship = 'HAS_GENRE' AND g1.id = 'g_Action' AND\n", " g2edge.relationship = 'HAS_GENRE' AND g2.id = 'g_Romance'\n", " \"\"\") \\\n", " .select(col(\"u.id\").alias(\"userId\")) \\\n", " .distinct() \\\n", " .orderBy(\n", " regexp_replace(col(\"userId\"), \"u_\", \"\").cast(\"int\").asc()\n", " ) \\\n", " .limit(10)\n", "\n", "action_romance_users.show(truncate=False)\n" ] }, { "cell_type": "code", "execution_count": null, "id": "148090ed-22de-4a46-ad76-ca6bc1f4fb70", "metadata": {}, "outputs": [], "source": [ "#Query - 5: Find users who have rated movies from at least 3 distinct genres, list the top-10 in descending order of the genre count." ] }, { "cell_type": "code", "execution_count": 331, "id": "d2a5c193-aaf4-4541-9c8e-c2bce047b9bf", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "[Stage 4240:============================================> (28 + 6) / 34]\r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+------+----------+\n", "|userId|genreCount|\n", "+------+----------+\n", "|u_16 |20 |\n", "|u_28 |20 |\n", "|u_35 |20 |\n", "|u_76 |20 |\n", "|u_94 |20 |\n", "|u_213 |20 |\n", "|u_228 |20 |\n", "|u_245 |20 |\n", "|u_258 |20 |\n", "|u_265 |20 |\n", "+------+----------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " \r" ] } ], "source": [ "from pyspark.sql.functions import countDistinct, col, regexp_replace\n", "\n", "user_genre_counts = g.find(\"(u)-[r]->(m); (m)-[grel]->(g)\") \\\n", " .filter(\"r.relationship = 'RATED' AND grel.relationship = 'HAS_GENRE'\") \\\n", " .select(\n", " col(\"u.id\").alias(\"userId\"),\n", " col(\"g.id\").alias(\"genreId\")\n", " ) \\\n", " .distinct() \\\n", " .groupBy(\"userId\") \\\n", " .agg(countDistinct(\"genreId\").alias(\"genreCount\")) \\\n", " .filter(col(\"genreCount\") >= 3) \\\n", " .orderBy(\n", " col(\"genreCount\").desc(),\n", " regexp_replace(col(\"userId\"), \"u_\", \"\").cast(\"int\").asc()\n", " ) \\\n", " .limit(10)\n", "\n", "user_genre_counts.show(truncate=False)\n" ] }, { "cell_type": "code", "execution_count": null, "id": "2d04a880-150c-4acd-bc97-f7d6d477e630", "metadata": {}, "outputs": [], "source": [ "#Query - 6: Find users who have rated movies from at least 3 distinct genres, list 10 such users along with the corresponding list of genres in descending order of genre count." ] }, { "cell_type": "code", "execution_count": 332, "id": "6a2be1d2-a7e9-46e2-bea6-2382b1499dcb", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "[Stage 4253:==========================================> (27 + 7) / 34]\r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n", "|userId|genres |\n", "+------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n", "|u_16 |[g_Mystery, g_(no genres listed), g_Horror, g_Action, g_Romance, g_Drama, g_Adventure, g_IMAX, g_Fantasy, g_War, g_Thriller, g_Sci-Fi, g_Film-Noir, g_Crime, g_Comedy, g_Documentary, g_Western, g_Animation, g_Children, g_Musical]|\n", "|u_28 |[g_Horror, g_(no genres listed), g_Musical, g_Action, g_Romance, g_Drama, g_Adventure, g_IMAX, g_Fantasy, g_War, g_Thriller, g_Sci-Fi, g_Film-Noir, g_Comedy, g_Crime, g_Documentary, g_Western, g_Children, g_Mystery, g_Animation]|\n", "|u_35 |[g_Horror, g_(no genres listed), g_Animation, g_Action, g_Romance, g_Drama, g_Adventure, g_IMAX, g_Fantasy, g_War, g_Thriller, g_Sci-Fi, g_Film-Noir, g_Crime, g_Comedy, g_Documentary, g_Western, g_Musical, g_Mystery, g_Children]|\n", "|u_76 |[g_Musical, g_(no genres listed), g_Horror, g_Action, g_Romance, g_Drama, g_Adventure, g_IMAX, g_Fantasy, g_War, g_Thriller, g_Sci-Fi, g_Film-Noir, g_Comedy, g_Crime, g_Documentary, g_Western, g_Children, g_Mystery, g_Animation]|\n", "|u_94 |[g_Mystery, g_(no genres listed), g_Children, g_Action, g_Romance, g_Drama, g_Adventure, g_IMAX, g_Fantasy, g_War, g_Thriller, g_Sci-Fi, g_Film-Noir, g_Comedy, g_Crime, g_Documentary, g_Western, g_Horror, g_Musical, g_Animation]|\n", "|u_213 |[g_Horror, g_(no genres listed), g_Animation, g_Action, g_Romance, g_Drama, g_Adventure, g_IMAX, g_Fantasy, g_War, g_Thriller, g_Sci-Fi, g_Film-Noir, g_Crime, g_Comedy, g_Documentary, g_Western, g_Musical, g_Mystery, g_Children]|\n", "|u_228 |[g_Children, g_(no genres listed), g_Animation, g_Action, g_Romance, g_Drama, g_Adventure, g_IMAX, g_Fantasy, g_War, g_Thriller, g_Sci-Fi, g_Film-Noir, g_Comedy, g_Crime, g_Documentary, g_Western, g_Musical, g_Mystery, g_Horror]|\n", "|u_245 |[g_Horror, g_(no genres listed), g_Musical, g_Action, g_Romance, g_Drama, g_Adventure, g_IMAX, g_Fantasy, g_War, g_Thriller, g_Sci-Fi, g_Film-Noir, g_Comedy, g_Crime, g_Documentary, g_Western, g_Children, g_Mystery, g_Animation]|\n", "|u_258 |[g_Animation, g_(no genres listed), g_Musical, g_Action, g_Romance, g_Drama, g_Adventure, g_IMAX, g_Fantasy, g_War, g_Thriller, g_Sci-Fi, g_Film-Noir, g_Comedy, g_Crime, g_Documentary, g_Western, g_Horror, g_Mystery, g_Children]|\n", "|u_265 |[g_Musical, g_(no genres listed), g_Horror, g_Action, g_Romance, g_Drama, g_Adventure, g_IMAX, g_Fantasy, g_War, g_Thriller, g_Sci-Fi, g_Film-Noir, g_Comedy, g_Crime, g_Documentary, g_Western, g_Children, g_Mystery, g_Animation]|\n", "+------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " \r" ] } ], "source": [ "from pyspark.sql.functions import collect_set, size, col, regexp_replace\n", "\n", "g.find(\"(u)-[r]->(m); (m)-[grel]->(g)\") \\\n", " .filter(\"r.relationship = 'RATED' AND grel.relationship = 'HAS_GENRE'\") \\\n", " .select(col(\"u.id\").alias(\"userId\"), col(\"g.id\").alias(\"genre\")) \\\n", " .distinct() \\\n", " .groupBy(\"userId\") \\\n", " .agg(collect_set(\"genre\").alias(\"genres\")) \\\n", " .filter(size(col(\"genres\")) >= 3) \\\n", " .orderBy(\n", " size(col(\"genres\")).desc(),\n", " regexp_replace(col(\"userId\"), \"u_\", \"\").cast(\"int\").asc()\n", " ) \\\n", " .limit(10) \\\n", " .show(truncate=False)\n" ] }, { "cell_type": "code", "execution_count": null, "id": "b4d21517-2d86-473f-9044-b2c57e7d700a", "metadata": {}, "outputs": [], "source": [ "# Query - 7: Find movies that span at least 3 genres and have at least 50 user ratings, return top-10 movies in descending order of their rating count, along with the list of genres and the rating count." ] }, { "cell_type": "code", "execution_count": 333, "id": "323d3ba9-a58b-40b4-9a02-82a6e48bbe33", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "[Stage 4272:====================================================> (31 + 1) / 32]\r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+-------+--------------------------------+----------------------------------------------------------------------+-----------+\n", "|movieId|movie |genres |ratingCount|\n", "+-------+--------------------------------+----------------------------------------------------------------------+-----------+\n", "|m_79132|Inception (2010) |[g_Sci-Fi, g_Action, g_Drama, g_Crime, g_IMAX, g_Thriller, g_Mystery] |405517 |\n", "|m_356 |Forrest Gump (1994) |[g_Romance, g_Drama, g_Comedy, g_War] |401184 |\n", "|m_296 |Pulp Fiction (1994) |[g_Drama, g_Crime, g_Comedy, g_Thriller] |393636 |\n", "|m_1 |Toy Story (1995) |[g_Comedy, g_Adventure, g_Fantasy, g_Children, g_Animation] |344985 |\n", "|m_4306 |Shrek (2001) |[g_Romance, g_Comedy, g_Adventure, g_Fantasy, g_Children, g_Animation]|329064 |\n", "|m_2959 |Fight Club (1999) |[g_Action, g_Drama, g_Crime, g_Thriller] |309328 |\n", "|m_364 |Lion King, The (1994) |[g_Drama, g_Adventure, g_IMAX, g_Musical, g_Children, g_Animation] |309108 |\n", "|m_480 |Jurassic Park (1993) |[g_Sci-Fi, g_Action, g_Adventure, g_Thriller] |300932 |\n", "|m_2571 |Matrix, The (1999) |[g_Sci-Fi, g_Action, g_Thriller] |281424 |\n", "|m_593 |Silence of the Lambs, The (1991)|[g_Crime, g_Thriller, g_Horror] |270990 |\n", "+-------+--------------------------------+----------------------------------------------------------------------+-----------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " \r" ] } ], "source": [ "from pyspark.sql.functions import collect_set, size, count, col\n", "\n", "# Step 1: Pattern match (User)-[:RATED]->(Movie)-[:HAS_GENRE]->(Genre)\n", "pattern_df = g.find(\"(u)-[r]->(m); (m)-[h]->(g)\") \\\n", " .filter(\"r.relationship = 'RATED' AND h.relationship = 'HAS_GENRE'\") \\\n", " .select(\n", " col(\"m.id\").alias(\"movieId\"),\n", " col(\"m.title\").alias(\"movie\"), # assuming title is already in the vertex\n", " col(\"g.id\").alias(\"genre\"),\n", " col(\"r.rating\").alias(\"rating\")\n", " )\n", "\n", "# Step 2: Group by movie, aggregate genres and rating count\n", "pattern_df.groupBy(\"movieId\", \"movie\") \\\n", " .agg(\n", " collect_set(\"genre\").alias(\"genres\"),\n", " count(\"rating\").alias(\"ratingCount\")\n", " ) \\\n", " .filter((size(col(\"genres\")) >= 3) & (col(\"ratingCount\") >= 50)) \\\n", " .orderBy(col(\"ratingCount\").desc()) \\\n", " .limit(10) \\\n", " .show(truncate=False)\n" ] }, { "cell_type": "code", "execution_count": 382, "id": "23577563-ee5e-413c-bc0a-293f09bc76ff", "metadata": {}, "outputs": [], "source": [ "#Query - 8: Find (user, movie) pairs where the movie is first rated by the user, and then tagged by the user, return 10 such user-movie pairs with the latest timestamps, along with the rating and the list of tags (tagged after the rating).\n" ] }, { "cell_type": "code", "execution_count": 410, "id": "60336401-bb5b-40a3-b835-84bb70b99d87", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "[Stage 5196:==========================> (17 + 17) / 34]\r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+--------+------------------------------+------+-------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n", "|movieId |title |rating|ratingTime |tags |\n", "+--------+------------------------------+------+-------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n", "|m_275243|Elvis (2022) |3.5 |2023-10-12 23:55:50|[1960s, Baz Luhrmann, Austin Butler, biographical, elvis presley, 1970s, rock and roll, based on a true story, biopic] |\n", "|m_132502|Diamond Head (1963) |3.5 |2023-10-12 22:05:56|[sibling relationship, Hawaii] |\n", "|m_183959|Tom Segura: Disgraceful (2018)|4.0 |2023-10-12 14:41:36|[racial humor, seen 2023] |\n", "|m_229125|The Mole Agent (2020) |4.0 |2023-10-12 06:58:32|[ageing, loneliness, chile] |\n", "|m_175303|It (2017) |3.5 |2023-10-12 05:06:42|[coming of age, clown, seen 2023, gore, underground, small town, violence, horror, murder, children, monster, Maine, sewer, fantasy, evil clown, dismemberment, based on novel or book, remake, graphic, missing child, serial killer] |\n", "|m_289689|Strays (2023) |4.0 |2023-10-12 04:57:26|[foul language, violence, crude humor, dogs, adventure, drugs, talking animals, animal attack, revenge, talking animal, abusive relationship, abandonment, vulgar, mushrooms, allegory, dark comedy, lost child, hallucination, Will Ferrell, terrier, seen 2023, marijuana, stray dog, toxic relationship, anthropomorphic animal, crude, reunion, heavy handed, deadbeat, heavy-handed, Border Terrier, heavy-handed message, hallucinogenic drug, mutilation]|\n", "|m_63131 |Role Models (2008) |3.0 |2023-10-12 04:50:25|[coming of age, marriage proposal, seen 2023, life lessons, foul language, comedy, mentorship, break-up, crude humor, role-playing game, critical parents, Nudity (Topless), adult child friendship, community service, LARP, salesperson, geeks, vulgar child, Elizabeth Banks, role playing game, energy drink, dead-end job, foul-mouth child, salesman] |\n", "|m_2084 |Newsies (1992) |4.0 |2023-10-12 02:01:41|[boy singer, newspaper, New York city , musical, Orphans, Teenagers ] |\n", "|m_169168|Girlfriend's Day (2017) |2.0 |2023-10-11 22:11:06|[Bob Odenkirk] |\n", "|m_83613 |Cowboys & Aliens (2011) |4.0 |2023-10-11 12:21:59|[better than expected, hillarious] |\n", "+--------+------------------------------+------+-------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " \r" ] } ], "source": [ "from pyspark.sql.functions import col, collect_set\n", "\n", "# Step 1: Match (User)-[:RATED]->(Movie)<-[:TAGGED]-(User) and filter on timestamp\n", "pattern_df = g.find(\"(u)-[r]->(m); (u)-[t]->(m)\") \\\n", " .filter(\"r.relationship = 'RATED' AND t.relationship = 'TAGGED' AND r.timestamp < t.timestamp\") \\\n", " .select(\n", " col(\"u.id\").alias(\"userId\"),\n", " col(\"m.id\").alias(\"movieId\"),\n", " col(\"r.rating\").alias(\"rating\"),\n", " col(\"r.timestamp\").alias(\"ratingTime\"),\n", " col(\"t.tag\").alias(\"tag\")\n", " )\n", "\n", "# Step 2: Aggregate tags per user, movie, rating\n", "aggregated = pattern_df.groupBy(\"userId\", \"movieId\", \"rating\", \"ratingTime\") \\\n", " .agg(collect_set(\"tag\").alias(\"tags\"))\n", "\n", "# Step 3: Join with movie titles\n", "movie_titles = movie_vertices.select(\n", " col(\"id\").cast(\"string\").alias(\"movieId\"),\n", " col(\"title\")\n", ")\n", "\n", "final = aggregated.join(movie_titles, on=\"movieId\", how=\"left\") \\\n", " .select(\"movieId\", \"title\", \"rating\", \"ratingTime\", \"tags\") \\\n", " .orderBy(col(\"ratingTime\").desc()) \\\n", " .limit(10)\n", "\n", "final.show(truncate=False)" ] }, { "cell_type": "code", "execution_count": 335, "id": "ca9981f1-7a69-48e1-9380-d3be3a5823b8", "metadata": {}, "outputs": [], "source": [ "## Reason for mismatch in tags is, I kept only one tag per user to movie to follow what is being done in neo4j DB\n", "\n", "# MATCH ()-[t:TAGGED]->()\n", "# WHERE t.tag IS NOT NULL\n", "# RETURN COUNT(t) AS tagCount;\n", "\n", "# Run the above command in neo4j db to see the count it will be 350042. \n", "# only one tag is preserved in both neo4j and spark so it may or may not be same." ] }, { "cell_type": "code", "execution_count": 336, "id": "fa879a0c-77f0-4b33-a00c-496c0e8649af", "metadata": {}, "outputs": [], "source": [ "# Query - 9: Find the top-50 most frequent tags, return the tags along with their frequencies." ] }, { "cell_type": "code", "execution_count": 406, "id": "68626c90-e74c-486a-ab7f-3d0da381e988", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "[Stage 5110:=============================================> (15 + 3) / 18]\r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+------------------+---------+\n", "|tag |frequency|\n", "+------------------+---------+\n", "|sci-fi |10996 |\n", "|atmospheric |9589 |\n", "|action |8473 |\n", "|comedy |8139 |\n", "|funny |7467 |\n", "|surreal |7231 |\n", "|visually appealing|7090 |\n", "|based on a book |6617 |\n", "|twist ending |6521 |\n", "|dark comedy |6053 |\n", "|thought-provoking |6000 |\n", "|dystopia |5646 |\n", "|violence |5389 |\n", "|cinematography |5277 |\n", "|romance |5271 |\n", "|murder |5180 |\n", "|social commentary |5152 |\n", "|stylized |4917 |\n", "|fantasy |4844 |\n", "|psychology |4827 |\n", "+------------------+---------+\n", "only showing top 20 rows\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " \r" ] } ], "source": [ "from pyspark.sql.functions import col, count\n", "\n", "# Top 50 most frequent non-null tags\n", "g.find(\"(u)-[t]->(m)\") \\\n", " .filter(\"t.relationship = 'TAGGED' AND t.tag IS NOT NULL\") \\\n", " .select(col(\"t.tag\").alias(\"tag\")) \\\n", " .groupBy(\"tag\") \\\n", " .agg(count(\"*\").alias(\"frequency\")) \\\n", " .orderBy(col(\"frequency\").desc()) \\\n", " .limit(50) \\\n", " .show(truncate=False)\n" ] }, { "cell_type": "code", "execution_count": 340, "id": "07cb2cab-f289-4d71-a032-c6ce1f9cc68a", "metadata": {}, "outputs": [], "source": [ "#Query - 10: Find top-20 movies with the highest number of genres + tags, return this number, list of their genres, and list of their tags" ] }, { "cell_type": "code", "execution_count": 407, "id": "b58c56e1-5117-457a-85e0-e12d7bb03ed5", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "[Stage 5130:===> (1 + 17) / 18]\r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+-----------------------------------------------------+-----+---------------------------------------------------+\n", "|title |total|genres |\n", "+-----------------------------------------------------+-----+---------------------------------------------------+\n", "|Star Wars: Episode IV - A New Hope (1977) |1114 |[g_Sci-Fi, g_Action, g_Adventure] |\n", "|Pulp Fiction (1994) |948 |[g_Drama, g_Crime, g_Comedy, g_Thriller] |\n", "|Avengers: Infinity War - Part II (2019) |944 |[g_Sci-Fi, g_Action, g_Adventure] |\n", "|Guardians of the Galaxy 2 (2017) |870 |[g_Sci-Fi, g_Action, g_Adventure] |\n", "|Wonder Woman (2017) |860 |[g_Action, g_Adventure, g_Fantasy] |\n", "|Dawn of the Dead (1978) |856 |[g_Action, g_Drama, g_Horror] |\n", "|The Batman (2022) |837 |[g_Action, g_Drama, g_Crime] |\n", "|Terminator 2: Judgment Day (1991) |831 |[g_Sci-Fi, g_Action] |\n", "|Black Panther (2017) |822 |[g_Sci-Fi, g_Action, g_Adventure] |\n", "|Avengers: Infinity War - Part I (2018) |777 |[g_Sci-Fi, g_Action, g_Adventure] |\n", "|Shawshank Redemption, The (1994) |774 |[g_Drama, g_Crime] |\n", "|Phenomena (a.k.a. Creepers) (1985) |772 |[g_Thriller, g_Mystery, g_Horror] |\n", "|Deadpool 2 (2018) |756 |[g_Sci-Fi, g_Action, g_Comedy] |\n", "|Forrest Gump (1994) |752 |[g_Romance, g_Drama, g_Comedy, g_War] |\n", "|Suspiria (2017) |744 |[g_Thriller, g_Horror] |\n", "|Paddington 2 (2017) |740 |[g_Comedy, g_Adventure, g_Children, g_Animation] |\n", "|Once Upon a Time in Hollywood (2019) |706 |[g_Drama, g_Comedy] |\n", "|Captain America: Civil War (2016) |701 |[g_Sci-Fi, g_Action, g_Thriller] |\n", "|Knives Out (2019) |700 |[g_Drama, g_Crime, g_Comedy, g_Thriller, g_Mystery]|\n", "|Lord of the Rings: The Return of the King, The (2003)|691 |[g_Action, g_Drama, g_Adventure, g_Fantasy] |\n", "+-----------------------------------------------------+-----+---------------------------------------------------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " \r" ] } ], "source": [ "from pyspark.sql.functions import collect_set, size, col, coalesce, lit\n", "\n", "# Step 1: Pattern match HAS_GENRE\n", "movie_genres = g.find(\"(m)-[h]->(g)\") \\\n", " .filter(\"h.relationship = 'HAS_GENRE'\") \\\n", " .select(\n", " col(\"m.id\").alias(\"movieId\"),\n", " col(\"g.id\").alias(\"genre\")\n", " )\n", "# movie_genres = g.find(\"(m)-[h]->(g)\") \\\n", "# .filter(\"h.relationship = 'HAS_GENRE'\") \\\n", "# .select(\n", "# col(\"m.id\").alias(\"movieId\"),\n", "# col(\"g.id\").alias(\"genre\")\n", "# ).dropDuplicates()\n", "\n", "#Step 2: Pattern match TAGGED\n", "movie_tags = g.find(\"(u)-[t]->(m)\") \\\n", " .filter(\"t.relationship = 'TAGGED' AND t.tag IS NOT NULL\") \\\n", " .select(\n", " col(\"u.id\").alias(\"userId\"),\n", " col(\"m.id\").alias(\"movieId\"),\n", " col(\"t.tag\").alias(\"tag\")\n", " ) \\\n", "\n", "# movie_tags = g.find(\"(u)-[t]->(m)\") \\\n", "# .filter(\"t.relationship = 'TAGGED' AND t.tag IS NOT NULL\") \\\n", "# .select(\n", "# col(\"u.id\").alias(\"userId\"),\n", "# col(\"m.id\").alias(\"movieId\"),\n", "# trim(lower(col(\"t.tag\"))).alias(\"tag\")\n", "# ).dropDuplicates([\"userId\", \"movieId\"])\n", "\n", "\n", "\n", "# Step 3: Aggregate genres and tags\n", "genres_per_movie = movie_genres.groupBy(\"movieId\") \\\n", " .agg(collect_set(\"genre\").alias(\"genres\"))\n", "\n", "tags_per_movie = movie_tags.groupBy(\"movieId\") \\\n", " .agg(collect_set(\"tag\").alias(\"tags\"))\n", "\n", "# Step 4: Full outer join to preserve all movies\n", "combined = genres_per_movie.join(tags_per_movie, on=\"movieId\", how=\"full_outer\")\n", "\n", "# Step 5: Calculate counts (handling nulls)\n", "final = combined \\\n", " .withColumn(\"genres\", coalesce(col(\"genres\"), lit([]))) \\\n", " .withColumn(\"tags\", coalesce(col(\"tags\"), lit([]))) \\\n", " .withColumn(\"genreCount\", size(col(\"genres\"))) \\\n", " .withColumn(\"tagCount\", size(col(\"tags\"))) \\\n", " .withColumn(\"total\", col(\"genreCount\") + col(\"tagCount\"))\n", "\n", "# Step 6: Join titles\n", "movie_titles = movie_vertices.select(\n", " col(\"id\").cast(\"string\").alias(\"movieId\"),\n", " \"title\"\n", ")\n", "\n", "result = final.join(movie_titles, on=\"movieId\", how=\"left\") \\\n", " .select(\"title\", \"total\", \"genres\") \\\n", " .orderBy(col(\"total\").desc(), col(\"title\").asc()) \\\n", " .limit(20)\n", "\n", "result.show(truncate=False)\n" ] }, { "cell_type": "code", "execution_count": 385, "id": "a6b7c692-faad-4e7b-985a-12472a1ad20b", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "[Stage 4891:================================================> (29 + 3) / 32]\r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+--------+-------------+\n", "| movieId|fiveStarCount|\n", "+--------+-------------+\n", "|m_139644| 1198|\n", "| m_1367| 694|\n", "| m_208| 1372|\n", "| m_387| 18|\n", "|m_105246| 27|\n", "| m_45081| 309|\n", "| m_612| 92|\n", "| m_94401| 4|\n", "|m_253690| 92|\n", "| m_3519| 72|\n", "| m_6986| 87|\n", "| m_174| 92|\n", "| m_26153| 5|\n", "| m_1379| 168|\n", "| m_39444| 65|\n", "|m_244368| 12|\n", "| m_523| 158|\n", "| m_4389| 47|\n", "| m_31184| 66|\n", "| m_433| 40|\n", "+--------+-------------+\n", "only showing top 20 rows\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " \r" ] } ], "source": [ "# Query - 11: Find top-20 movies with the highest number of 5-ratings, return the movies and this rating number.\n", "\n", "five_star_counts.show()" ] }, { "cell_type": "code", "execution_count": 388, "id": "5c676b7f-56e1-477a-ad27-9ddbb13e4c31", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "[Stage 4949:===============================> (19 + 13) / 32]\r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+------------------------------------------------------------------------------+-------------+\n", "|title |fiveStarCount|\n", "+------------------------------------------------------------------------------+-------------+\n", "|Shawshank Redemption, The (1994) |49079 |\n", "|Pulp Fiction (1994) |39248 |\n", "|Matrix, The (1999) |32709 |\n", "|Forrest Gump (1994) |31637 |\n", "|Star Wars: Episode IV - A New Hope (1977) |30898 |\n", "|Godfather, The (1972) |30392 |\n", "|Schindler's List (1993) |29743 |\n", "|Silence of the Lambs, The (1991) |29628 |\n", "|Fight Club (1999) |27930 |\n", "|Star Wars: Episode V - The Empire Strikes Back (1980) |25775 |\n", "|Usual Suspects, The (1995) |25391 |\n", "|Lord of the Rings: The Fellowship of the Ring, The (2001) |24723 |\n", "|Lord of the Rings: The Return of the King, The (2003) |22751 |\n", "|Braveheart (1995) |21362 |\n", "|Lord of the Rings: The Two Towers, The (2002) |21332 |\n", "|Raiders of the Lost Ark (Indiana Jones and the Raiders of the Lost Ark) (1981)|21332 |\n", "|American Beauty (1999) |20010 |\n", "|Dark Knight, The (2008) |19420 |\n", "|Fargo (1996) |19075 |\n", "|Star Wars: Episode VI - Return of the Jedi (1983) |18903 |\n", "+------------------------------------------------------------------------------+-------------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " \r" ] } ], "source": [ "from pyspark.sql.functions import col, count\n", "\n", "# Step 1: Pattern match RATED edges with rating = 5.0\n", "five_star_df = g.find(\"(u)-[r]->(m)\") \\\n", " .filter(\"r.relationship = 'RATED' AND r.rating = 5.0\") \\\n", " .select(col(\"m.id\").alias(\"movieId\"))\n", "\n", "# Step 2: Count how many times each movie got a 5.0\n", "five_star_counts = five_star_df.groupBy(\"movieId\") \\\n", " .agg(count(\"*\").alias(\"fiveStarCount\"))\n", "\n", "# Step 3: Join with movie titles\n", "movie_titles = movie_vertices.select(\n", " col(\"id\").cast(\"string\").alias(\"movieId\"),\n", " col(\"title\")\n", ")\n", "\n", "result = five_star_counts.join(movie_titles, on=\"movieId\", how=\"left\") \\\n", " .select(\"title\", \"fiveStarCount\") \\\n", " .orderBy(col(\"fiveStarCount\").desc()) \\\n", " .limit(20)\n", "\n", "result.show(truncate=False)\n" ] }, { "cell_type": "code", "execution_count": 387, "id": "ce93d9d6-f1db-48b6-9a3f-fc1fd59ae14a", "metadata": {}, "outputs": [ { "name": "stderr", "output_type": "stream", "text": [ "[Stage 4933:======> (4 + 28) / 32]\r" ] }, { "name": "stdout", "output_type": "stream", "text": [ "+------------------------------------------------------------------------------+-------------+\n", "|title |fiveStarCount|\n", "+------------------------------------------------------------------------------+-------------+\n", "|Shawshank Redemption, The (1994) |49079 |\n", "|Pulp Fiction (1994) |39248 |\n", "|Matrix, The (1999) |32709 |\n", "|Forrest Gump (1994) |31637 |\n", "|Star Wars: Episode IV - A New Hope (1977) |30898 |\n", "|Godfather, The (1972) |30392 |\n", "|Schindler's List (1993) |29743 |\n", "|Silence of the Lambs, The (1991) |29628 |\n", "|Fight Club (1999) |27930 |\n", "|Star Wars: Episode V - The Empire Strikes Back (1980) |25775 |\n", "|Usual Suspects, The (1995) |25391 |\n", "|Lord of the Rings: The Fellowship of the Ring, The (2001) |24723 |\n", "|Lord of the Rings: The Return of the King, The (2003) |22751 |\n", "|Braveheart (1995) |21362 |\n", "|Raiders of the Lost Ark (Indiana Jones and the Raiders of the Lost Ark) (1981)|21332 |\n", "|Lord of the Rings: The Two Towers, The (2002) |21332 |\n", "|American Beauty (1999) |20010 |\n", "|Dark Knight, The (2008) |19420 |\n", "|Fargo (1996) |19075 |\n", "|Star Wars: Episode VI - Return of the Jedi (1983) |18903 |\n", "+------------------------------------------------------------------------------+-------------+\n", "\n" ] }, { "name": "stderr", "output_type": "stream", "text": [ " \r" ] } ], "source": [ "from pyspark.sql.functions import col, count\n", "\n", "# Step 1: Pattern match RATED edges with rating = 5.0 and extract movie titles\n", "five_star_df = g.find(\"(u)-[r]->(m)\") \\\n", " .filter(\"r.relationship = 'RATED' AND r.rating = 5.0\") \\\n", " .select(\n", " col(\"m.id\").alias(\"movieId\"),\n", " col(\"m.title\").alias(\"title\")\n", " )\n", "\n", "# Step 2: Group by title and count 5-star ratings\n", "top_movies = five_star_df.groupBy(\"title\") \\\n", " .agg(count(\"*\").alias(\"fiveStarCount\")) \\\n", " .orderBy(col(\"fiveStarCount\").desc()) \\\n", " .limit(20)\n", "\n", "# Step 3: Show result\n", "top_movies.show(truncate=False)" ] }, { "cell_type": "code", "execution_count": null, "id": "4b58d228-7fb0-4855-962d-708763d44b02", "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.12.7" } }, "nbformat": 4, "nbformat_minor": 5 }