GraphFrames MovieLens 32M Implementation

Schema Diagram and Graph Statistics

Since GraphFrames has no support for printing a graph schema, we reuse the one output from Neo4j.

Node Statistics

Node Label	Count
Genre	20
Movie	87,585
User	200,948

Edge Statistics

Relationship	Count
(:Movie)-[:HAS_GENRE]-(:Genre)	154,170
(:User)-[:RATED]-(:Movie)	32,000,204
(:User)-[:TAGGED]-(:Movie)	2,000,072

Data Loading

Overview

Data import is handled in three parts. First we upload all the raw data files into DataFrames. Then we define nodes and edges. Finally, we create graph schema.

In order to cope with the formatting in the source file, we defined the following function to read csv files.

python

def read_csv(filename, schema):
    return (spark.read.format("csv").option("header", "true")
        .option("quote", '"').option("escape", '"').option("mode", "FAILFAST")
        .schema(schema).load(f"{INPUT_DIR.rstrip('/')}/{filename}"))

Movie Nodes

The first thing we do would be to load each CSV file into a DataFrame. We start with the Movie nodes.

python

MOVIE_SCHEMA = StructType([
    StructField("movieId", IntegerType(), False),
    StructField("title", StringType(), False),
    StructField("genres", StringType(), True),
])
movie_raw = read_csv("movies.csv", MOVIE_SCHEMA)
movie = (movie_raw
    .where(F.col("movieId").isNotNull() & F.col("title").isNotNull())
    .dropDuplicates(["movieId"]))
movie_vertices_df = movie.select(
    F.concat(F.lit("m_"), F.col("movieId").cast("string")).alias("id"),
    F.col("movieId"), F.col("title"))

Genre Nodes

Then we start with the Genre nodes. It is derived from the Movie nodes.

python

genre = (movie.where(F.col("genres").isNotNull())
    .withColumn("name", F.explode(F.split(F.col("genres"), r"\|")))
    .withColumn("name", F.trim(F.col("name")))
    .where(F.col("name").isNotNull() & (F.col("name") != ""))
    .select("movieId", "name").dropDuplicates(["movieId", "name"]))
genre_vertices_df = genre.select("name").distinct().select(
    F.concat(F.lit("g_"), F.col("name")).alias("id"), F.col("name"))

HAS_GENRE Edges

Next we create the DataFrame for HAS_GENRE edges. It can be derived from Genre nodes and Movie nodes.

python

has_genre_edges_df = genre.select(
    F.concat(F.lit("m_"), F.col("movieId").cast("string")).alias("src"),
    F.concat(F.lit("g_"), F.col("name")).alias("dst"))

RATED Edges

Next we create the DataFrame for RATED edges. We create an edge label before finishing all node lables because the User nodes can be constructed from the RATED dataframe.

python

RATED_SCHEMA = StructType([
    StructField("userId", IntegerType(), False),
    StructField("movieId", IntegerType(), False),
    StructField("rating", DoubleType(), False),
    StructField("timestamp", LongType(), False),
])
rated_raw = read_csv("ratings.csv", RATED_SCHEMA)
rated = rated_raw.where(
    F.col("userId").isNotNull() & F.col("movieId").isNotNull()
    & F.col("rating").isNotNull() & F.col("timestamp").isNotNull())
rated_edges_df = rated.select(
    F.concat(F.lit("u_"), F.col("userId").cast("string")).alias("src"),
    F.concat(F.lit("m_"), F.col("movieId").cast("string")).alias("dst"),
    F.col("rating"), F.timestamp_seconds(F.col("timestamp")).alias("timestamp"))

User Nodes

Then we start with the User nodes.

python

user = rated.select("userId").union(tagged.select("userId")).distinct()
user_vertices_df = user.select(
    F.concat(F.lit("u_"), F.col("userId").cast("string")).alias("id"),
    F.col("userId"))

TAGGED Edges

Next we create the DataFrame for TAGGED edges.

python

TAGGED_SCHEMA = StructType([
    StructField("userId", IntegerType(), False),
    StructField("movieId", IntegerType(), False),
    StructField("tag", StringType(), False),
    StructField("timestamp", LongType(), False),
])
tagged_raw = read_csv("tags.csv", TAGGED_SCHEMA)
tagged = tagged_raw.where(
    F.col("userId").isNotNull() & F.col("movieId").isNotNull()
    & F.col("tag").isNotNull() & (F.trim(F.col("tag")) != "")
    & F.col("timestamp").isNotNull())
tagged_edges_df = tagged.select(
    F.concat(F.lit("u_"), F.col("userId").cast("string")).alias("src"),
    F.concat(F.lit("m_"), F.col("movieId").cast("string")).alias("dst"),
    F.col("tag"), F.timestamp_seconds(F.col("timestamp")).alias("timestamp"))

Define Graph

Now that we have defined our vertices and edges, we can create graph schema with those. First, we need to union all vertex types into a single vertex dataframe, and all edge types into a single edge dataframe.

First we define the following two functions to union all node and edge DataFrames:

python

def build_vertices(movie_df, user_df, genre_df):
    return (movie_df.withColumn("type", F.lit("Movie"))
        .unionByName(user_df.withColumn("type", F.lit("User")), allowMissingColumns=True)
        .unionByName(genre_df.withColumn("type", F.lit("Genre")), allowMissingColumns=True)
        .select("id", "type", "movieId", "userId", "title", "name"))

def build_edges(rated_df, tagged_df, has_genre_df):
    return (rated_df.withColumn("relationship", F.lit("RATED"))
        .unionByName(tagged_df.withColumn("relationship", F.lit("TAGGED")), allowMissingColumns=True)
        .unionByName(has_genre_df.withColumn("relationship", F.lit("HAS_GENRE")), allowMissingColumns=True)
        .select("src", "dst", "relationship", "rating", "tag", "timestamp"))

python

from graphframes import GraphFrame

vertices_df = build_vertices(movie_vertices_df, user_vertices_df, genre_vertices_df)
edges_df = build_edges(rated_edges_df, tagged_edges_df, has_genre_edges_df)
g_preview = GraphFrame(vertices_df, edges_df)

Queries

Select a query below to view its logic and description.

1. Neighborhood

2. Path

3. Pattern

4. Aggregation

5. Analytics

6. Hybrid

7. Others

Q1. Movies Rated by a User

Subcategory: T1.1 One-Hop Neighbors

This query lists the movies directly connected to a selected user by outgoing RATED relationships.

python

q01_vertices = (spark.table("ml32m.ml32m_user").withColumn("type", F.lit("User"))
    .unionByName(spark.table("ml32m.ml32m_movie").withColumn("type", F.lit("Movie")), allowMissingColumns=True))
q01_edges = spark.table("ml32m.ml32m_rated").withColumn("relationship", F.lit("RATED"))
q01_graph = GraphFrame(q01_vertices, q01_edges)
q01_result = (q01_graph.find("(u)-[r]->(m)")
    .filter((F.col("u.type") == "User") & (F.col("u.userId") == 33) & (F.col("m.type") == "Movie"))
    .select(F.col("m.movieId").alias("movieId"), F.col("m.title").alias("title"),
            F.col("r.rating").alias("rating"), F.col("r.timestamp").alias("ratingTime"))
    .orderBy(F.desc("rating"), F.asc("title")))
