← Back to MovieLens Tag Genome Dataset

GraphFrames Implementation

Deployment and Query logic for GraphFrames on a Spark cluster with 7 worker nodes.

Schema Diagram and Graph Statistics

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

MovieLens Tag Genome Neo4j Schema

Node Statistics

Node Label Count
Actor 122,974
Director 33,660
Movie 84,661
Review 2,623,824
Survey 58,903
Tag 1,094
User 200,948

Edge Statistics

Relationship Count
(:Survey)-[:ABOUT_MOVIE]-(:Movie) 58,903
(:Survey)-[:ABOUT_TAG]-(:Tag) 58,903
(:Actor)-[:ACTED_IN]-(:Movie) 371,629
(:User)-[:ANSWERED]-(:Survey) 58,903
(:Director)-[:DIRECTED]-(:Movie) 81,511
(:Movie)-[:HAS_REVIEW]-(:Review) 2,623,824
(:Movie)-[:HAS_TAG]-(:Tag) 212,092
(:User)-[:RATED]-(:Movie) 18,874,784

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.

DataFrames

We defined a few helper functions to help us process the raw data files.

python 
def read_json_lines(filename, schema):
    return spark.read.option('mode', 'FAILFAST').schema(schema).json(f'{INPUT_ROOT}/{filename}')

def vertex_id(kind, column):
    return F.concat(F.lit(f'{kind}:'), column.cast('string'))

def clean_person_name(column):
    return F.trim(F.regexp_replace(column, r'[\t\r\n]+', ' '))

Define Nodes

First we define the Movie node.

python 
MOVIE_SCHEMA = StructType([
    StructField('item_id', LongType(), False), StructField('title', StringType(), True),
    StructField('directedBy', StringType(), True), StructField('starring', StringType(), True),
    StructField('dateAdded', StringType(), True), StructField('avgRating', DoubleType(), True),
    StructField('imdbId', StringType(), True)])
movie_raw = read_json_lines('metadata.json', MOVIE_SCHEMA)
movie_df = movie_raw.filter(F.col('item_id').isNotNull()).dropDuplicates(['item_id'])
movie_vertices_df = movie_df.select(
    vertex_id('movie', F.col('item_id')).alias('id'), F.lit('movie').alias('type'),
    'item_id', 'title', 'avgRating', 'imdbId', 'directedBy', 'starring')
movie_vertices_df.printSchema(); movie_vertices_df.show(3, truncate=80)

Then we define the Tag node.

python 
TAG_SCHEMA = StructType([StructField('tag', StringType(), True), StructField('id', LongType(), False)])
tag_raw = read_json_lines('tags.json', TAG_SCHEMA)
tag_df = tag_raw.filter(F.col('id').isNotNull()).dropDuplicates(['id'])
tag_vertices_df = tag_df.select(vertex_id('tag', F.col('id')).alias('id'),
    F.lit('tag').alias('type'), F.col('id').alias('tag_id'), 'tag')
tag_vertices_df.printSchema(); tag_vertices_df.show(3, truncate=80)

Then we define the Survey node.

python 
SURVEY_SCHEMA = StructType([
    StructField('item_id', LongType(), False), StructField('score', LongType(), True),
    StructField('tag_id', LongType(), False), StructField('user_id', LongType(), False),
    StructField('survey_id', LongType(), False)])
survey_raw = read_json_lines('surveys_id.json', SURVEY_SCHEMA)
survey_df = (survey_raw.join(F.broadcast(movie_df.select('item_id')), 'item_id', 'inner')
    .join(F.broadcast(tag_df.select(F.col('id').alias('tag_id'))), 'tag_id', 'inner')
    .dropDuplicates(['survey_id']))
survey_vertices_df = survey_df.select(vertex_id('survey', F.col('survey_id')).alias('id'),
    F.lit('survey').alias('type'), 'survey_id', 'user_id', 'item_id', 'tag_id', 'score')
survey_vertices_df.printSchema(); survey_vertices_df.show(3, truncate=80)

Then we define the User node.

python 
USER_SCHEMA = StructType([StructField('user_id', LongType(), False)])
user_rating_raw = read_json_lines('ratings.json', USER_SCHEMA)
user_survey_raw = read_json_lines('surveys_id.json', USER_SCHEMA)
user_raw = user_rating_raw.unionByName(user_survey_raw)
user_df = user_raw.filter(F.col('user_id').isNotNull()).distinct()
user_vertices_df = user_df.select(vertex_id('user', F.col('user_id')).alias('id'),
    F.lit('user').alias('type'), 'user_id')
user_vertices_df.printSchema(); user_vertices_df.show(3, truncate=80)

Then we define the Review node.

python 
REVIEW_SCHEMA = StructType([StructField('item_id', LongType(), False),
    StructField('txt', StringType(), True), StructField('review_id', LongType(), False)])
review_raw = read_json_lines('review_with_id.json', REVIEW_SCHEMA)
review_df = (review_raw.join(F.broadcast(movie_df.select('item_id')), 'item_id', 'inner')
    .dropDuplicates(['review_id']))
review_vertices_df = review_df.select(vertex_id('review', F.col('review_id')).alias('id'),
    F.lit('review').alias('type'), 'review_id', 'item_id', F.col('txt').alias('text'))
review_vertices_df.printSchema(); review_vertices_df.show(3, truncate=80)

Then we define the Director node.

python 
DIRECTOR_SCHEMA = StructType([StructField('item_id', LongType(), False),
    StructField('directedBy', StringType(), True)])
director_raw = read_json_lines('metadata.json', DIRECTOR_SCHEMA)
director_df = (director_raw.filter(F.col('directedBy').isNotNull())
    .select('item_id', F.explode(F.split('directedBy', ',')).alias('raw_name'))
    .select('item_id', clean_person_name(F.col('raw_name')).alias('name'))
    .filter(F.length('name') > 0).dropDuplicates(['item_id', 'name']))
director_vertices_df = (director_df.select('name').distinct()
    .select(vertex_id('director', F.col('name')).alias('id'), F.lit('director').alias('type'), 'name'))
director_vertices_df.printSchema(); director_vertices_df.show(3, truncate=80)

Then we define the Actor node.

python 
ACTOR_SCHEMA = StructType([StructField('item_id', LongType(), False),
    StructField('starring', StringType(), True)])
actor_raw = read_json_lines('metadata.json', ACTOR_SCHEMA)
actor_df = (actor_raw.filter(F.col('starring').isNotNull())
    .select('item_id', F.explode(F.split('starring', ',')).alias('raw_name'))
    .select('item_id', clean_person_name(F.col('raw_name')).alias('name'))
    .filter(F.length('name') > 0).dropDuplicates(['item_id', 'name']))
actor_vertices_df = (actor_df.select('name').distinct()
    .select(vertex_id('actor', F.col('name')).alias('id'), F.lit('actor').alias('type'), 'name'))
actor_vertices_df.printSchema(); actor_vertices_df.show(3, truncate=80)

Define Edges

Next step is to define edges. First we define the RATED edge.

python 
RATED_SCHEMA = StructType([StructField('item_id', LongType(), False),
StructField('user_id', LongType(), False), StructField('rating', DoubleType(), True)])
rated_raw = read_json_lines('ratings.json', RATED_SCHEMA)
rated_df = rated_raw.join(F.broadcast(movie_df.select('item_id')), 'item_id', 'inner')
rated_edges_df = rated_df.select(vertex_id('user', F.col('user_id')).alias('src'),
vertex_id('movie', F.col('item_id')).alias('dst'), F.lit('rated').alias('etype'), 'rating')
rated_edges_df.printSchema(); rated_edges_df.show(3, truncate=80)

Then we define the HAS_TAG edge.

python 
HAS_TAG_SCHEMA = StructType([StructField('item_id', LongType(), False),
    StructField('tag_id', LongType(), False), StructField('num', LongType(), True)])
has_tag_raw = read_json_lines('tag_count.json', HAS_TAG_SCHEMA)
has_tag_df = (has_tag_raw.join(F.broadcast(movie_df.select('item_id')), 'item_id', 'inner')
    .join(F.broadcast(tag_df.select(F.col('id').alias('tag_id'))), 'tag_id', 'inner')
    .dropDuplicates(['item_id', 'tag_id']))
has_tag_edges_df = has_tag_df.select(vertex_id('movie', F.col('item_id')).alias('src'),
    vertex_id('tag', F.col('tag_id')).alias('dst'), F.lit('has_tag').alias('etype'),
    F.col('num').alias('count'))
has_tag_edges_df.printSchema(); has_tag_edges_df.show(3, truncate=80)

Then we define the HAS_REVIEW edge.

python 
HAS_REVIEW_SCHEMA = StructType([StructField('item_id', LongType(), False),
    StructField('review_id', LongType(), False)])
has_review_raw = read_json_lines('review_with_id.json', HAS_REVIEW_SCHEMA)
has_review_df = (has_review_raw.join(F.broadcast(movie_df.select('item_id')), 'item_id', 'inner')
    .dropDuplicates(['review_id']))
has_review_edges_df = has_review_df.select(vertex_id('movie', F.col('item_id')).alias('src'),
    vertex_id('review', F.col('review_id')).alias('dst'), F.lit('has_review').alias('etype'))
has_review_edges_df.printSchema(); has_review_edges_df.show(3, truncate=80)

Then we define the ANSWERED edge.

python 
ANSWERED_SCHEMA = StructType([StructField('survey_id', LongType(), False),
    StructField('user_id', LongType(), False)])
answered_raw = read_json_lines('surveys_id.json', ANSWERED_SCHEMA)
answered_df = answered_raw.join(survey_df.select('survey_id'), 'survey_id', 'inner').dropDuplicates(['survey_id'])
answered_edges_df = answered_df.select(vertex_id('user', F.col('user_id')).alias('src'),
    vertex_id('survey', F.col('survey_id')).alias('dst'), F.lit('answered').alias('etype'))
answered_edges_df.printSchema(); answered_edges_df.show(3, truncate=80)

Then we define the ABOUT_MOVIE edge.

python 
ABOUT_MOVIE_SCHEMA = StructType([StructField('survey_id', LongType(), False),
    StructField('item_id', LongType(), False)])
about_movie_raw = read_json_lines('surveys_id.json', ABOUT_MOVIE_SCHEMA)
about_movie_df = about_movie_raw.join(survey_df.select('survey_id'), 'survey_id', 'inner').dropDuplicates(['survey_id'])
about_movie_edges_df = about_movie_df.select(vertex_id('survey', F.col('survey_id')).alias('src'),
    vertex_id('movie', F.col('item_id')).alias('dst'), F.lit('about_movie').alias('etype'))
about_movie_edges_df.printSchema(); about_movie_edges_df.show(3, truncate=80)

Then we define the ABOUT_TAG edge.

python 
ABOUT_TAG_SCHEMA = StructType([StructField('survey_id', LongType(), False),
    StructField('tag_id', LongType(), False)])
about_tag_raw = read_json_lines('surveys_id.json', ABOUT_TAG_SCHEMA)
about_tag_df = about_tag_raw.join(survey_df.select('survey_id'), 'survey_id', 'inner').dropDuplicates(['survey_id'])
about_tag_edges_df = about_tag_df.select(vertex_id('survey', F.col('survey_id')).alias('src'),
    vertex_id('tag', F.col('tag_id')).alias('dst'), F.lit('about_tag').alias('etype'))
about_tag_edges_df.printSchema(); about_tag_edges_df.show(3, truncate=80)

Then we define the DIRECTED edge.

python 
DIRECTED_SCHEMA = StructType([StructField('item_id', LongType(), False),
    StructField('directedBy', StringType(), True)])
directed_raw = read_json_lines('metadata.json', DIRECTED_SCHEMA)
directed_df = (directed_raw.filter(F.col('directedBy').isNotNull())
    .select('item_id', F.explode(F.split('directedBy', ',')).alias('raw_name'))
    .select('item_id', clean_person_name(F.col('raw_name')).alias('name'))
    .filter(F.length('name') > 0)
    .join(F.broadcast(movie_df.select('item_id')), 'item_id', 'inner')
    .dropDuplicates(['item_id', 'name']))
directed_edges_df = directed_df.select(vertex_id('director', F.col('name')).alias('src'),
    vertex_id('movie', F.col('item_id')).alias('dst'), F.lit('directed').alias('etype'))
directed_edges_df.printSchema(); directed_edges_df.show(3, truncate=80)

Then we define the ACTED_IN edge.

python 
ACTED_IN_SCHEMA = StructType([StructField('item_id', LongType(), False),
    StructField('starring', StringType(), True)])
acted_in_raw = read_json_lines('metadata.json', ACTED_IN_SCHEMA)
acted_in_df = (acted_in_raw.filter(F.col('starring').isNotNull())
    .select('item_id', F.explode(F.split('starring', ',')).alias('raw_name'))
    .select('item_id', clean_person_name(F.col('raw_name')).alias('name'))
    .filter(F.length('name') > 0)
    .join(F.broadcast(movie_df.select('item_id')), 'item_id', 'inner')
    .dropDuplicates(['item_id', 'name']))
acted_in_edges_df = acted_in_df.select(vertex_id('actor', F.col('name')).alias('src'),
    vertex_id('movie', F.col('item_id')).alias('dst'), F.lit('acted_in').alias('etype'))
acted_in_edges_df.printSchema(); acted_in_edges_df.show(3, truncate=80)

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.

python 
vertex_source_dfs = {
    'movie': movie_vertices_df, 'tag': tag_vertices_df, 'survey': survey_vertices_df,
    'user': user_vertices_df, 'review': review_vertices_df,
    'director': director_vertices_df, 'actor': actor_vertices_df}
edge_source_dfs = {
    'rated': rated_edges_df, 'has_tag': has_tag_edges_df, 'has_review': has_review_edges_df,
    'answered': answered_edges_df, 'about_movie': about_movie_edges_df,
    'about_tag': about_tag_edges_df, 'directed': directed_edges_df, 'acted_in': acted_in_edges_df}

Finally, we can create the whole graph schema.

python 
vertices = union_all(list(vertex_source_dfs.values()))
edges = union_all(list(edge_source_dfs.values()))
g = GraphFrame(vertices, edges)

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 Selected User

Subcategory: T1.1 One-Hop Neighbors

This query lists the movies directly connected to a selected user through outgoing RATED edges.

python 
q1_g = GraphFrame(vertices_df, edges_df)

q1_result = (
    q1_g.find('(u)-[r]->(m)')
    .filter(
        (F.col('u.type') == 'user') &
        (F.col('u.user_id') == 997206) &
        (F.col('r.etype') == 'rated') &
        (F.col('m.type') == 'movie')
    )
    .select(
        F.col('m.item_id').alias('MovieID'),
        F.col('m.title').alias('Movie'),
        F.col('r.rating').alias('Rating'),
    )
    .orderBy(F.desc('Rating'))
)

q1_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (u:mlgenome_user {user_id: 1})-[r:mlgenome_rated]->(m:mlgenome_movie)
RETURN m.item_id AS MovieID, m.title AS Movie, r.rating AS Rating
ORDER BY Rating DESC
LIMIT 20

Q2. Two-Hop Rating Neighborhood of a User

Subcategory: T1.2 k-Hop Neighborhood

This query finds distinct users reachable within two RATED hops from a selected user through shared movies.

python 
q2_g = GraphFrame(vertices_df, edges_df)

q2_hop1 = (
    q2_g.find('(u)-[e1]-(n)')
    .filter(
        (F.col('u.type') == 'user') &
        (F.col('u.user_id') == 997206) &
        (F.col('e1.etype') == 'rated') &
        (F.col('n.id') != F.col('u.id'))
    )
    .select(
        F.array(F.col('n.type')).alias('NodeLabels'),
        F.coalesce(
            F.col('n.title'),
            F.col('n.user_id').cast('string'),
            F.col('n.id'),
        ).alias('Neighbor'),
        F.lit(1).alias('HopCount'),
    )
)

q2_hop2 = (
    q2_g.find('(u)-[e1]-(v1); (v1)-[e2]-(n)')
    .filter(
        (F.col('u.type') == 'user') &
        (F.col('u.user_id') == 997206) &
        (F.col('e1.etype') == 'rated') &
        (F.col('e2.etype') == 'rated') &
        (F.col('e1._eid') != F.col('e2._eid')) &
        (F.col('n.id') != F.col('u.id'))
    )
    .select(
        F.array(F.col('n.type')).alias('NodeLabels'),
        F.coalesce(
            F.col('n.title'),
            F.col('n.user_id').cast('string'),
            F.col('n.id'),
        ).alias('Neighbor'),
        F.lit(2).alias('HopCount'),
    )
)

q2_result = (
    q2_hop1.unionByName(q2_hop2)
    .dropDuplicates(['NodeLabels', 'Neighbor', 'HopCount'])
    .orderBy('HopCount', 'Neighbor')
)

q2_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH p = (u:mlgenome_user {user_id: 1})-[:mlgenome_rated]-{1,2}(n)
WHERE n != u
RETURN DISTINCT labels(n) AS NodeLabels,
       coalesce(n.title, toString(n.user_id)) AS Neighbor,
       length(p) AS HopCount
ORDER BY HopCount ASC, Neighbor
LIMIT 50

Q3. Users Exactly Two Hops Away Through Shared Movies

Subcategory: T1.3 Exact kth-Hop Neighbors