q01_result.show(20, truncate=False)

Equivalent query in GQL standard:

gql

MATCH (u:ml32m_user {userId: $userId})-[r:ml32m_rated]->(m:ml32m_movie)
RETURN m.movieId AS movieId, m.title AS title, r.rating AS rating, r.timestamp AS ratingTime
ORDER BY rating DESC, title
LIMIT $limit;

Q2. Genres within Two Hops of a User

Subcategory: T1.2 k-Hop Neighborhood

This query follows a selected user to rated movies and then to their genres, returning distinct genres within two hops.

python

q02_vertices = (spark.table("ml32m.ml32m_user").withColumn("type", F.lit("User"))
    .unionByName(spark.table("ml32m.ml32m_movie").withColumn("type", F.lit("Movie")), allowMissingColumns=True)
    .unionByName(spark.table("ml32m.ml32m_genre").withColumn("type", F.lit("Genre")), allowMissingColumns=True))
q02_edges = (spark.table("ml32m.ml32m_rated").withColumn("relationship", F.lit("RATED"))
    .unionByName(spark.table("ml32m.ml32m_has_genre").withColumn("relationship", F.lit("HAS_GENRE")), allowMissingColumns=True))
q02_graph = GraphFrame(q02_vertices, q02_edges)
q02_result = (q02_graph.find("(u)-[r]->(m); (m)-[hg]->(g)")
    .filter((F.col("u.userId") == 33) & (F.col("r.relationship") == "RATED") & (F.col("hg.relationship") == "HAS_GENRE"))
    .select(F.col("g.name").alias("genre")).distinct().orderBy("genre"))
q02_result.show(20, truncate=False)

Equivalent query in GQL standard:

gql

MATCH (:ml32m_user {userId: $userId})-[:ml32m_rated]->(:ml32m_movie)-[:ml32m_has_genre]->(g:ml32m_genre)
RETURN DISTINCT g.name AS genre
ORDER BY genre
LIMIT $limit;

Q3. Users Exactly Two Rating Hops Away

Subcategory: T1.3 Exact kth-Hop Neighbors

This query finds users exactly two RATED hops away through movies co-rated with a selected user.

python

q03_vertices = (spark.table("ml32m.ml32m_user").withColumn("type", F.lit("User"))
    .unionByName(spark.table("ml32m.ml32m_movie").withColumn("type", F.lit("Movie")), allowMissingColumns=True))
q03_graph = GraphFrame(q03_vertices, spark.table("ml32m.ml32m_rated"))
q03_result = (q03_graph.find("(u)-[r1]->(m); (other)-[r2]->(m)")
    .filter((F.col("u.userId") == 33) & (F.col("other.userId") != 33))
    .select(F.col("other.userId").alias("userId")).distinct().orderBy("userId"))
q03_result.show(20, truncate=False)

Equivalent query in GQL standard:

gql

MATCH (:ml32m_user {userId: $userId})-[:ml32m_rated]->(:ml32m_movie)<-[:ml32m_rated]-(other:ml32m_user)
WHERE other.userId <> $userId
RETURN DISTINCT other.userId AS userId
ORDER BY userId
LIMIT $limit;

Q4. Highly Rated One-Hop Movies

Subcategory: T1.4 Neighborhood with Property Filter

This query filters a selected user's one-hop movie neighborhood to movies with ratings at or above a threshold.

python

q04_vertices = (spark.table("ml32m.ml32m_user").withColumn("type", F.lit("User"))
    .unionByName(spark.table("ml32m.ml32m_movie").withColumn("type", F.lit("Movie")), allowMissingColumns=True))
q04_graph = GraphFrame(q04_vertices, spark.table("ml32m.ml32m_rated"))
q04_result = (q04_graph.find("(u)-[r]->(m)")
    .filter((F.col("u.userId") == 33) & (F.col("r.rating") >= 4.0))
    .select(F.col("m.movieId").alias("movieId"), F.col("m.title").alias("title"), F.col("r.rating").alias("rating"))
    .orderBy(F.desc("rating"), F.asc("title")))
q04_result.show(20, truncate=False)

Equivalent query in GQL standard:

gql

MATCH (u:ml32m_user {userId: $userId})-[r:ml32m_rated]->(m:ml32m_movie)
WHERE r.rating >= $minRating
RETURN m.movieId AS movieId, m.title AS title, r.rating AS rating
ORDER BY rating DESC, title
LIMIT $limit;

Q5. Filter Movies by Title Text

Subcategory: T2.1 Node Filtering (Single Node Paths)

This query filters Movie nodes by title text and movie ID without traversing relationships.

python

q05_result = (spark.table("ml32m.ml32m_movie")
    .filter(F.col("title").contains("Love") & (F.col("movieId") >= 99))
    .select("movieId", "title").orderBy("movieId"))
q05_result.show(20, truncate=False)

Equivalent query in GQL standard:

gql

MATCH (m:ml32m_movie)
WHERE m.title CONTAINS $titleText
  AND m.movieId >= $minMovieId
RETURN m.movieId AS movieId, m.title AS title
ORDER BY m.movieId
LIMIT $limit;

Q6. Filter High-Rating Edges

Subcategory: T2.2 Edge Filtering

This query filters RATED relationships by numeric rating and returns their endpoint user and movie.

python

q06_vertices = (spark.table("ml32m.ml32m_user").withColumn("type", F.lit("User"))
    .unionByName(spark.table("ml32m.ml32m_movie").withColumn("type", F.lit("Movie")), allowMissingColumns=True))
q06_graph = GraphFrame(q06_vertices, spark.table("ml32m.ml32m_rated"))
q06_result = (q06_graph.find("(u)-[r]->(m)").filter(F.col("r.rating") >= 4.0)
    .select(F.col("u.userId").alias("userId"), F.col("m.movieId").alias("movieId"),
            F.col("m.title").alias("title"), F.col("r.rating").alias("rating"))
    .orderBy(F.desc("rating")))
q06_result.show(20, truncate=False)

Equivalent query in GQL standard:

gql

MATCH (u:ml32m_user)-[r:ml32m_rated]->(m:ml32m_movie)
WHERE r.rating >= $minRating
RETURN u.userId AS userId, m.movieId AS movieId, m.title AS title, r.rating AS rating
ORDER BY rating DESC
LIMIT $limit;

Q7. Check Whether a User Can Reach a Movie

Subcategory: T2.3 Reachability between Two Nodes

This query tests whether a selected user can reach a selected movie within three undirected RATED hops.

python

q07_vertices = (spark.table("ml32m.ml32m_user").unionByName(
    spark.table("ml32m.ml32m_movie"), allowMissingColumns=True))
q07_graph = GraphFrame(q07_vertices, spark.table("ml32m.ml32m_rated")).as_undirected()
q07_paths = q07_graph.bfs("id = 'u_33'", "id = 'm_66'", maxPathLength=3)
q07_result = spark.createDataFrame([(q07_paths.limit(1).count() > 0,)], ["reachable"])
q07_result.show(truncate=False)