This query returns users exactly two RATED hops away from a selected user, meaning users who rated at least one same movie.

python 
q3_g = GraphFrame(vertices_df, edges_df)

q3_result = (
    q3_g.find('(u)-[r1]->(m); (other)-[r2]->(m)')
    .filter(
        (F.col('u.type') == 'user') &
        (F.col('u.user_id') == 997206) &
        (F.col('other.type') == 'user') &
        (F.col('other.id') != F.col('u.id')) &
        (F.col('m.type') == 'movie') &
        (F.col('r1.etype') == 'rated') &
        (F.col('r2.etype') == 'rated')
    )
    .groupBy(F.col('other.user_id').alias('SimilarUser'))
    .agg(F.countDistinct('m.id').alias('SharedMovies'))
    .orderBy(F.desc('SharedMovies'))
)

q3_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (u:mlgenome_user {user_id: 1})-[:mlgenome_rated]->(m:mlgenome_movie)<-[:mlgenome_rated]-(other:mlgenome_user)
WHERE other != u
RETURN other.user_id AS SimilarUser, count(DISTINCT m) AS SharedMovies
ORDER BY SharedMovies DESC
LIMIT 20

Q4. High-Rated One-Hop Movie Neighborhood

Subcategory: T1.4 Neighborhood with Property Filter

This query searches a selected user's immediate movie neighborhood while filtering on both the RATED edge value and the movie's average rating.

python 
q4_g = GraphFrame(vertices_df, edges_df)

q4_result = (
    q4_g.find('(u)-[r]->(m)')
    .filter(
        (F.col('u.type') == 'user') &
        (F.col('u.user_id') == 997206) &
        (F.col('r.etype') == 'rated') &
        (F.col('m.type') == 'movie') &
        (F.col('r.rating') >= 4.0) &
        (F.col('m.avgRating') >= 4.0)
    )
    .select(
        F.col('m.item_id').alias('MovieID'),
        F.col('m.title').alias('Movie'),
        F.col('r.rating').alias('UserRating'),
        F.col('m.avgRating').alias('AverageRating'),
    )
    .orderBy(F.desc('UserRating'), F.desc('AverageRating'))
)

q4_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (u:mlgenome_user {user_id: 1})-[r:mlgenome_rated]->(m:mlgenome_movie)
WHERE r.rating >= 4.0 AND m.avgRating >= 4.0
RETURN m.item_id AS MovieID, m.title AS Movie, r.rating AS UserRating, m.avgRating AS AverageRating
ORDER BY UserRating DESC, AverageRating DESC
LIMIT 20

Q5. Users Who Did NOT Participate in Any Survey

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

This query identifies users who have not participated in any tag genome surveys.

python 
q5_g = GraphFrame(vertices_df, edges_df)

# filterEdges/dropIsolatedVertices obtains the vertices incident to ANSWERED edges.
# subtract is used only for the NOT-EXISTS set difference; GraphFrames has no
# negative-edge motif operator.
q5_answered_graph = q5_g.filterEdges("etype = 'answered'").dropIsolatedVertices()

q5_all_users = q5_g.vertices.filter(F.col('type') == 'user').select('id', 'user_id')
q5_answered_users = (
    q5_answered_graph.vertices
    .filter(F.col('type') == 'user')
    .select('id', 'user_id')
)

q5_result = (
    q5_all_users.subtract(q5_answered_users)
    .select(F.col('user_id').alias('UserID'))
)

q5_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (u:mlgenome_user)
WHERE NOT (u)-[:mlgenome_answered]->(:mlgenome_survey)
RETURN u.user_id AS UserID
LIMIT 20

Q6. High-Count Movie Tag Edges

Subcategory: T2.2 Edge Filtering

This query filters HAS_TAG edges by their count property to find strong movie-tag associations.

python 
q6_g = GraphFrame(vertices_df, edges_df)

q6_result = (
    q6_g.find('(m)-[ht]->(t)')
    .filter(
        (F.col('m.type') == 'movie') &
        (F.col('ht.etype') == 'has_tag') &
        (F.col('t.type') == 'tag') &
        (F.col('ht.count') >= 20)
    )
    .select(
        F.col('m.title').alias('Movie'),
        F.col('t.tag').alias('Tag'),
        F.col('ht.count').alias('TagCount'),
    )
    .orderBy(F.desc('TagCount'))
)

q6_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (m:mlgenome_movie)-[ht:mlgenome_has_tag]->(t:mlgenome_tag)
WHERE ht.count >= 20
RETURN m.title AS Movie, t.tag AS Tag, ht.count AS TagCount
ORDER BY TagCount DESC
LIMIT 20

Q7. User Reachability to a Tag Through Rated Movies

Subcategory: T2.3 Reachability between Two Nodes

This query checks whether a selected user can reach a selected tag through movies they rated.

python 
q7_g = GraphFrame(vertices_df, edges_df)

q7_paths = (
    q7_g.find('(u)-[r]->(m); (m)-[ht]->(t)')
    .filter(
        (F.col('u.type') == 'user') &
        (F.col('u.user_id') == 997206) &
        (F.col('r.etype') == 'rated') &
        (F.col('m.type') == 'movie') &
        (F.col('ht.etype') == 'has_tag') &
        (F.col('t.type') == 'tag') &
        (F.col('t.tag_id') == 22)
    )
)

q7_tag = (
    q7_g.vertices
    .filter((F.col('type') == 'tag') & (F.col('tag_id') == 22))
    .select('tag')
    .first()
)
q7_reachable = q7_paths.limit(1).count() > 0

q7_result = spark.createDataFrame(
    [(997206, q7_tag['tag'] if q7_tag else None, q7_reachable)],
    'UserID long, Tag string, Reachable boolean',
)

q7_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (u:mlgenome_user {user_id: 1})
MATCH (t:mlgenome_tag)
WITH u, t
LIMIT 1
OPTIONAL MATCH p = (u)-[:mlgenome_rated]->(:mlgenome_movie)-[:mlgenome_has_tag]->(t)
RETURN u.user_id AS UserID, t.tag AS Tag, count(p) > 0 AS Reachable

Q8. Shortest User-to-Tag Explanation Path

Subcategory: T2.4 Shortest Path

This query finds the shortest path from a user to a tag over ratings and movie-tag relationships.

python 
q8_g = GraphFrame(vertices_df, edges_df)

q8_bfs = q8_g.bfs(
    fromExpr="type = 'user' AND user_id = 997206",
    toExpr="type = 'tag' AND tag_id = 22",
    edgeFilter="etype IN ('rated', 'has_tag')",
    maxPathLength=4,
)

# With these directed edge types, a successful path is User -> Movie -> Tag.
q8_result = q8_bfs.select(
    F.array(
        F.col('from.user_id').cast('string'),
        F.col('v1.title'),
        F.col('to.tag'),
    ).alias('Path'),
    F.lit(2).alias('Hops'),
)

q8_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH p = SHORTEST (u:mlgenome_user {user_id: 1})-[:mlgenome_rated|mlgenome_has_tag]-{1,4}(t:mlgenome_tag)
RETURN p

Q9. Simple Co-Rating Paths Between Users

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

This query enumerates simple RATED paths up to four hops between one user and another user in the rating graph.

python 
q9_g = GraphFrame(vertices_df, edges_df)

# Neo4j selects an arbitrary target with LIMIT 1, so this translation intentionally
# keeps the same unspecified target choice. In a User--Movie RATED graph, a
# User-to-User path of length 1..3 can only have length 2.
q9_target = (
    q9_g.vertices
    .filter((F.col('type') == 'user') & (F.col('user_id') != 997206))
    .select('id', 'user_id')
    .first()
)

q9_result = (
    q9_g.find('(source)-[r1]-(m); (m)-[r2]-(target)')
    .filter(
        (F.col('source.type') == 'user') &
        (F.col('source.user_id') == 997206) &
        (F.col('target.id') == q9_target['id']) &
        (F.col('m.type') == 'movie') &
        (F.col('r1.etype') == 'rated') &
        (F.col('r2.etype') == 'rated') &
        (F.col('r1._eid') != F.col('r2._eid'))
    )
    .select(
        F.array(
            F.col('source.user_id').cast('string'),
            F.col('m.title'),
            F.col('target.user_id').cast('string'),
        ).alias('Path'),
        F.lit(2).alias('Hops'),
    )
)

q9_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH p = ACYCLIC (source:mlgenome_user {user_id: 1})-[:mlgenome_rated]-{1,4}(target:mlgenome_user)
WHERE target != source
RETURN p
LIMIT 20

Q10. User-Movie Co-Rating Squares

Subcategory: T3.2 Square Detection

This query finds two users who both rated the same pair of movies, forming a four-node square in the user-movie rating graph.

python 
q10_g = GraphFrame(vertices_df, edges_df)

q10_result = (
    q10_g.find(
        '(u1)-[r1]->(m1); (u2)-[r2]->(m1); '
        '(u2)-[r3]->(m2); (u1)-[r4]->(m2)'
    )
    .filter(
        (F.col('u1.type') == 'user') &
        (F.col('u2.type') == 'user') &
        (F.col('m1.type') == 'movie') &
        (F.col('m2.type') == 'movie') &
        (F.col('r1.etype') == 'rated') &
        (F.col('r2.etype') == 'rated') &
        (F.col('r3.etype') == 'rated') &
        (F.col('r4.etype') == 'rated') &
        (F.col('u1.user_id') < F.col('u2.user_id')) &
        (F.col('m1.item_id') < F.col('m2.item_id'))
    )
    .select(
        F.col('u1.user_id').alias('User1'),
        F.col('u2.user_id').alias('User2'),
        F.col('m1.title').alias('Movie1'),
        F.col('m2.title').alias('Movie2'),
    )
)

q10_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (u1:mlgenome_user)-[:mlgenome_rated]->(m1:mlgenome_movie)<-[:mlgenome_rated]-(u2:mlgenome_user)-[:mlgenome_rated]->(m2:mlgenome_movie)<-[:mlgenome_rated]-(u1)
WHERE u1.user_id < u2.user_id AND m1.item_id < m2.item_id
RETURN u1.user_id AS User1, u2.user_id AS User2, m1.title AS Movie1, m2.title AS Movie2
LIMIT 20

Q11. Movie Star Pattern Across Cast, Tags, and Reviews

Subcategory: T3.3 Star Pattern Query

This query treats each movie as a star center connected to actors, tags, and reviews, then returns the busiest centers.

python 
q11_g = GraphFrame(vertices_df, edges_df)

q11_to_src = F.struct(
    AM.src['type'].alias('receiver_type'),
    AM.src['title'].alias('receiver_title'),
    AM.edge['etype'].alias('etype'),
    AM.dst['id'].alias('neighbor_id'),
)
q11_to_dst = F.struct(
    AM.dst['type'].alias('receiver_type'),
    AM.dst['title'].alias('receiver_title'),
    AM.edge['etype'].alias('etype'),
    AM.src['id'].alias('neighbor_id'),
)

q11_aggregated = q11_g.aggregateMessages(
    [
        F.first(AM.msg['receiver_type'], ignorenulls=True).alias('receiver_type'),
        F.first(AM.msg['receiver_title'], ignorenulls=True).alias('receiver_title'),
        F.countDistinct(
            F.when(AM.msg['etype'] == 'acted_in', AM.msg['neighbor_id'])
        ).alias('ActorCount'),
        F.countDistinct(
            F.when(AM.msg['etype'] == 'has_tag', AM.msg['neighbor_id'])
        ).alias('TagCount'),
        F.countDistinct(
            F.when(AM.msg['etype'] == 'has_review', AM.msg['neighbor_id'])
        ).alias('ReviewCount'),
    ],
    sendToSrc=[q11_to_src],
    sendToDst=[q11_to_dst],
)

q11_result = (
    q11_aggregated
    .filter(
        (F.col('receiver_type') == 'movie') &
        (F.col('ActorCount') >= 2) &
        (F.col('TagCount') >= 2)
    )
    .select(
        F.col('receiver_title').alias('Movie'),
        'ActorCount',
        'TagCount',
        'ReviewCount',
        (F.col('ActorCount') + F.col('TagCount') + F.col('ReviewCount')).alias('StarDegree'),
    )
    .orderBy(F.desc('StarDegree'))
)

q11_result.show(truncate=False)
q11_aggregated.unpersist()

Equivalent query in GQL standard:

gql 
MATCH (m:mlgenome_movie)
OPTIONAL MATCH (a:mlgenome_actor)-[:mlgenome_acted_in]->(m)
OPTIONAL MATCH (m)-[:mlgenome_has_tag]->(t:mlgenome_tag)
OPTIONAL MATCH (m)-[:mlgenome_has_review]->(r:mlgenome_review)
WITH m, count(DISTINCT a) AS ActorCount, count(DISTINCT t) AS TagCount, count(DISTINCT r) AS ReviewCount
WHERE ActorCount >= 2 AND TagCount >= 2
RETURN m.title AS Movie, ActorCount, TagCount, ReviewCount, ActorCount + TagCount + ReviewCount AS StarDegree
ORDER BY StarDegree DESC
LIMIT 20

Q12. Four-Actor Co-Starring Cliques

Subcategory: T3.4 4-Clique

This query finds four actors who co-star in the same movie, representing a four-clique in the projected co-actor graph.

python 
q12_g = GraphFrame(vertices_df, edges_df)

q12_result = (
    q12_g.find(
        '(a1)-[e1]->(m); (a2)-[e2]->(m); '
        '(a3)-[e3]->(m); (a4)-[e4]->(m)'
    )
    .filter(
        (F.col('a1.type') == 'actor') &
        (F.col('a2.type') == 'actor') &
        (F.col('a3.type') == 'actor') &
        (F.col('a4.type') == 'actor') &
        (F.col('m.type') == 'movie') &
        (F.col('e1.etype') == 'acted_in') &
        (F.col('e2.etype') == 'acted_in') &
        (F.col('e3.etype') == 'acted_in') &
        (F.col('e4.etype') == 'acted_in') &
        (F.col('a1.name') < F.col('a2.name')) &
        (F.col('a2.name') < F.col('a3.name')) &
        (F.col('a3.name') < F.col('a4.name'))
    )
    .select(
        F.col('m.title').alias('Movie'),
        F.col('a1.name').alias('Actor1'),
        F.col('a2.name').alias('Actor2'),
        F.col('a3.name').alias('Actor3'),
        F.col('a4.name').alias('Actor4'),
    )
)

q12_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (a1:mlgenome_actor)-[:mlgenome_acted_in]->(m:mlgenome_movie)<-[:mlgenome_acted_in]-(a2:mlgenome_actor),
      (a3:mlgenome_actor)-[:mlgenome_acted_in]->(m)<-[:mlgenome_acted_in]-(a4:mlgenome_actor)
WHERE a1.name < a2.name AND a2.name < a3.name AND a3.name < a4.name
RETURN m.title AS Movie, a1.name AS Actor1, a2.name AS Actor2, a3.name AS Actor3, a4.name AS Actor4
LIMIT 20

Q13. Most Similar Users Based on Ratings for the Same Movies (Find pairs of users with overlapping movie preferences. (Collaborative filtering foundation))

Subcategory: T3.5 Two-hop Join Pattern

This query finds pairs of users who have rated the same movies similarly (within a difference of 1 rating point) and share at least 5 such movie preferences.

python 
q13_g = GraphFrame(vertices_df, edges_df)

q13_result = (
    q13_g.find('(u1)-[r1]->(m); (u2)-[r2]->(m)')
    .filter(
        (F.col('u1.type') == 'user') &
        (F.col('u2.type') == 'user') &
        (F.col('u1.id') != F.col('u2.id')) &
        (F.col('m.type') == 'movie') &
        (F.col('r1.etype') == 'rated') &
        (F.col('r2.etype') == 'rated') &
        (F.abs(F.col('r1.rating') - F.col('r2.rating')) <= 1)
    )
    .groupBy(
        F.col('u1.user_id').alias('User1'),
        F.col('u2.user_id').alias('User2'),
    )
    .agg(F.countDistinct('m.id').alias('SharedMovies'))
    .filter(F.col('SharedMovies') >= 5)
    .orderBy(F.desc('SharedMovies'))
)

q13_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (u1:mlgenome_user)-[r1:mlgenome_rated]->(m:mlgenome_movie)<-[r2:mlgenome_rated]-(u2:mlgenome_user)
WHERE u1 != u2 AND abs(r1.rating - r2.rating) <= 1
WITH u1.user_id AS User1, u2.user_id AS User2, count(m) AS SharedMovies
WHERE SharedMovies >= 5
RETURN User1, User2, SharedMovies
ORDER BY SharedMovies DESC
LIMIT 10

Q14. User-Survey-Movie-Tag Line Pattern

Subcategory: T3.6 Line Pattern

This query follows a four-node line from users to surveys, then to movies, and finally to tags attached to those movies.

python 
q14_g = GraphFrame(vertices_df, edges_df)

q14_result = (
    q14_g.find('(u)-[a]->(s); (s)-[am]->(m); (m)-[ht]->(t)')
    .filter(
        (F.col('u.type') == 'user') &
        (F.col('a.etype') == 'answered') &
        (F.col('s.type') == 'survey') &
        (F.col('am.etype') == 'about_movie') &
        (F.col('m.type') == 'movie') &
        (F.col('ht.etype') == 'has_tag') &
        (F.col('t.type') == 'tag')
    )
    .select(
        F.col('u.user_id').alias('UserID'),
        F.col('s.survey_id').alias('SurveyID'),
        F.col('m.title').alias('Movie'),
        F.col('t.tag').alias('Tag'),
        F.col('s.score').alias('SurveyScore'),
    )
)