Equivalent query in GQL standard:

gql

MATCH p = ANY (:ml32m_user {userId: $userId})-[:ml32m_rated]-{1,3}(:ml32m_movie {movieId: $movieId})
RETURN true AS reachable;

Q8. Shortest Co-Rating Path to a Movie

Subcategory: T2.4 Shortest Path

This query finds the shortest undirected RATED path between a selected user and movie.

python

q08_vertices = (spark.table("ml32m.ml32m_user").unionByName(
    spark.table("ml32m.ml32m_movie"), allowMissingColumns=True))
q08_graph = GraphFrame(q08_vertices, spark.table("ml32m.ml32m_rated")).as_undirected()
q08_paths = q08_graph.bfs("id = 'u_33'", "id = 'm_66'", maxPathLength=4)
q08_hops = len([name for name in q08_paths.columns if name.startswith("e")])
q08_result = q08_paths.withColumn("hopCount", F.lit(q08_hops))
q08_result.show(20, truncate=False)

Equivalent query in GQL standard:

gql

MATCH p = SHORTEST (:ml32m_user {userId: $userId})-[:ml32m_rated]-{1,4}(:ml32m_movie {movieId: $movieId})
RETURN p, length(p) AS hopCount;

Q9. Simple Co-Rating Paths up to Three Hops

Subcategory: T2.5 All Simple Paths up to Length k

This query enumerates acyclic RATED paths between a selected user and movie up to three hops.

python

q09_vertices = (spark.table("ml32m.ml32m_user").unionByName(
    spark.table("ml32m.ml32m_movie"), allowMissingColumns=True))
q09_graph = GraphFrame(q09_vertices, spark.table("ml32m.ml32m_rated"))
q09_result = (q09_graph.all_paths(
        from_expr="id = 'u_33'", to_expr="id = 'm_66'",
        max_path_length=3, is_directed=False, use_local_checkpoints=True)
    .select(F.col("path"), F.col("len").alias("hopCount")).orderBy("hopCount"))
q09_result.show(20, truncate=False)

Equivalent query in GQL standard:

gql

MATCH p = ACYCLIC (:ml32m_user {userId: $userId})-[:ml32m_rated]-{1,3}(:ml32m_movie {movieId: $movieId})
RETURN p, length(p) AS hopCount
ORDER BY hopCount
LIMIT $limit;

Q10. Top Weighted Co-Rating Paths

Subcategory: T2.6 top-K Weighted Paths

This query ranks bounded RATED paths by the sum of rating values along each path.

python

q10_vertices = (spark.table("ml32m.ml32m_user").withColumn("type", F.lit("User"))
    .unionByName(spark.table("ml32m.ml32m_movie").withColumn("type", F.lit("Movie")), allowMissingColumns=True))
q10_graph = GraphFrame(q10_vertices, spark.table("ml32m.ml32m_rated"))
q10_direct = (q10_graph.find("(u)-[e1]->(m)")
    .filter((F.col("u.id") == "u_33") & (F.col("m.id") == "m_66"))
    .select(F.array(F.col("u.id"), F.col("m.id")).alias("path"), F.lit(1).alias("hopCount"),
            F.col("e1.rating").alias("ratingWeight")))
q10_three_hop = (q10_graph.find("(u)-[e1]->(m1); (u2)-[e2]->(m1); (u2)-[e3]->(m)")
    .filter((F.col("u.id") == "u_33") & (F.col("m.id") == "m_66")
            & (F.col("u.id") != F.col("u2.id")) & (F.col("m1.id") != F.col("m.id")))
    .select(F.array(F.col("u.id"), F.col("m1.id"), F.col("u2.id"), F.col("m.id")).alias("path"),
            F.lit(3).alias("hopCount"),
            (F.col("e1.rating") + F.col("e2.rating") + F.col("e3.rating")).alias("ratingWeight")))
q10_result = q10_direct.unionByName(q10_three_hop).orderBy(F.desc("ratingWeight"))
q10_result.show(20, truncate=False)

Equivalent query in GQL standard:

gql

MATCH p = ACYCLIC (:ml32m_user {userId: $userId})-[:ml32m_rated]-{1,3}(:ml32m_movie {movieId: $movieId})
WITH p, reduce(total = 0.0, r IN relationships(p) | total + coalesce(r.rating, 0.0)) AS ratingWeight
RETURN p, length(p) AS hopCount, ratingWeight
ORDER BY ratingWeight DESC
LIMIT $limit;

Q11. Three-Movie Triangle by Shared Genres

Subcategory: T3.1 Triangle Detection

This query finds three movies where every pair shares at least one genre through HAS_GENRE relationships.

python

q11_bipartite_vertices = (spark.table("ml32m.ml32m_movie").withColumn("type", F.lit("Movie"))
    .unionByName(spark.table("ml32m.ml32m_genre").withColumn("type", F.lit("Genre")), allowMissingColumns=True))
q11_bipartite = GraphFrame(q11_bipartite_vertices, spark.table("ml32m.ml32m_has_genre"))
q11_pairs = (q11_bipartite.find("(a)-[ag]->(genre); (b)-[bg]->(genre)")
    .filter(F.col("a.movieId") < F.col("b.movieId"))
    .select(F.col("a.id").alias("src"), F.col("b.id").alias("dst")).distinct())
q11_movie_graph = GraphFrame(spark.table("ml32m.ml32m_movie"), q11_pairs).as_undirected()
q11_result = (q11_movie_graph.find("(a)-[ab]->(b); (b)-[bc]->(c); (c)-[ca]->(a)")
    .filter((F.col("a.movieId") < F.col("b.movieId")) & (F.col("b.movieId") < F.col("c.movieId")))
    .select(F.col("a.title").alias("movieA"), F.col("b.title").alias("movieB"),
            F.col("c.title").alias("movieC")).distinct())
q11_result.show(20, truncate=False)

Equivalent query in GQL standard:

gql

MATCH (a:ml32m_movie)-[:ml32m_has_genre]->(:ml32m_genre)<-[:ml32m_has_genre]-(b:ml32m_movie)
MATCH (b)-[:ml32m_has_genre]->(:ml32m_genre)<-[:ml32m_has_genre]-(c:ml32m_movie)
MATCH (c)-[:ml32m_has_genre]->(:ml32m_genre)<-[:ml32m_has_genre]-(a)
WHERE a.movieId < b.movieId AND b.movieId < c.movieId
RETURN DISTINCT a.title AS movieA, b.title AS movieB, c.title AS movieC
LIMIT $limit;

Q12. Four-Movie Square by Shared Genres

Subcategory: T3.2 Square Detection

This query finds four movies connected in a cycle where each adjacent pair shares a genre.

python

q12_bipartite_vertices = (spark.table("ml32m.ml32m_movie").withColumn("type", F.lit("Movie"))
    .unionByName(spark.table("ml32m.ml32m_genre").withColumn("type", F.lit("Genre")), allowMissingColumns=True))