q14_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (u:mlgenome_user)-[:mlgenome_answered]->(s:mlgenome_survey)-[:mlgenome_about_movie]->(m:mlgenome_movie)-[:mlgenome_has_tag]->(t:mlgenome_tag)
RETURN u.user_id AS UserID, s.survey_id AS SurveyID, m.title AS Movie, t.tag AS Tag, s.score AS SurveyScore
LIMIT 20

Q15. Movie Degree Summary Across All Relationship Types

Subcategory: T4.1 Ungrouped Node Degree Statistics by Label

This query computes ungrouped degree statistics for Movie nodes across their adjacent relationships.

python 
q15_g = GraphFrame(vertices_df, edges_df)

q15_to_src = F.struct(
    AM.src['type'].alias('receiver_type'),
    AM.dst['id'].alias('neighbor_id'),
)
q15_to_dst = F.struct(
    AM.dst['type'].alias('receiver_type'),
    AM.src['id'].alias('neighbor_id'),
)

q15_aggregated = q15_g.aggregateMessages(
    [
        F.first(AM.msg['receiver_type'], ignorenulls=True).alias('receiver_type'),
        F.countDistinct(AM.msg['neighbor_id']).alias('Degree'),
    ],
    sendToSrc=[q15_to_src],
    sendToDst=[q15_to_dst],
)

q15_nonisolated_movie_degrees = (
    q15_aggregated
    .filter(F.col('receiver_type') == 'movie')
    .select('Degree')
)

# GraphFrames has no negative motif. Use its isolated-vertex operation plus
# set subtraction to add Degree=0 movies without a table join.
q15_connected_movie_ids = (
    q15_g.dropIsolatedVertices().vertices
    .filter(F.col('type') == 'movie')
    .select('id')
)
q15_isolated_movie_degrees = (
    q15_g.vertices
    .filter(F.col('type') == 'movie')
    .select('id')
    .subtract(q15_connected_movie_ids)
    .select(F.lit(0).cast('long').alias('Degree'))
)
q15_movie_degrees = q15_nonisolated_movie_degrees.unionByName(
    q15_isolated_movie_degrees
)

q15_result = q15_movie_degrees.agg(
    F.count(F.lit(1)).alias('MovieCount'),
    F.round(F.avg('Degree'), 2).alias('AvgDegree'),
    F.min('Degree').alias('MinDegree'),
    F.max('Degree').alias('MaxDegree'),
)

q15_result.show(truncate=False)
q15_aggregated.unpersist()

Equivalent query in GQL standard:

gql 
MATCH (m:mlgenome_movie)
OPTIONAL MATCH (m)-[]-(n)
WITH m, count(DISTINCT n) AS Degree
RETURN count(m) AS MovieCount,
       round(avg(Degree), 2) AS AvgDegree,
       min(Degree) AS MinDegree,
       max(Degree) AS MaxDegree

Q16. Directors by Average Movie Rating (Filtering to avoid bias from 1-hit wonders, so considering movies >= 3)

Subcategory: T4.2 Grouped Node Degree Statistics by Label

This query computes the average rating and movie count for each director with at least 3 movies, sorting them to find the top 10 highest-rated directors.

python 
q16_g = GraphFrame(vertices_df, edges_df)

q16_result = (
    q16_g.find('(d)-[e]->(m)')
    .filter(
        (F.col('d.type') == 'director') &
        (F.col('e.etype') == 'directed') &
        (F.col('m.type') == 'movie') &
        F.col('m.avgRating').isNotNull()
    )
    .groupBy(F.col('d.name').alias('Director'))
    .agg(
        F.round(F.avg('m.avgRating'), 2).alias('AvgRating'),
        F.count('m.id').alias('Movies'),
    )
    .filter(F.col('Movies') >= 3)
    .orderBy(F.desc('AvgRating'))
)

q16_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (d:mlgenome_director)-[:mlgenome_directed]->(m:mlgenome_movie)
WHERE m.avgRating IS NOT NULL
WITH d.name AS Director, avg(m.avgRating) AS AvgRating, count(m) AS Movies
WHERE Movies >= 3
RETURN Director, round(AvgRating, 2) AS AvgRating, Movies
ORDER BY AvgRating DESC
LIMIT 10

Q17. Actors by Number of Movies Acted In

Subcategory: T4.2 Grouped Node Degree Statistics by Label

This query calculates the total number of movies each actor has starred in and returns the top 10 most prolific actors.

python 
q17_g = GraphFrame(vertices_df, edges_df)

q17_result = (
    q17_g.find('(a)-[e]->(m)')
    .filter(
        (F.col('a.type') == 'actor') &
        (F.col('e.etype') == 'acted_in') &
        (F.col('m.type') == 'movie')
    )
    .groupBy(F.col('a.name').alias('Actor'))
    .agg(F.count('m.id').alias('TotalMovies'))
    .orderBy(F.desc('TotalMovies'))
)

q17_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (a:mlgenome_actor)-[:mlgenome_acted_in]->(m:mlgenome_movie)
RETURN a.name AS Actor, count(m) AS TotalMovies
ORDER BY TotalMovies DESC
LIMIT 10

Q18. Rating Edge Summary

Subcategory: T4.3 Ungrouped Edge Count and Property Statistics (Optional)

This query summarizes all RATED edges by count and rating-value statistics.

python 
q18_g = GraphFrame(vertices_df, edges_df)

q18_result = (
    q18_g.find('(u)-[r]->(m)')
    .filter(
        (F.col('u.type') == 'user') &
        (F.col('r.etype') == 'rated') &
        (F.col('m.type') == 'movie')
    )
    .agg(
        F.count(F.lit(1)).alias('RatingCount'),
        F.round(F.avg('r.rating'), 2).alias('AvgRating'),
        F.min('r.rating').alias('MinRating'),
        F.max('r.rating').alias('MaxRating'),
    )
)

q18_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (:mlgenome_user)-[r:mlgenome_rated]->(:mlgenome_movie)
RETURN count(r) AS RatingCount,
       round(avg(r.rating), 2) AS AvgRating,
       min(r.rating) AS MinRating,
       max(r.rating) AS MaxRating

Q19. Users Who Rate Harshly (Lowest Average Rating)

Subcategory: T4.4 Grouped Edge Count and Property Statistics

This query identifies the top 10 users with the lowest average ratings among those who have submitted at least 5 ratings, highlighting the harshest critics.

python 
q19_g = GraphFrame(vertices_df, edges_df)

q19_result = (
    q19_g.find('(u)-[r]->(m)')
    .filter(
        (F.col('u.type') == 'user') &
        (F.col('r.etype') == 'rated') &
        (F.col('m.type') == 'movie')
    )
    .filter(F.col('r.rating').isNotNull())
    .groupBy(F.col('u.user_id').alias('UserID'))
    .agg(
        F.count(F.lit(1)).alias('TotalRatings'),
        F.round(F.avg('r.rating'), 2).alias('AvgRating'),
    )
    .filter(F.col('TotalRatings') >= 5)
    .select('UserID', 'AvgRating', 'TotalRatings')
    .orderBy('AvgRating')
)

q19_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (u:mlgenome_user)-[r:mlgenome_rated]->(m:mlgenome_movie)
WHERE r.rating IS NOT NULL
WITH u.user_id AS UserID, count(r) AS TotalRatings, avg(r.rating) AS AvgRating
WHERE TotalRatings >= 5
RETURN UserID, round(AvgRating, 2) AS AvgRating, TotalRatings
ORDER BY AvgRating ASC
LIMIT 10

Q20. Bounded Rating Path Counts by Length

Subcategory: T4.5 Path Counts by Length

This query counts bounded RATED paths from a selected user and groups the result only by path length.

python 
q20_g = GraphFrame(vertices_df, edges_df)

q20_h1 = (
    q20_g.find('(u)-[e1]-(n)')
    .filter(
        (F.col('u.type') == 'user') &
        (F.col('u.user_id') == 997206) &
        (F.col('e1.etype') == 'rated')
    )
    .agg(
        F.count(F.lit(1)).alias('PathCount'),
        F.countDistinct('n.id').alias('DistinctTargets'),
    )
    .withColumn('PathLength', F.lit(1))
)

q20_h2 = (
    q20_g.find('(u)-[e1]-(v1); (v1)-[e2]-(n)')
    .filter(
        (F.col('u.type') == 'user') &
        (F.col('u.user_id') == 997206) &
        (F.col('e1.etype') == 'rated') &
        (F.col('e2.etype') == 'rated') &
        (F.col('e1._eid') != F.col('e2._eid'))
    )
    .agg(
        F.count(F.lit(1)).alias('PathCount'),
        F.countDistinct('n.id').alias('DistinctTargets'),
    )
    .withColumn('PathLength', F.lit(2))
)

q20_h3 = (
    q20_g.find('(u)-[e1]-(v1); (v1)-[e2]-(v2); (v2)-[e3]-(n)')
    .filter(
        (F.col('u.type') == 'user') &
        (F.col('u.user_id') == 997206) &
        (F.col('e1.etype') == 'rated') &
        (F.col('e2.etype') == 'rated') &
        (F.col('e3.etype') == 'rated') &
        (F.col('e1._eid') != F.col('e2._eid')) &
        (F.col('e1._eid') != F.col('e3._eid')) &
        (F.col('e2._eid') != F.col('e3._eid'))
    )
    .agg(
        F.count(F.lit(1)).alias('PathCount'),
        F.countDistinct('n.id').alias('DistinctTargets'),
    )
    .withColumn('PathLength', F.lit(3))
)

q20_result = (
    q20_h1.unionByName(q20_h2).unionByName(q20_h3)
    .select('PathLength', 'PathCount', 'DistinctTargets')
    .orderBy('PathLength')
)

q20_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH p = (:mlgenome_user {user_id: 1})-[:mlgenome_rated]-{1,3}(target)
RETURN length(p) AS PathLength,
       count(p) AS PathCount,
       count(DISTINCT target) AS DistinctTargets
ORDER BY PathLength ASC

Q21. Bounded Rating Path Counts by Length and Endpoint Type

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

This query extends path-count aggregation by grouping bounded paths by both path length and endpoint label.

python 
q21_g = GraphFrame(vertices_df, edges_df)

q21_h1 = (
    q21_g.find('(u)-[e1]-(n)')
    .filter(
        (F.col('u.type') == 'user') &
        (F.col('u.user_id') == 997206) &
        (F.col('e1.etype') == 'rated')
    )
    .groupBy(F.col('n.type').alias('EndpointType'))
    .agg(
        F.count(F.lit(1)).alias('PathCount'),
        F.countDistinct('n.id').alias('DistinctTargets'),
    )
    .withColumn('PathLength', F.lit(1))
)

q21_h2 = (
    q21_g.find('(u)-[e1]-(v1); (v1)-[e2]-(n)')
    .filter(
        (F.col('u.type') == 'user') &
        (F.col('u.user_id') == 997206) &
        (F.col('e1.etype') == 'rated') &
        (F.col('e2.etype') == 'rated') &
        (F.col('e1._eid') != F.col('e2._eid'))
    )
    .groupBy(F.col('n.type').alias('EndpointType'))
    .agg(
        F.count(F.lit(1)).alias('PathCount'),
        F.countDistinct('n.id').alias('DistinctTargets'),
    )
    .withColumn('PathLength', F.lit(2))
)

q21_h3 = (
    q21_g.find('(u)-[e1]-(v1); (v1)-[e2]-(v2); (v2)-[e3]-(n)')
    .filter(
        (F.col('u.type') == 'user') &
        (F.col('u.user_id') == 997206) &
        (F.col('e1.etype') == 'rated') &
        (F.col('e2.etype') == 'rated') &
        (F.col('e3.etype') == 'rated') &
        (F.col('e1._eid') != F.col('e2._eid')) &
        (F.col('e1._eid') != F.col('e3._eid')) &
        (F.col('e2._eid') != F.col('e3._eid'))
    )
    .groupBy(F.col('n.type').alias('EndpointType'))
    .agg(
        F.count(F.lit(1)).alias('PathCount'),
        F.countDistinct('n.id').alias('DistinctTargets'),
    )
    .withColumn('PathLength', F.lit(3))
)

q21_result = (
    q21_h1.unionByName(q21_h2).unionByName(q21_h3)
    .select('PathLength', 'EndpointType', 'PathCount', 'DistinctTargets')
    .orderBy('PathLength', F.desc('PathCount'))
)

q21_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH p = (:mlgenome_user {user_id: 1})-[:mlgenome_rated]-{1,3}(target)
WITH length(p) AS PathLength, labels(target)[0] AS EndpointType, target, p
RETURN PathLength,
       EndpointType,
       count(p) AS PathCount,
       count(DISTINCT target) AS DistinctTargets
ORDER BY PathLength ASC, PathCount DESC

Q22. Directors with the Most Movies Above 4.0 Rating (Not just average, how many great movies have they made? Considered minimum of 2 movies)

Subcategory: T5.1 Degree Centrality

This query finds the top 10 directors with the most movies rated 4.0 or above, considering only directors who have at least 2 such highly-rated movies.

python 
q22_g = GraphFrame(vertices_df, edges_df)

q22_result = (
    q22_g.find('(d)-[e]->(m)')
    .filter(
        (F.col('d.type') == 'director') &
        (F.col('e.etype') == 'directed') &
        (F.col('m.type') == 'movie') &
        (F.col('m.avgRating') >= 4.0)
    )
    .groupBy(F.col('d.name').alias('Director'))
    .agg(F.count('m.id').alias('HighRatedMovies'))
    .filter(F.col('HighRatedMovies') >= 2)
    .orderBy(F.desc('HighRatedMovies'))
)

q22_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (d:mlgenome_director)-[:mlgenome_directed]->(m:mlgenome_movie)
WHERE m.avgRating >= 4.0
WITH d.name AS Director, count(m) AS HighRatedMovies
WHERE HighRatedMovies >= 2
RETURN Director, HighRatedMovies
ORDER BY HighRatedMovies DESC
LIMIT 10

Q23. Top Tags by Frequency in Surveys (Which tags are evaluated most often by users?)

Subcategory: T5.1 Degree Centrality

This query counts the total number of survey evaluations for each tag and returns the top 10 most frequently evaluated tags.

python 
q23_g = GraphFrame(vertices_df, edges_df)

q23_result = (
    q23_g.find('(s)-[e]->(t)')
    .filter(
        (F.col('s.type') == 'survey') &
        (F.col('e.etype') == 'about_tag') &
        (F.col('t.type') == 'tag')
    )
    .groupBy(F.col('t.tag').alias('Tag'))
    .agg(F.count(F.lit(1)).alias('SurveyCount'))
    .orderBy(F.desc('SurveyCount'))
)

q23_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (:mlgenome_survey)-[:mlgenome_about_tag]->(t:mlgenome_tag)
RETURN t.tag AS Tag, count(*) AS SurveyCount
ORDER BY SurveyCount DESC
LIMIT 10

Q24. Tags Most Frequently Scored Highly in Surveys (Which tags do users consistently agree are applicable to movies (score ≥ 4)?)

Subcategory: T5.1 Degree Centrality

This query identifies tags that users consistently score highly (survey score >= 4) and returns the top 10 highest-scoring tags.

python 
q24_g = GraphFrame(vertices_df, edges_df)

q24_result = (
    q24_g.find('(s)-[e]->(t)')
    .filter(
        (F.col('s.type') == 'survey') &
        (F.col('s.score') >= 4) &
        (F.col('e.etype') == 'about_tag') &
        (F.col('t.type') == 'tag')
    )
    .groupBy(F.col('t.tag').alias('Tag'))
    .agg(F.count(F.lit(1)).alias('HighScoreCount'))
    .orderBy(F.desc('HighScoreCount'))
)

q24_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (s:mlgenome_survey)-[:mlgenome_about_tag]->(t:mlgenome_tag)
WHERE s.score >= 4
RETURN t.tag AS Tag, count(*) AS HighScoreCount
ORDER BY HighScoreCount DESC
LIMIT 10

Q25. Movies With Most Tagged Survey Feedback (Which movies have been analyzed most often in surveys?)

Subcategory: T5.1 Degree Centrality

This query finds the top 10 movies that have been evaluated the most number of times in tag genome surveys.

python 
q25_g = GraphFrame(vertices_df, edges_df)

q25_result = (
    q25_g.find('(s)-[e]->(m)')
    .filter(
        (F.col('s.type') == 'survey') &
        (F.col('e.etype') == 'about_movie') &
        (F.col('m.type') == 'movie')
    )
    .groupBy(F.col('m.title').alias('Movie'))
    .agg(F.count(F.lit(1)).alias('SurveyMentions'))
    .orderBy(F.desc('SurveyMentions'))
)

q25_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (s:mlgenome_survey)-[:mlgenome_about_movie]->(m:mlgenome_movie)
RETURN m.title AS Movie, count(*) AS SurveyMentions
ORDER BY SurveyMentions DESC
LIMIT 10

Q26. Users Who Participated in the Most Surveys

Subcategory: T5.1 Degree Centrality

This query lists the top 10 users who have participated in the highest number of tag genome surveys.

python 
q26_g = GraphFrame(vertices_df, edges_df)