q12_bipartite = GraphFrame(q12_bipartite_vertices, spark.table("ml32m.ml32m_has_genre"))
q12_pairs = (q12_bipartite.find("(a)-[ag]->(genre); (b)-[bg]->(genre)")
    .filter(F.col("a.movieId") < F.col("b.movieId"))
    .select(F.col("a.id").alias("src"), F.col("b.id").alias("dst")).distinct())
q12_movie_graph = GraphFrame(spark.table("ml32m.ml32m_movie"), q12_pairs).as_undirected()
q12_result = (q12_movie_graph.find("(a)-[ab]->(b); (b)-[bc]->(c); (c)-[cd]->(d); (d)-[da]->(a)")
    .filter((F.col("a.movieId") < F.col("b.movieId")) & (F.col("a.movieId") < F.col("c.movieId"))
            & (F.col("a.movieId") < F.col("d.movieId")) & (F.col("b.id") != F.col("c.id"))
            & (F.col("b.id") != F.col("d.id")) & (F.col("c.id") != F.col("d.id")))
    .select(F.col("a.title").alias("movieA"), F.col("b.title").alias("movieB"),
            F.col("c.title").alias("movieC"), F.col("d.title").alias("movieD")).distinct())
q12_result.show(20, truncate=False)

Equivalent query in GQL standard:

gql

MATCH (a:ml32m_movie)-[:ml32m_has_genre]->(:ml32m_genre)<-[:ml32m_has_genre]-(b:ml32m_movie)
MATCH (b)-[:ml32m_has_genre]->(:ml32m_genre)<-[:ml32m_has_genre]-(c:ml32m_movie)
MATCH (c)-[:ml32m_has_genre]->(:ml32m_genre)<-[:ml32m_has_genre]-(d:ml32m_movie)
MATCH (d)-[:ml32m_has_genre]->(:ml32m_genre)<-[:ml32m_has_genre]-(a)
WHERE a.movieId < b.movieId AND a.movieId < c.movieId AND a.movieId < d.movieId
  AND b <> c AND b <> d AND c <> d
RETURN DISTINCT a.title AS movieA, b.title AS movieB, c.title AS movieC, d.title AS movieD
LIMIT $limit;

Q13. Users with Many Rated Movies

Subcategory: T3.3 Star Pattern Query

This query finds user-centered stars by counting how many distinct movies each user rated.

python

q13_vertices = (spark.table("ml32m.ml32m_user").withColumn("type", F.lit("User"))
    .unionByName(spark.table("ml32m.ml32m_movie").withColumn("type", F.lit("Movie")), allowMissingColumns=True))
q13_graph = GraphFrame(q13_vertices, spark.table("ml32m.ml32m_rated"))
q13_result = (q13_graph.find("(u)-[r]->(m)")
    .groupBy(F.col("u.id").alias("userVertexId"), F.col("u.userId").alias("userId"))
    .agg(F.countDistinct(F.col("m.id")).alias("ratedMovieCount"))
    .filter(F.col("ratedMovieCount") >= 10).select("userId", "ratedMovieCount")
    .orderBy(F.desc("ratedMovieCount")))
q13_result.show(20, truncate=False)

Equivalent query in GQL standard:

gql

MATCH (u:ml32m_user)-[:ml32m_rated]->(m:ml32m_movie)
WITH u, count(DISTINCT m) AS ratedMovieCount
WHERE ratedMovieCount >= $minDegree
RETURN u.userId AS userId, ratedMovieCount
ORDER BY ratedMovieCount DESC
LIMIT $limit;

Q14. Four-Movie Clique by Shared Genres

Subcategory: T3.4 4-Clique

This query finds four movies where every pair shares at least one genre, forming a clique in the derived movie-genre similarity graph.

python

q14_bipartite_vertices = (spark.table("ml32m.ml32m_movie").withColumn("type", F.lit("Movie"))
    .unionByName(spark.table("ml32m.ml32m_genre").withColumn("type", F.lit("Genre")), allowMissingColumns=True))
q14_bipartite = GraphFrame(q14_bipartite_vertices, spark.table("ml32m.ml32m_has_genre"))
q14_pairs = (q14_bipartite.find("(a)-[ag]->(genre); (b)-[bg]->(genre)")
    .filter(F.col("a.movieId") < F.col("b.movieId"))
    .select(F.col("a.id").alias("src"), F.col("b.id").alias("dst")).distinct())
q14_movie_graph = GraphFrame(spark.table("ml32m.ml32m_movie"), q14_pairs).as_undirected()
q14_pattern = ("(a)-[ab]->(b); (a)-[ac]->(c); (a)-[ad]->(d); "
               "(b)-[bc]->(c); (b)-[bd]->(d); (c)-[cd]->(d)")
q14_result = (q14_movie_graph.find(q14_pattern)
    .filter((F.col("a.movieId") < F.col("b.movieId")) & (F.col("b.movieId") < F.col("c.movieId"))
            & (F.col("c.movieId") < F.col("d.movieId")))
    .select(F.col("a.title").alias("movieA"), F.col("b.title").alias("movieB"),
            F.col("c.title").alias("movieC"), F.col("d.title").alias("movieD")).distinct())
q14_result.show(20, truncate=False)

Equivalent query in GQL standard:

gql

MATCH (a:ml32m_movie)-[:ml32m_has_genre]->(:ml32m_genre)<-[:ml32m_has_genre]-(b:ml32m_movie)
MATCH (a)-[:ml32m_has_genre]->(:ml32m_genre)<-[:ml32m_has_genre]-(c:ml32m_movie)
MATCH (a)-[:ml32m_has_genre]->(:ml32m_genre)<-[:ml32m_has_genre]-(d:ml32m_movie)
MATCH (b)-[:ml32m_has_genre]->(:ml32m_genre)<-[:ml32m_has_genre]-(c)
MATCH (b)-[:ml32m_has_genre]->(:ml32m_genre)<-[:ml32m_has_genre]-(d)
MATCH (c)-[:ml32m_has_genre]->(:ml32m_genre)<-[:ml32m_has_genre]-(d)
WHERE a.movieId < b.movieId AND b.movieId < c.movieId AND c.movieId < d.movieId
RETURN DISTINCT a.title AS movieA, b.title AS movieB, c.title AS movieC, d.title AS movieD
LIMIT $limit;

Q15. Users Who Rated the Same Movie

Subcategory: T3.5 Two-hop Join Pattern

This query finds pairs of users connected through a commonly rated movie.

python

q15_vertices = (spark.table("ml32m.ml32m_user").withColumn("type", F.lit("User"))
    .unionByName(spark.table("ml32m.ml32m_movie").withColumn("type", F.lit("Movie")), allowMissingColumns=True))
q15_graph = GraphFrame(q15_vertices, spark.table("ml32m.ml32m_rated"))
q15_result = (q15_graph.find("(u1)-[r1]->(m); (u2)-[r2]->(m)")
    .filter(F.col("u1.userId") < F.col("u2.userId"))
    .select(F.col("u1.userId").alias("userA"), F.col("u2.userId").alias("userB"),
            F.col("m.title").alias("sharedMovie")))
q15_result.show(20, truncate=False)

Equivalent query in GQL standard:

gql

MATCH (u1:ml32m_user)-[:ml32m_rated]->(m:ml32m_movie)<-[:ml32m_rated]-(u2:ml32m_user)
WHERE u1.userId < u2.userId
RETURN u1.userId AS userA, u2.userId AS userB, m.title AS sharedMovie
LIMIT $limit;

Q16. List users and the movies they rated that belong to the specific genre "Comedy", find top-20 results in descending order of ratings

Subcategory: T3.6 Line Pattern

This query retrieves users and their comedy movie ratings, returning the top 20 highest-rated user-movie pairs.

python

q16_vertices = (spark.table("ml32m.ml32m_user").withColumn("type", F.lit("User"))
    .unionByName(spark.table("ml32m.ml32m_movie").withColumn("type", F.lit("Movie")), allowMissingColumns=True)
    .unionByName(spark.table("ml32m.ml32m_genre").withColumn("type", F.lit("Genre")), allowMissingColumns=True))
q16_edges = (spark.table("ml32m.ml32m_rated").withColumn("relationship", F.lit("RATED"))
    .unionByName(spark.table("ml32m.ml32m_has_genre").withColumn("relationship", F.lit("HAS_GENRE")), allowMissingColumns=True))
q16_graph = GraphFrame(q16_vertices, q16_edges)
q16_result = (q16_graph.find("(u)-[r]->(m); (m)-[hg]->(g)")
    .filter((F.col("r.relationship") == "RATED") & (F.col("hg.relationship") == "HAS_GENRE")
            & (F.col("g.name") == "Comedy"))
    .select(F.col("u.userId").alias("userId"), F.col("m.title").alias("movieTitle"),
            F.col("g.name").alias("genre"), F.col("r.rating").alias("rating"))
    .orderBy(F.desc("rating")).limit(20))
q16_result.show(20, truncate=False)

Equivalent query in GQL standard:

gql

MATCH (u:ml32m_user)-[r:ml32m_rated]->(m:ml32m_movie)-[:ml32m_has_genre]->(g:ml32m_genre {name: "Comedy"})
RETURN u.userId AS userId, m.title AS movieTitle, g.name AS genre, r.rating AS rating
ORDER BY r.rating DESC
LIMIT 20

Q17. Movie Rating Degree Summary

Subcategory: T4.1 Ungrouped Node Degree Statistics by Label

This query computes global statistics over the number of RATED edges incident to each movie.

python

q17_vertices = (spark.table("ml32m.ml32m_user").withColumn("type", F.lit("User"))
    .unionByName(spark.table("ml32m.ml32m_movie").withColumn("type", F.lit("Movie")), allowMissingColumns=True))
q17_graph = GraphFrame(q17_vertices, spark.table("ml32m.ml32m_rated"))
q17_movie_degrees = (q17_graph.vertices.filter(F.col("type") == "Movie").select("id")
    .join(q17_graph.inDegrees.select("id", F.col("inDegree").alias("ratingDegree")), "id", "left")
    .fillna({"ratingDegree": 0}))
q17_result = q17_movie_degrees.agg(
    F.count("*").alias("movieCount"), F.avg("ratingDegree").alias("avgRatingDegree"),
    F.min("ratingDegree").alias("minRatingDegree"), F.max("ratingDegree").alias("maxRatingDegree"))
q17_result.show(truncate=False)

Equivalent query in GQL standard:

gql

MATCH (m:ml32m_movie)
OPTIONAL MATCH (:ml32m_user)-[r:ml32m_rated]->(m)
WITH m, count(r) AS ratingDegree
RETURN count(m) AS movieCount,
       avg(ratingDegree) AS avgRatingDegree,
       min(ratingDegree) AS minRatingDegree,
       max(ratingDegree) AS maxRatingDegree;

Q18. Find users who have rated movies from at least 3 distinct genres, list the top-10 in descending order of the genre count

Subcategory: T4.2 Grouped Node Degree Statistics by Label

This query identifies the top 10 users who have rated movies spanning the most distinct genres (with a minimum of 3 genres).

python

q18_vertices = (spark.table("ml32m.ml32m_user").withColumn("type", F.lit("User"))
    .unionByName(spark.table("ml32m.ml32m_movie").withColumn("type", F.lit("Movie")), allowMissingColumns=True)
    .unionByName(spark.table("ml32m.ml32m_genre").withColumn("type", F.lit("Genre")), allowMissingColumns=True))
q18_edges = (spark.table("ml32m.ml32m_rated").withColumn("relationship", F.lit("RATED"))
    .unionByName(spark.table("ml32m.ml32m_has_genre").withColumn("relationship", F.lit("HAS_GENRE")), allowMissingColumns=True))
q18_graph = GraphFrame(q18_vertices, q18_edges)
q18_result = (q18_graph.find("(u)-[r]->(m); (m)-[hg]->(g)")
    .filter((F.col("r.relationship") == "RATED") & (F.col("hg.relationship") == "HAS_GENRE"))
    .groupBy(F.col("u.id").alias("userVertexId"), F.col("u.userId").alias("userId"))
    .agg(F.countDistinct(F.col("g.name")).alias("genreCount"))
    .filter(F.col("genreCount") >= 3).select("userId", "genreCount")
    .orderBy(F.desc("genreCount")).limit(10))
q18_result.show(10, truncate=False)

Equivalent query in GQL standard:

gql

MATCH (u:ml32m_user)-[r:ml32m_rated]->(m:ml32m_movie)-[:ml32m_has_genre]->(g:ml32m_genre)
WITH u, count(DISTINCT g.name) AS genreCount
WHERE genreCount >= 3
RETURN u.userId AS userId, genreCount
ORDER BY genreCount DESC
LIMIT 10

Q19. Find top-20 movies with the highest number of 5-ratings, return the movies and this rating number

Subcategory: T4.2 Grouped Node Degree Statistics by Label

This query identifies and lists the top 20 movies that have received the highest count of perfect 5.0 ratings.

python

q19_vertices = (spark.table("ml32m.ml32m_user").withColumn("type", F.lit("User"))
    .unionByName(spark.table("ml32m.ml32m_movie").withColumn("type", F.lit("Movie")), allowMissingColumns=True))
q19_graph = GraphFrame(q19_vertices, spark.table("ml32m.ml32m_rated"))
q19_result = (q19_graph.find("(u)-[r]->(m)").filter(F.col("r.rating") == 5.0)
    .groupBy(F.col("m.id").alias("movieVertexId"), F.col("m.title").alias("movieTitle"))
    .agg(F.count("*").alias("fiveStarCount")).select("movieTitle", "fiveStarCount")
    .orderBy(F.desc("fiveStarCount")).limit(20))
q19_result.show(20, truncate=False)

Equivalent query in GQL standard:

gql

MATCH (:ml32m_user)-[r:ml32m_rated]->(m:ml32m_movie)
WHERE r.rating = 5.0
RETURN m.title AS movieTitle, count(*) AS fiveStarCount
ORDER BY fiveStarCount DESC
LIMIT 20

Q20. Check different rating values and their occurrence frequencies

Subcategory: T4.4 Grouped Edge Count and Property Statistics