q26_result = (
    q26_g.find('(u)-[e]->(s)')
    .filter(
        (F.col('u.type') == 'user') &
        (F.col('e.etype') == 'answered') &
        (F.col('s.type') == 'survey')
    )
    .groupBy(F.col('u.user_id').alias('User'))
    .agg(F.count('s.id').alias('TotalSurveys'))
    .orderBy(F.desc('TotalSurveys'))
)

q26_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (u:mlgenome_user)-[:mlgenome_answered]->(s:mlgenome_survey)
RETURN u.user_id AS User, count(s) AS TotalSurveys
ORDER BY TotalSurveys DESC
LIMIT 10

Q27. PageRank on the MovieLens Genome Rating Graph

Subcategory: T5.2 PageRank

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

python 
q27_g = GraphFrame(vertices_df, edges_df)

q27_rating_graph = GraphFrame(
    q27_g.vertices.filter(F.col('type').isin('user', 'movie')),
    q27_g.edges.filter(F.col('etype') == 'rated'),
).as_undirected()

q27_pagerank = q27_rating_graph.pageRank(
    resetProbability=0.15,
    maxIter=20,
)

q27_result = (
    q27_pagerank.vertices
    .select(
        F.array(F.col('type')).alias('Labels'),
        F.coalesce(
            F.col('user_id').cast('string'),
            F.col('title'),
            F.col('id'),
        ).alias('Node'),
        F.col('pagerank').alias('Score'),
    )
    .orderBy(F.desc('Score'))
)