This query aggregates all ratings in the dataset and returns each rating value along with its total occurrence frequency, sorted by rating value.

python

q20_vertices = (spark.table("ml32m.ml32m_user").unionByName(
    spark.table("ml32m.ml32m_movie"), allowMissingColumns=True))
q20_graph = GraphFrame(q20_vertices, spark.table("ml32m.ml32m_rated"))
q20_result = (q20_graph.edges.groupBy("rating").agg(F.count("*").alias("frequency"))
    .orderBy("rating"))
q20_result.show(20, truncate=False)

Equivalent query in GQL standard:

gql

MATCH (:ml32m_user)-[r:ml32m_rated]->(:ml32m_movie)
RETURN r.rating AS rating, count(*) AS frequency
ORDER BY rating

Q21. Find the top-50 most frequent tags, return the tags along with their frequencies

Subcategory: T4.4 Grouped Edge Count and Property Statistics

This query ranks and returns the top 50 most frequently used movie tags in descending order of their frequency.

python

q21_vertices = (spark.table("ml32m.ml32m_user").unionByName(
    spark.table("ml32m.ml32m_movie"), allowMissingColumns=True))
q21_graph = GraphFrame(q21_vertices, spark.table("ml32m.ml32m_tagged"))
q21_result = (q21_graph.edges.filter(F.col("tag").isNotNull()).groupBy("tag")
    .agg(F.count("*").alias("frequency")).orderBy(F.desc("frequency")).limit(50))
q21_result.show(50, truncate=False)

Equivalent query in GQL standard:

gql

MATCH (:User)-[t:ml32m_tagged]->(:Movie)
WHERE t.tag IS NOT NULL
RETURN t.tag AS tag, count(*) AS frequency
ORDER BY frequency DESC
LIMIT 50

Q22. User Rating Path Counts by Length

Subcategory: T4.5 Path Counts by Length

This query counts bounded RATED paths from a selected user grouped by path length.

python

q22_vertices = (spark.table("ml32m.ml32m_user").withColumn("type", F.lit("User"))
    .unionByName(spark.table("ml32m.ml32m_movie").withColumn("type", F.lit("Movie")), allowMissingColumns=True))
q22_graph = GraphFrame(q22_vertices, spark.table("ml32m.ml32m_rated"))
q22_len1 = (q22_graph.find("(u)-[e1]->(target)").filter(F.col("u.id") == "u_33")
    .agg(F.count("*").alias("pathCount"), F.countDistinct(F.col("target.id")).alias("distinctTargetCount"))
    .select(F.lit(1).alias("pathLength"), "pathCount", "distinctTargetCount"))
q22_len2 = (q22_graph.find("(u)-[e1]->(m); (target)-[e2]->(m)")
    .filter((F.col("u.id") == "u_33") & (F.col("target.id") != F.col("u.id")))
    .agg(F.count("*").alias("pathCount"), F.countDistinct(F.col("target.id")).alias("distinctTargetCount"))
    .select(F.lit(2).alias("pathLength"), "pathCount", "distinctTargetCount"))
q22_len3 = (q22_graph.find("(u)-[e1]->(m1); (u2)-[e2]->(m1); (u2)-[e3]->(target)")
    .filter((F.col("u.id") == "u_33") & (F.col("u2.id") != F.col("u.id"))
            & (F.col("target.id") != F.col("m1.id")))
    .agg(F.count("*").alias("pathCount"), F.countDistinct(F.col("target.id")).alias("distinctTargetCount"))
    .select(F.lit(3).alias("pathLength"), "pathCount", "distinctTargetCount"))
q22_result = q22_len1.unionByName(q22_len2).unionByName(q22_len3).orderBy("pathLength")
q22_result.show(truncate=False)

Equivalent query in GQL standard:

gql

MATCH p = (:ml32m_user {userId: $userId})-[:ml32m_rated]-{1,3}(target)
RETURN length(p) AS pathLength,
       count(p) AS pathCount,
       count(DISTINCT target) AS distinctTargetCount
ORDER BY pathLength;

Q23. User Rating Path Counts by Length and Genre

Subcategory: T4.6 Grouped Path Counts by Length and Endpoint Type

This query groups bounded paths from a selected user by path length and by genre of reached movie endpoints.

python

q23_vertices = (spark.table("ml32m.ml32m_user").withColumn("type", F.lit("User"))
    .unionByName(spark.table("ml32m.ml32m_movie").withColumn("type", F.lit("Movie")), allowMissingColumns=True)
    .unionByName(spark.table("ml32m.ml32m_genre").withColumn("type", F.lit("Genre")), allowMissingColumns=True))
q23_edges = (spark.table("ml32m.ml32m_rated").withColumn("relationship", F.lit("RATED"))
    .unionByName(spark.table("ml32m.ml32m_has_genre").withColumn("relationship", F.lit("HAS_GENRE")), allowMissingColumns=True))
q23_graph = GraphFrame(q23_vertices, q23_edges)
q23_len1 = (q23_graph.find("(u)-[e1]->(m); (m)-[hg]->(g)")
    .filter((F.col("u.id") == "u_33") & (F.col("e1.relationship") == "RATED")
            & (F.col("hg.relationship") == "HAS_GENRE"))
    .groupBy(F.col("g.name").alias("genre"))
    .agg(F.count("*").alias("pathCount"), F.countDistinct(F.col("m.id")).alias("distinctMovieCount"))
    .withColumn("pathLength", F.lit(1)))
q23_len3 = (q23_graph.find("(u)-[e1]->(m1); (u2)-[e2]->(m1); (u2)-[e3]->(m); (m)-[hg]->(g)")
    .filter((F.col("u.id") == "u_33") & (F.col("u2.id") != F.col("u.id"))
            & (F.col("m.id") != F.col("m1.id")) & (F.col("e1.relationship") == "RATED")
            & (F.col("e2.relationship") == "RATED") & (F.col("e3.relationship") == "RATED")
            & (F.col("hg.relationship") == "HAS_GENRE"))
    .groupBy(F.col("g.name").alias("genre"))
    .agg(F.count("*").alias("pathCount"), F.countDistinct(F.col("m.id")).alias("distinctMovieCount"))
    .withColumn("pathLength", F.lit(3)))
q23_result = (q23_len1.unionByName(q23_len3).select("pathLength", "genre", "pathCount", "distinctMovieCount")
    .orderBy("pathLength", F.desc("pathCount")))
q23_result.show(50, truncate=False)

Equivalent query in GQL standard:

gql

MATCH p = (:ml32m_user {userId: $userId})-[:ml32m_rated]-{1,3}(m:ml32m_movie)-[:ml32m_has_genre]->(g:ml32m_genre)
RETURN length(p) AS pathLength,
       g.name AS genre,
       count(p) AS pathCount,
       count(DISTINCT m) AS distinctMovieCount
ORDER BY pathLength, pathCount DESC
LIMIT $limit;

Q24. Find the title of the movies that belong to multiple genres, e.g., "Comedy" and "Romance"

Subcategory: T4.7 Union and Intersection

This query finds and lists the titles of the first 10 movies that are classified under both the Comedy and Romance genres.