q27_result.show(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 the Rating Graph

Subcategory: T5.3 Connected Components / Weakly Connected Components

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

python 
q28_g = GraphFrame(vertices_df, edges_df)

q28_rating_graph = GraphFrame(
    q28_g.vertices.filter(F.col('type').isin('user', 'movie')),
    q28_g.edges.filter(F.col('etype') == 'rated'),
).as_undirected()

q28_components = q28_rating_graph.connectedComponents(
    algorithm='graphframes',
    broadcastThreshold=-1,
    use_local_checkpoints=True,
)

q28_result = (
    q28_components
    .groupBy(F.col('component').alias('componentId'))
    .agg(F.count(F.lit(1)).alias('ComponentSize'))
    .orderBy(F.desc('ComponentSize'))
)

q28_result.show(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 Rating Edges

Subcategory: T5.4 Strongly Connected Components

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

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 and returns the largest communities.

python 
q30_g = GraphFrame(vertices_df, edges_df)

q30_rating_graph = GraphFrame(
    q30_g.vertices.filter(F.col('type').isin('user', 'movie')),
    q30_g.edges.filter(F.col('etype') == 'rated'),
).as_undirected()

q30_communities = q30_rating_graph.labelPropagation(
    maxIter=10,
    algorithm='graphx',
)

q30_result = (
    q30_communities
    .groupBy(F.col('label').alias('communityId'))
    .agg(F.count(F.lit(1)).alias('CommunitySize'))
    .orderBy(F.desc('CommunitySize'))
)

q30_result.show(truncate=False)
q30_communities.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 community detection (Louvain).

Q31. Best Tags per Movie by Survey and Tag_Count

Subcategory: T3.1 Triangle Detection; T4.2 Grouped Node Degree Statistics by Label; T4.4 Grouped Edge Count and Property Statistics

This query correlates tags applied to movies via direct tag counts with survey evaluations to find movie-tag pairings with the highest average consensus score.

python 
q31_g = GraphFrame(vertices_df, edges_df)

q31_result = (
    q31_g.find('(s)-[am]->(m); (s)-[at]->(t); (m)-[ht]->(t)')
    .filter(
        (F.col('s.type') == 'survey') &
        (F.col('am.etype') == 'about_movie') &
        (F.col('m.type') == 'movie') &
        (F.col('at.etype') == 'about_tag') &
        (F.col('t.type') == 'tag') &
        (F.col('ht.etype') == 'has_tag')
    )
    .groupBy(
        F.col('m.title').alias('Movie'),
        F.col('t.tag').alias('Tag'),
    )
    .agg(
        F.sum('ht.count').alias('TagApplications'),
        F.avg('s.score').alias('AvgSurveyScore'),
        F.count('s.id').alias('SurveyMentions'),
    )
    .filter(
        (F.col('TagApplications') > 5) &
        (F.col('SurveyMentions') >= 5)
    )
    .select(
        'Movie',
        'Tag',
        'TagApplications',
        F.round('AvgSurveyScore', 2).alias('AvgSurveyScore'),
        'SurveyMentions',
    )
    .orderBy(F.desc('AvgSurveyScore'), F.desc('SurveyMentions'))
)

q31_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (m:mlgenome_movie)-[ht:mlgenome_has_tag]->(t:mlgenome_tag)
MATCH (s:mlgenome_survey)-[:mlgenome_about_movie]->(m)
MATCH (s)-[:mlgenome_about_tag]->(t)
WITH m.title AS Movie, t.tag AS Tag,
    sum(ht.count) AS TagApplications,
    avg(s.score) AS AvgSurveyScore,
    count(s) AS SurveyMentions
WHERE TagApplications > 5 AND SurveyMentions >= 5
RETURN Movie, Tag, TagApplications, round(AvgSurveyScore, 2) AS AvgSurveyScore, SurveyMentions
ORDER BY AvgSurveyScore DESC, SurveyMentions DESC
LIMIT 20

Q32. Top Actor-Director Pairs by Number of Collaborations

Subcategory: T3.5 Two-hop Join Pattern; T4.2 Grouped Node Degree Statistics by Label

This query finds the top 10 actor-director pairs with the highest number of movie collaborations.

python 
q32_g = GraphFrame(vertices_df, edges_df)

q32_result = (
    q32_g.find('(d)-[de]->(m); (a)-[ae]->(m)')
    .filter(
        (F.col('d.type') == 'director') &
        (F.col('de.etype') == 'directed') &
        (F.col('m.type') == 'movie') &
        (F.col('a.type') == 'actor') &
        (F.col('ae.etype') == 'acted_in')
    )
    .groupBy(
        F.col('d.name').alias('Director'),
        F.col('a.name').alias('Actor'),
    )
    .agg(F.count('m.id').alias('Collaborations'))
    .orderBy(F.desc('Collaborations'))
)

q32_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (d:mlgenome_director)-[:mlgenome_directed]->(m:mlgenome_movie)<-[:mlgenome_acted_in]-(a:mlgenome_actor)
RETURN d.name AS Director, a.name AS Actor, count(m) AS Collaborations
ORDER BY Collaborations DESC
LIMIT 10

Q33. Directors Whose Movies Collect the Most Reviews

Subcategory: T3.5 Two-hop Join Pattern; T4.2 Grouped Node Degree Statistics by Label

This query finds the top 10 directors whose movies have collectively accumulated the highest number of text reviews.

python 
q33_g = GraphFrame(vertices_df, edges_df)

q33_result = (
    q33_g.find('(d)-[de]->(m); (m)-[hr]->(review)')
    .filter(
        (F.col('d.type') == 'director') &
        (F.col('de.etype') == 'directed') &
        (F.col('m.type') == 'movie') &
        (F.col('hr.etype') == 'has_review') &
        (F.col('review.type') == 'review')
    )
    .groupBy(F.col('d.name').alias('Director'))
    .agg(F.count('review.id').alias('TotalReviews'))
    .orderBy(F.desc('TotalReviews'))
)

q33_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (d:mlgenome_director)-[:mlgenome_directed]->(m:mlgenome_movie)-[:mlgenome_has_review]->(r:mlgenome_review)
RETURN d.name AS Director, count(r) AS TotalReviews
ORDER BY TotalReviews DESC
LIMIT 10

Q34. Actor-Director Collaboration Quality (Which pairs consistently produce great movies?)

Subcategory: T4.2 Grouped Node Degree Statistics by Label; T4.7 Union and Intersection

This query evaluates the quality of actor-director collaborations by computing the average rating of their shared movies, returning the top 10 pairs with at least 3 collaborations.

python 
q34_g = GraphFrame(vertices_df, edges_df)

q34_result = (
    q34_g.find('(d)-[de]->(m); (a)-[ae]->(m)')
    .filter(
        (F.col('d.type') == 'director') &
        (F.col('de.etype') == 'directed') &
        (F.col('m.type') == 'movie') &
        F.col('m.avgRating').isNotNull() &
        (F.col('a.type') == 'actor') &
        (F.col('ae.etype') == 'acted_in')
    )
    .groupBy(
        F.col('d.name').alias('Director'),
        F.col('a.name').alias('Actor'),
    )
    .agg(
        F.count('m.id').alias('Collabs'),
        F.round(F.avg('m.avgRating'), 2).alias('AvgRating'),
    )
    .filter(F.col('Collabs') >= 3)
    .orderBy(F.desc('AvgRating'))
)

q34_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (d:mlgenome_director)-[:mlgenome_directed]->(m:mlgenome_movie)<-[:mlgenome_acted_in]-(a:mlgenome_actor)
WHERE m.avgRating IS NOT NULL
WITH d.name AS Director, a.name AS Actor, count(m) AS Collabs, avg(m.avgRating) AS AvgRating
WHERE Collabs >= 3
RETURN Director, Actor, Collabs, round(AvgRating, 2) AS AvgRating
ORDER BY AvgRating DESC
LIMIT 10

Q35. Tag-Movie Associations with High Confidence (Which movie-tag pairs were most confidently agreed upon by users?)

Subcategory: T4.2 Grouped Node Degree Statistics by Label; T4.7 Union and Intersection

This query finds movie-tag pairs that are strongly agreed upon by users, based on the frequency of high survey scores (score >= 4).

python 
q35_g = GraphFrame(vertices_df, edges_df)

q35_result = (
    q35_g.find('(s)-[am]->(m); (s)-[at]->(t)')
    .filter(
        (F.col('s.type') == 'survey') &
        (F.col('s.score') >= 4) &
        (F.col('am.etype') == 'about_movie') &
        (F.col('m.type') == 'movie') &
        (F.col('at.etype') == 'about_tag') &
        (F.col('t.type') == 'tag')
    )
    .groupBy(
        F.col('m.title').alias('Movie'),
        F.col('t.tag').alias('Tag'),
    )
    .agg(F.count(F.lit(1)).alias('HighScoreCount'))
    .orderBy(F.desc('HighScoreCount'))
)

q35_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (s:mlgenome_survey)-[:mlgenome_about_movie]->(m:mlgenome_movie), (s)-[:mlgenome_about_tag]->(t:mlgenome_tag)
WHERE s.score >= 4
RETURN m.title AS Movie, t.tag AS Tag, count(*) AS HighScoreCount
ORDER BY HighScoreCount DESC
LIMIT 10

Q36. Actors Frequently Appearing in High-Rated Movies with Different Directors (Highlights high-performing, collaborative actors trusted by many top directors.)

Subcategory: Other

This query finds actors who frequently appear in highly-rated movies (rating >= 4.0) with different directors, showcasing versatile and collaborative actors.

python 
q36_g = GraphFrame(vertices_df, edges_df)

q36_result = (
    q36_g.find('(a)-[ae]->(m); (d)-[de]->(m)')
    .filter(
        (F.col('a.type') == 'actor') &
        (F.col('ae.etype') == 'acted_in') &
        (F.col('m.type') == 'movie') &
        (F.col('m.avgRating') >= 4.0) &
        (F.col('d.type') == 'director') &
        (F.col('de.etype') == 'directed')
    )
    .groupBy(F.col('a.name').alias('Actor'))
    .agg(
        F.countDistinct('d.name').alias('UniqueTopDirectors'),
        F.countDistinct('m.id').alias('TopRatedMovies'),
    )
    .filter(F.col('TopRatedMovies') >= 3)
    .orderBy(F.desc('UniqueTopDirectors'), F.desc('TopRatedMovies'))
)

q36_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (a:mlgenome_actor)-[:mlgenome_acted_in]->(m:mlgenome_movie)<-[:mlgenome_directed]-(d:mlgenome_director)
WHERE m.avgRating >= 4.0
WITH a.name AS Actor, count(DISTINCT d.name) AS UniqueTopDirectors, count(DISTINCT m) AS TopRatedMovies
WHERE TopRatedMovies >= 3
RETURN Actor, UniqueTopDirectors, TopRatedMovies
ORDER BY UniqueTopDirectors DESC, TopRatedMovies DESC
LIMIT 10

Q37. Most Frequently Used Words in Reviews

Subcategory: Other

This query tokenizes text reviews to extract the most frequently used words, filtering out common stop words.

python 
q37_g = GraphFrame(vertices_df, edges_df)

q37_stop_words = [
    'this', 'that', 'with', 'from', 'have', 'your', 'just', 'what', 'when',
    'they', 'were', 'then', 'there', 'would', 'which', 'could', 'really',
]

q37_result = (
    q37_g.vertices
    .filter((F.col('type') == 'review') & F.col('text').isNotNull())
    .select(F.explode(F.split(F.lower('text'), r'\W+')).alias('word'))
    .filter((F.length('word') > 3) & ~F.col('word').isin(q37_stop_words))
    .groupBy('word')
    .agg(F.count(F.lit(1)).alias('freq'))
    .orderBy(F.desc('freq'))
    .limit(20)
)

q37_result.show(20, truncate=False)
Note: Cannot be expressed in standard GQL because standard GQL defines no string tokenization function (split) and no list-unnesting clause (UNWIND) with which to expand the words of a review into rows.

Q38. Movies Most Frequently Called "Boring", "Disgusting", etc.

Subcategory: Other

This query counts reviews containing negative keywords (e.g., 'boring', 'disgusting', 'waste') for each movie to identify those with the most negative feedback.

python 
q38_g = GraphFrame(vertices_df, edges_df)

q38_result = (
    q38_g.find('(m)-[hr]->(r)')
    .filter(
        (F.col('m.type') == 'movie') &
        (F.col('hr.etype') == 'has_review') &
        (F.col('r.type') == 'review') &
        (
            F.lower(F.col('r.text')).contains('boring') |
            F.lower(F.col('r.text')).contains('disgusting') |
            F.lower(F.col('r.text')).contains('waste') |
            F.lower(F.col('r.text')).contains('terrible') |
            F.lower(F.col('r.text')).contains('bad')
        )
    )
    .groupBy(F.col('m.title').alias('Movie'))
    .agg(F.count(F.lit(1)).alias('NegativeMentions'))
    .orderBy(F.desc('NegativeMentions'))
)

q38_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (m:mlgenome_movie)-[:mlgenome_has_review]->(r:mlgenome_review)
WHERE toLower(r.text) CONTAINS 'boring'
OR toLower(r.text) CONTAINS 'disgusting'
OR toLower(r.text) CONTAINS 'waste'
OR toLower(r.text) CONTAINS 'terrible'
OR toLower(r.text) CONTAINS 'bad'
WITH m.title AS Movie, count(*) AS NegativeMentions
RETURN Movie, NegativeMentions
ORDER BY NegativeMentions DESC
LIMIT 10

Q39. Users With Highest Survey Agreement Accuracy (Measures users whose survey scores closely align with consensus (like those who gave 5 when others did too))

Subcategory: Other

This query ranks users by how closely their survey ratings align with the consensus ratings for each tag, highlighting the most consistent/accurate survey responders.

python 
q39_g = GraphFrame(vertices_df, edges_df)

q39_consensus = (
    q39_g.find('(u)-[ans]->(s); (s)-[at1]->(t); (s2)-[at2]->(t)')
    .filter(
        (F.col('u.type') == 'user') &
        (F.col('ans.etype') == 'answered') &
        (F.col('s.type') == 'survey') &
        (F.col('at1.etype') == 'about_tag') &
        (F.col('t.type') == 'tag') &
        (F.col('s2.type') == 'survey') &
        (F.col('at2.etype') == 'about_tag') &
        F.col('s2.score').isNotNull()
    )
    .groupBy(
        F.col('u.user_id').alias('User'),
        F.col('t.tag').alias('Tag'),
        F.col('s.score').alias('UserScore'),
    )
    .agg(F.avg('s2.score').alias('Consensus'))
)

q39_result = (
    q39_consensus
    .groupBy('User')
    .agg(
        F.avg(F.abs(F.col('UserScore') - F.col('Consensus'))).alias('AvgDeviation'),
        F.count(F.lit(1)).alias('Count'),
    )
    .filter(F.col('Count') >= 10)
    .select(
        'User',
        F.round('AvgDeviation', 2).alias('AvgTagDisagreement'),
        'Count',
    )
    .orderBy('AvgTagDisagreement')
)

q39_result.show(truncate=False)

Equivalent query in GQL standard:

gql 
MATCH (u:mlgenome_user)-[:mlgenome_answered]->(s:mlgenome_survey)-[:mlgenome_about_tag]->(t:mlgenome_tag)
WITH u.user_id AS User, t.tag AS Tag, s.score AS UserScore
MATCH (s2:mlgenome_survey)-[:mlgenome_about_tag]->(t)
WHERE s2.score IS NOT NULL
WITH User, Tag, UserScore, avg(s2.score) AS Consensus
WITH User, avg(abs(UserScore - Consensus)) AS AvgDeviation, count(*) AS Count
WHERE Count >= 10
RETURN User, round(AvgDeviation, 2) AS AvgTagDisagreement, Count
ORDER BY AvgTagDisagreement ASC
LIMIT 10