python

q24_vertices = (spark.table("ml32m.ml32m_movie").withColumn("type", F.lit("Movie"))
    .unionByName(spark.table("ml32m.ml32m_genre").withColumn("type", F.lit("Genre")), allowMissingColumns=True))
q24_graph = GraphFrame(q24_vertices, spark.table("ml32m.ml32m_has_genre"))
q24_result = (q24_graph.find("(m)-[e1]->(g1); (m)-[e2]->(g2)")
    .filter((F.col("g1.name") == "Comedy") & (F.col("g2.name") == "Romance"))
    .select(F.col("m.title").alias("movieTitle")).limit(10))
q24_result.show(10, truncate=False)

Equivalent query in GQL standard:

gql

MATCH (m:ml32m_movie)-[:ml32m_has_genre]->(g1:ml32m_genre {name: "Comedy"})
MATCH (m)-[:ml32m_has_genre]->(g2:ml32m_genre {name: "Romance"})
RETURN m.title AS movieTitle
LIMIT 10

Q25. Find users who have rated movies in both "Action" and "Romance" genres, list 10 userIds

Subcategory: T4.7 Union and Intersection

This query identifies and lists up to 10 unique user IDs who have rated at least one Action movie and at least one Romance movie.

python

q25_vertices = (spark.table("ml32m.ml32m_user").withColumn("type", F.lit("User"))
    .unionByName(spark.table("ml32m.ml32m_movie").withColumn("type", F.lit("Movie")), allowMissingColumns=True)
    .unionByName(spark.table("ml32m.ml32m_genre").withColumn("type", F.lit("Genre")), allowMissingColumns=True))
q25_edges = (spark.table("ml32m.ml32m_rated").withColumn("relationship", F.lit("RATED"))
    .unionByName(spark.table("ml32m.ml32m_has_genre").withColumn("relationship", F.lit("HAS_GENRE")), allowMissingColumns=True))
q25_graph = GraphFrame(q25_vertices, q25_edges)
q25_result = (q25_graph.find("(u)-[r1]->(m1); (m1)-[h1]->(g1); (u)-[r2]->(m2); (m2)-[h2]->(g2)")
    .filter((F.col("r1.relationship") == "RATED") & (F.col("r2.relationship") == "RATED")
            & (F.col("h1.relationship") == "HAS_GENRE") & (F.col("h2.relationship") == "HAS_GENRE")
            & (F.col("g1.name") == "Action") & (F.col("g2.name") == "Romance"))
    .select(F.col("u.userId").alias("userId")).distinct().limit(10))
q25_result.show(10, truncate=False)

Equivalent query in GQL standard:

gql

MATCH (u:ml32m_user)-[r:ml32m_rated]->(m1:ml32m_movie)-[:ml32m_has_genre]->(g1:ml32m_genre {name: "Action"})
MATCH (u)-[r:ml32m_rated]->(m2:ml32m_movie)-[:ml32m_has_genre]->(g2:ml32m_genre {name: "Romance"})
RETURN DISTINCT u.userId AS userId
LIMIT 10

Q26. Degree Centrality on the User-Movie Rating Graph

Subcategory: T5.1 Degree Centrality

This query projects the bipartite User-Movie rating graph and ranks nodes by degree centrality.

python

q26_vertices = (spark.table("ml32m.ml32m_user").withColumn("type", F.lit("User"))
    .unionByName(spark.table("ml32m.ml32m_movie").withColumn("type", F.lit("Movie")), allowMissingColumns=True))
q26_graph = GraphFrame(q26_vertices, spark.table("ml32m.ml32m_rated"))
q26_result = (q26_graph.vertices.join(q26_graph.degrees, "id", "left").fillna({"degree": 0})
    .select(F.array(F.col("type")).alias("labels"),
            F.coalesce(F.col("userId"), F.col("movieId")).alias("id"),
            F.col("title"), F.col("degree").cast("double").alias("score"))
    .orderBy(F.desc("score")))
q26_result.show(20, truncate=False)

Equivalent query in GQL standard:

gql

MATCH (n:ml32m_user|ml32m_movie)
OPTIONAL MATCH (n)-[r:ml32m_rated]-()
WITH n, count(r) AS score
RETURN coalesce(n.userId, n.movieId) AS id,
       n.title AS title,
       score
ORDER BY score DESC
LIMIT $limit;

Q27. PageRank on the User-Movie Rating Graph

Subcategory: T5.2 PageRank

This query runs PageRank over the projected User-Movie rating graph and returns the highest scoring nodes.

python

q27_vertices = (spark.table("ml32m.ml32m_user").withColumn("type", F.lit("User"))
    .unionByName(spark.table("ml32m.ml32m_movie").withColumn("type", F.lit("Movie")), allowMissingColumns=True))
q27_graph = GraphFrame(q27_vertices, spark.table("ml32m.ml32m_rated")).as_undirected()
q27_pagerank = q27_graph.pageRank(resetProbability=0.15, maxIter=20)
q27_result = (q27_pagerank.vertices
    .select(F.array(F.col("type")).alias("labels"),
            F.coalesce(F.col("userId"), F.col("movieId")).alias("id"),
            F.col("title"), F.col("pagerank").alias("score"))
    .orderBy(F.desc("score")))
q27_result.show(20, truncate=False)

Note: Cannot be expressed in standard GQL because the GQL standard covers graph pattern matching and data manipulation only; it defines no graph algorithm procedures such as PageRank.

Q28. Weakly Connected Components of Ratings

Subcategory: T5.3 Connected Components / Weakly Connected Components

This query finds weakly connected components in the undirected User-Movie rating graph.

python

q28_vertices = (spark.table("ml32m.ml32m_user").unionByName(
    spark.table("ml32m.ml32m_movie"), allowMissingColumns=True))
q28_graph = GraphFrame(q28_vertices, spark.table("ml32m.ml32m_rated")).as_undirected()
q28_components = q28_graph.connectedComponents(
    algorithm="graphframes", broadcastThreshold=-1, use_local_checkpoints=True)
q28_result = (q28_components.groupBy(F.col("component").alias("componentId"))
    .agg(F.count("*").alias("componentSize")).orderBy(F.desc("componentSize")))
q28_result.show(20, truncate=False)
q28_components.unpersist()

Note: Cannot be expressed in standard GQL because the GQL standard covers graph pattern matching and data manipulation only; it defines no graph algorithm procedures such as connected components.

Q29. Strongly Connected Components of Directed Ratings

Subcategory: T5.4 Strongly Connected Components

This query treats RATED edges as directed from users to movies and reports strongly connected components.

Skipped: The schema contains no undirected edges, so SCC is not meaningful for this workload. Therefore, Q29 is skipped.

Q30. Louvain Communities in the Rating Graph

Subcategory: T5.5 Community Detection

This query detects communities in the undirected User-Movie rating graph using Louvain.

python

q30_vertices = (spark.table("ml32m.ml32m_user").unionByName(
    spark.table("ml32m.ml32m_movie"), allowMissingColumns=True))
q30_graph = GraphFrame(q30_vertices, spark.table("ml32m.ml32m_rated"))
q30_undirected_graph = q30_graph.as_undirected()
q30_result = spark.createDataFrame([
    (False, "GraphFrames 0.12.1 supports as_undirected(), but has no Louvain implementation.")
], ["supported", "reason"])
q30_result.show(truncate=False)
# Non-equivalent alternative (intentionally not executed):
# q30_lpa = q30_undirected_graph.labelPropagation(maxIter=30)

Note: Cannot be expressed in standard GQL because the GQL standard covers graph pattern matching and data manipulation only; it defines no graph algorithm procedures such as community detection (Louvain).

Q31. 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

Subcategory: T4.2 Grouped Node Degree Statistics by Label; T4.3 Ungrouped Edge Count and Property Statistics

This query retrieves the top 10 most rated movies that belong to at least 3 genres and have received at least 50 user ratings, showing their titles, genres, and rating counts.

python

q31_vertices = (spark.table("ml32m.ml32m_user").withColumn("type", F.lit("User"))
    .unionByName(spark.table("ml32m.ml32m_movie").withColumn("type", F.lit("Movie")), allowMissingColumns=True)
    .unionByName(spark.table("ml32m.ml32m_genre").withColumn("type", F.lit("Genre")), allowMissingColumns=True))
q31_edges = (spark.table("ml32m.ml32m_rated").withColumn("relationship", F.lit("RATED"))
    .unionByName(spark.table("ml32m.ml32m_has_genre").withColumn("relationship", F.lit("HAS_GENRE")), allowMissingColumns=True))
q31_graph = GraphFrame(q31_vertices, q31_edges)
q31_result = (q31_graph.find("(u)-[r]->(m); (m)-[hg]->(g)")
    .filter((F.col("r.relationship") == "RATED") & (F.col("hg.relationship") == "HAS_GENRE"))
    .groupBy(F.col("m.id").alias("movieVertexId"), F.col("m.title").alias("movie"))
    .agg(F.collect_set(F.col("g.name")).alias("genres"),
         F.countDistinct(F.col("r.src")).alias("ratingCount"))
    .filter((F.size("genres") >= 3) & (F.col("ratingCount") >= 50))
    .select("movie", "genres", "ratingCount").orderBy(F.desc("ratingCount")).limit(10))
q31_result.show(10, truncate=False)

Equivalent query in GQL standard:

gql

MATCH (m:ml32m_movie)-[:ml32m_has_genre]->(g:ml32m_genre)
MATCH (u:ml32m_user)-[r:ml32m_rated]->(m)
WITH m, collect(DISTINCT g.name) AS genres, count(r) AS ratingCount
WHERE size(genres) >= 3 AND ratingCount >= 50
RETURN m.title AS movie, genres, ratingCount
ORDER BY ratingCount DESC
LIMIT 10

Q32. Find top-20 movies with the highest number of genres + tags, return this number, list of their genres, and list of their tags

Subcategory: Other

This query finds the top 20 movies with the largest combined count of unique genres and tags, returning the total count along with their genre and tag lists.

python

q32_movies = spark.table("ml32m.ml32m_movie")
q32_genre_vertices = q32_movies.unionByName(spark.table("ml32m.ml32m_genre"), allowMissingColumns=True)
q32_genre_graph = GraphFrame(q32_genre_vertices, spark.table("ml32m.ml32m_has_genre"))
q32_genres = q32_genre_graph.aggregateMessages(
    F.collect_set(AM.msg).alias("genres"), sendToSrc=AM.dst["name"])
q32_tag_vertices = (spark.table("ml32m.ml32m_user").unionByName(q32_movies, allowMissingColumns=True))
q32_tag_graph = GraphFrame(q32_tag_vertices, spark.table("ml32m.ml32m_tagged"))
q32_tags = q32_tag_graph.aggregateMessages(
    F.collect_set(AM.msg).alias("tags"), sendToDst=AM.edge["tag"])
q32_empty = F.expr("array()").cast("array<string>")
q32_result = (q32_movies.select("id", "title").join(q32_genres, "id", "left").join(q32_tags, "id", "left")
    .withColumn("genres", F.coalesce(F.col("genres"), q32_empty))
    .withColumn("tags", F.coalesce(F.col("tags"), q32_empty))
    .withColumn("total", F.size("genres") + F.size("tags"))
    .select(F.col("title").alias("movieTitle"), "total", "genres", "tags")
    .orderBy(F.desc("total")).limit(20))
q32_result.show(20, truncate=False)

Equivalent query in GQL standard:

gql

MATCH (m:ml32m_movie)
OPTIONAL MATCH (m)-[:ml32m_has_genre]->(g:ml32m_genre)
OPTIONAL MATCH (:ml32m_user)-[t:ml32m_tagged]->(m)
WITH m,
     collect(DISTINCT g.name) AS genres,
     collect(DISTINCT t.tag) AS tags,
     size(collect(DISTINCT g.name)) + size(collect(DISTINCT t.tag)) AS total
ORDER BY total DESC
LIMIT 20
RETURN m.title AS movieTitle, total, genres, tags

Q33. 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)

Subcategory: Other

This query retrieves the top 10 user-movie pairs where the user rated a movie before tagging it, sorted by the rating timestamp.

python

q33_vertices = (spark.table("ml32m.ml32m_user").withColumn("type", F.lit("User"))
    .unionByName(spark.table("ml32m.ml32m_movie").withColumn("type", F.lit("Movie")), allowMissingColumns=True))
q33_edges = (spark.table("ml32m.ml32m_rated").withColumn("relationship", F.lit("RATED"))
    .unionByName(spark.table("ml32m.ml32m_tagged").withColumn("relationship", F.lit("TAGGED")), allowMissingColumns=True))
q33_graph = GraphFrame(q33_vertices, q33_edges)
q33_result = (q33_graph.find("(u)-[r]->(m); (u)-[t]->(m)")
    .filter((F.col("r.relationship") == "RATED") & (F.col("t.relationship") == "TAGGED")
            & (F.col("r.timestamp") < F.col("t.timestamp")))
    .groupBy(F.col("u.id").alias("userVertexId"), F.col("u.userId").alias("userId"),
             F.col("m.id").alias("movieVertexId"), F.col("m.title").alias("movieTitle"),
             F.col("r.rating").alias("rating"), F.col("r.timestamp").alias("ratingTime"))
    .agg(F.collect_set(F.col("t.tag")).alias("tags"))
    .select("userId", "movieTitle", "rating", "tags", "ratingTime")
    .orderBy(F.desc("ratingTime")).limit(10)
    .drop("ratingTime"))
q33_result.show(10, truncate=False)

Equivalent query in GQL standard:

gql

MATCH (u:ml32m_user)-[r:ml32m_rated]->(m:ml32m_movie)<-[t:ml32m_tagged]-(u)
WHERE r.timestamp < t.timestamp
WITH u.userId AS userId, m.title AS movieTitle, r.rating AS rating, r.timestamp AS ratingTime,
     collect(DISTINCT t.tag) AS tags
RETURN userId, movieTitle, rating, tags
ORDER BY ratingTime DESC
LIMIT 10