← Back to Amazon Dataset

GraphFrames Implementation

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

Schema Diagram and Graph Statistics

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

Amazon Neo4j Schema

Node Statistics

Node Label Count
Product 23,643,222
Category 2,487,388
Review 230,139,802
Image 17,081,240
ReviewImage 6,538,802
Reviewer 43,249,276

Edge Statistics

Relationship Count
(:Review)-[:REVIEWS]->(:Product) 230,139,802
(:Product)-[:HAS_PRODUCT_IMAGE]->(:Image) 22,531,444
(:Product)-[:ALSO_BUY]->(:Product) 97,123,542
(:Reviewer)-[:WROTE]->(:Review) 230,139,802
(:Review)-[:HAS_REVIEW_IMAGE]->(:ReviewImage) 6,840,462
(:Product)-[:IN_CATEGORY]->(:Category) 62,832,310
(:Product)-[:ALSO_VIEW]->(:Product) 92,658,319

Every vertex and edge count matches the Neo4j load exactly. One category name differs: Neo4j holds a Category whose name is the empty string, which here, as in the TigerGraph load, is named <null>.

Data Loading

Overview

Data import is handled in three parts. First we read the input files into DataFrames. Then we define nodes and edges, applying the same MERGE semantics as the Neo4j loader: every node and edge is deduplicated on its key, and a Product that is only referenced by an edge or a review exists with null attributes. Finally, we save every node and edge type as a Delta table in the Hive metastore (database amazon, shared with PuppyGraph) and create the graph from those tables.

The input is the same set of tab-separated files the TigerGraph load reads: eleven files per Amazon category folder (product.tsv, review.tsv, also_buy.tsv, ...), each with a header row and no quoting, with tab, newline, carriage return, backslash and other control characters inside a field written as escape sequences. Every node id is prefixed with its type (product:B00006RVQO) so that ids are unique across the graph, and the Neo4j key property (asin, name, url, reviewerID) is kept as its own column.

DataFrames

We defined a few helper functions to read the files, decode the escape sequences, turn empty numeric fields into nulls, and keep one row per key.

python 
TSV_ROOT = Path('/datasets/prop_datasets/amazon/tg_tsv2')
CATEGORIES = sorted(p.name for p in TSV_ROOT.iterdir() if p.is_dir())

def read_tsv(name, columns):
    # One file per category folder, header row, tab-separated, no quoting.
    paths = [(TSV_ROOT / c / f'{name}.tsv').as_uri() for c in CATEGORIES]
    schema = StructType([StructField(c, StringType()) for c in columns])
    df = (spark.read.format('csv').schema(schema)
          .option('sep', '\t').option('header', 'true')
          .option('quote', '').option('escape', '\u0000')
          .option('ignoreLeadingWhiteSpace', 'false').option('ignoreTrailingWhiteSpace', 'false')
          .option('mode', 'FAILFAST')
          .load(paths))
    # Spark reads an unquoted empty field as null; the files hold no nulls, so restore "".
    return df.select(*[F.coalesce(F.col(c), F.lit('')).alias(c) for c in columns])

# Decode the \\ \t \n \r \xNN escapes back to the raw characters Neo4j holds.
_CTRL = [i for i in list(range(0x20)) + [0x7F] if i not in (0x09, 0x0A, 0x0D)]

def _decode_segment(s):
    for escape, char in (('\\t', '\t'), ('\\n', '\n'), ('\\r', '\r')):
        s = F.replace(s, F.lit(escape), F.lit(char))
    for i in _CTRL:
        s = F.replace(s, F.lit('\\x%02x' % i), F.lit(chr(i)))
    return s

def decoded(column):
    c = F.col(column)
    full = F.array_join(F.transform(F.split(c, r'\\\\'), _decode_segment), '\\')
    return F.when(c.contains('\\'), full).otherwise(c).alias(column)

def decode_columns(df, columns):
    return df.select(*[decoded(c) if c in columns else F.col(c) for c in df.columns])

def empty_to_null(column, spark_type):
    # "" in a numeric/datetime column is Cypher's null (toFloat/toInteger failed).
    c = F.col(column)
    return F.when(c == '', F.lit(None)).otherwise(c).cast(spark_type).alias(column)

def vid(family, column):
    return F.concat(F.lit(f'{family}:'), F.col(column))

def greatest_per_key(df, key, value_columns):
    # MERGE + SET keeps one row per key; the greatest row wins, deterministically.
    return (df.groupBy(key)
            .agg(F.max(F.struct(*value_columns)).alias('_v'))
            .select(key, *[F.col(f'_v.{c}').alias(c) for c in value_columns]))

Define Nodes

First we define the Review node. review_id is a key added for the TigerGraph load (Neo4j's Review nodes have none); it is unique per review row.

python 
review_raw = read_tsv('review', ['review_id', 'reviewerID', 'asin', 'unixReviewTime',
                                 'reviewDatetime', 'summary', 'reviewText', 'verified',
                                 'vote', 'overall'])
review_raw = decode_columns(review_raw, ['review_id', 'reviewerID', 'asin', 'summary', 'reviewText'])
review_vertices_df = review_raw.select(
    vid('review', 'review_id').alias('id'), F.lit('review').alias('type'),
    'review_id', 'reviewerID', 'asin',
    empty_to_null('unixReviewTime', 'bigint'),
    F.to_timestamp(F.when(F.col('reviewDatetime') == '', None)
                   .otherwise(F.col('reviewDatetime')), 'yyyy-MM-dd HH:mm:ss').alias('reviewDatetime'),
    'summary', 'reviewText',
    F.when(F.col('verified') == 'true', True)
     .when(F.col('verified') == 'false', False).alias('verified'),
    empty_to_null('vote', 'int'),
    empty_to_null('overall', 'int'))
review_vertices_df.printSchema(); review_vertices_df.show(3, truncate=80)

Then we define the Reviewer node.

python 
reviewer_raw = decode_columns(read_tsv('reviewer', ['reviewerID', 'name']), ['reviewerID', 'name'])
reviewer_vertices_df = greatest_per_key(reviewer_raw, 'reviewerID', ['name']).select(
    vid('reviewer', 'reviewerID').alias('id'), F.lit('reviewer').alias('type'),
    'reviewerID', 'name')
reviewer_vertices_df.printSchema(); reviewer_vertices_df.show(3, truncate=80)

Then we define the ReviewImage node.

python 
reviewimage_raw = decode_columns(read_tsv('review_image', ['url']), ['url'])
reviewimage_vertices_df = reviewimage_raw.distinct().select(
    vid('reviewimage', 'url').alias('id'), F.lit('reviewimage').alias('type'), 'url')
reviewimage_vertices_df.printSchema(); reviewimage_vertices_df.show(3, truncate=80)

Then we define the Category node.

python 
category_raw = decode_columns(read_tsv('category', ['name']), ['name'])
category_vertices_df = category_raw.distinct().select(
    vid('category', 'name').alias('id'), F.lit('category').alias('type'), 'name')
category_vertices_df.printSchema(); category_vertices_df.show(3, truncate=80)

Then we define the Image node.

python 
image_raw = decode_columns(read_tsv('image', ['url', 'highRes']), ['url', 'highRes'])
image_vertices_df = greatest_per_key(image_raw, 'url', ['highRes']).select(
    vid('image', 'url').alias('id'), F.lit('image').alias('type'), 'url', 'highRes')
image_vertices_df.printSchema(); image_vertices_df.show(3, truncate=80)

Then we define the Product node, from the product metadata plus every ASIN referenced by a co-buy, co-view or review.

python 
text = ['asin', 'title', 'brand', 'description']
product_raw = decode_columns(read_tsv('product', text + ['price_min', 'price_max']), text)
product_meta = greatest_per_key(
    product_raw.select(*text, empty_to_null('price_min', 'double'),
                       empty_to_null('price_max', 'double')),
    'asin', ['title', 'brand', 'description', 'price_min', 'price_max'])

# Products that only exist because an ALSO_BUY / ALSO_VIEW edge or a review points at
# them: Neo4j's MERGE (p:Product {asin}) creates them with no other property.
referenced = (
    decode_columns(read_tsv('also_buy', ['asin', 'related_asin']), ['related_asin'])
        .select(F.col('related_asin').alias('asin'))
    .unionByName(decode_columns(read_tsv('also_view', ['asin', 'view_asin']), ['view_asin'])
        .select(F.col('view_asin').alias('asin')))
    .unionByName(review_vertices_df.select('asin'))
    .distinct())
product_vertices_df = (referenced.join(product_meta, 'asin', 'full_outer')
    .select(vid('product', 'asin').alias('id'), F.lit('product').alias('type'),
            'asin', 'title', 'brand', 'description', 'price_min', 'price_max'))
product_vertices_df.printSchema(); product_vertices_df.show(3, truncate=80)

Define Edges

Next step is to define edges. First we define the REVIEWS and WROTE edges, which come from columns of the review rows.

python 
reviews_edges_df = review_vertices_df.select(
    F.col('id').alias('src'), vid('product', 'asin').alias('dst'), F.lit('reviews').alias('etype'))
reviews_edges_df.printSchema(); reviews_edges_df.show(3, truncate=80)
python 
wrote_edges_df = review_vertices_df.select(
    vid('reviewer', 'reviewerID').alias('src'), F.col('id').alias('dst'), F.lit('wrote').alias('etype'))
wrote_edges_df.printSchema(); wrote_edges_df.show(3, truncate=80)

Then we define the HAS_REVIEW_IMAGE edge.

python 
has_review_image_raw = decode_columns(read_tsv('review_has_image', ['review_id', 'url']),
                                      ['review_id', 'url'])
has_review_image_edges_df = has_review_image_raw.distinct().select(
    vid('review', 'review_id').alias('src'), vid('reviewimage', 'url').alias('dst'),
    F.lit('has_review_image').alias('etype'))
has_review_image_edges_df.printSchema(); has_review_image_edges_df.show(3, truncate=80)

Then we define the ALSO_BUY edge.

python 
also_buy_raw = decode_columns(read_tsv('also_buy', ['asin', 'related_asin']), ['asin', 'related_asin'])
also_buy_edges_df = also_buy_raw.distinct().select(
    vid('product', 'asin').alias('src'), vid('product', 'related_asin').alias('dst'),
    F.lit('also_buy').alias('etype'))
also_buy_edges_df.printSchema(); also_buy_edges_df.show(3, truncate=80)

Then we define the ALSO_VIEW edge.

python 
also_view_raw = decode_columns(read_tsv('also_view', ['asin', 'view_asin']), ['asin', 'view_asin'])
also_view_edges_df = also_view_raw.distinct().select(
    vid('product', 'asin').alias('src'), vid('product', 'view_asin').alias('dst'),
    F.lit('also_view').alias('etype'))
also_view_edges_df.printSchema(); also_view_edges_df.show(3, truncate=80)

Then we define the IN_CATEGORY edge.

python 
in_category_raw = decode_columns(read_tsv('product_category', ['asin', 'name']), ['asin', 'name'])
in_category_edges_df = in_category_raw.distinct().select(
    vid('product', 'asin').alias('src'), vid('category', 'name').alias('dst'),
    F.lit('in_category').alias('etype'))
in_category_edges_df.printSchema(); in_category_edges_df.show(3, truncate=80)

Then we define the HAS_PRODUCT_IMAGE edge.

python 
has_product_image_raw = decode_columns(read_tsv('product_image', ['asin', 'url']), ['asin', 'url'])
has_product_image_edges_df = has_product_image_raw.distinct().select(
    vid('product', 'asin').alias('src'), vid('image', 'url').alias('dst'),
    F.lit('has_product_image').alias('etype'))
has_product_image_edges_df.printSchema(); has_product_image_edges_df.show(3, truncate=80)

Define Graph

Now that we have defined our vertices and edges, we save each type as a Delta table and register it in the Hive metastore.

python 
vertex_source_dfs = {
    'product': product_vertices_df, 'category': category_vertices_df, 'image': image_vertices_df,
    'reviewimage': reviewimage_vertices_df, 'reviewer': reviewer_vertices_df,
    'review': review_vertices_df}
edge_source_dfs = {
    'product_alsobuy_product': also_buy_edges_df, 'product_alsoview_product': also_view_edges_df,
    'product_incategory_category': in_category_edges_df,
    'product_hasproductimage_image': has_product_image_edges_df,
    'review_reviews_product': reviews_edges_df, 'reviewer_wrote_review': wrote_edges_df,
    'review_hasreviewimage_reviewimage': has_review_image_edges_df}

OUTPUT_ROOT = 'hdfs://master:9000/user/hive/warehouse/amazon'
spark.sql('CREATE DATABASE IF NOT EXISTS amazon')
for name, df in {**vertex_source_dfs, **edge_source_dfs}.items():
    path = f'{OUTPUT_ROOT}/{name}'
    df.write.format('delta').mode('overwrite').option('overwriteSchema', 'true').save(path)
    spark.sql(f'DROP TABLE IF EXISTS amazon.amazon_{name}')
    spark.sql(f"CREATE TABLE amazon.amazon_{name} USING DELTA LOCATION '{path}'")

Finally, we can create the whole graph. All vertex tables are unioned into a single vertex DataFrame and all edge tables into a single edge DataFrame.

python 
def load_table(name, column):
    # `type` / `etype` are constant per table; as a literal, a filter on them
    # prunes the other tables from the plan.
    frame = spark.table(f'amazon.amazon_{name}')
    kind = frame.select(column).first()[column]
    return frame.drop(column).withColumn(column, F.lit(kind))

def union_all(frames):
    return reduce(lambda left, right: left.unionByName(right, allowMissingColumns=True), frames)

vertices = union_all([load_table(name, 'type') for name in vertex_source_dfs])
edges = union_all([load_table(name, 'etype') for name in edge_source_dfs])
g = GraphFrame(vertices, edges)

Queries

Select a query below to view its logic and description. Every query runs on the graph g built above; the definitions below are shared by several queries.

python 
from graphframes import GraphFrame
from pyspark.sql import functions as F

SEED_ASIN = "0000031895"
TARGET_ASIN = "B00JM8YXXG"

products = g.vertices.filter(F.col("type") == "product")
reviews = g.vertices.filter(F.col("type") == "review")
reviewers = g.vertices.filter(F.col("type") == "reviewer")


def edges_of(etype):
    return g.edges.filter(F.col("etype") == etype).select("src", "dst", "etype")


# Co-buy graph: Product -[ALSO_BUY]-> Product, plus an undirected copy.
cobuy_vertices = products.select("id", "type", "asin", "title", "price_min")
cobuy_edges = edges_of("also_buy")
cobuy = GraphFrame(cobuy_vertices, cobuy_edges)
cobuy_edges_undirected = cobuy_edges.unionByName(
    cobuy_edges.selectExpr("dst AS src", "src AS dst", "etype")
)
cobuy_undirected = GraphFrame(cobuy_vertices, cobuy_edges_undirected)


def display_title(vertex):
    """coalesce(v.title, v.asin) for a motif vertex column."""
    return F.coalesce(F.col(f"{vertex}.title"), F.col(f"{vertex}.asin"))


# Q16's ranking, used by Q1, Q16 and Q30: products by incoming REVIEWS.
product_review_counts = (
    g.find("(r)-[e]->(p)")
    .filter(
        (F.col("e.etype") == "reviews")
        & (F.col("r.type") == "review")
        & (F.col("p.type") == "product")
    )
    .groupBy(F.col("p.id").alias("id"))
    .agg(F.count("*").alias("ReviewCount"))
)
seed_products = (
    product_review_counts
    .join(products.select("id", "asin", "title"), "id")
    .orderBy(F.col("ReviewCount").desc(), F.col("asin").asc())
    .limit(200)
)

# Q18's ranking, used by Q15 and Q18: reviewers by outgoing WROTE.
top_reviewers = (
    g.find("(u)-[w]->(r)")
    .filter(
        (F.col("w.etype") == "wrote")
        & (F.col("u.type") == "reviewer")
        & (F.col("r.type") == "review")
    )
    .groupBy(
        F.col("u.id").alias("id"),
        F.col("u.reviewerID").alias("reviewerID"),
        F.col("u.name").alias("name"),
    )
    .agg(F.count("*").alias("ReviewCount"))
    .orderBy(F.col("ReviewCount").desc(), F.col("reviewerID").asc())
    .limit(200)
)




# Used by Q27-Q29.
CHECKPOINT_DIR = 'hdfs://master:9000/tmp/spark-checkpoints'

def component_sizes(frame, column, id_name, size_name):
    return (
        frame.groupBy(column)
        .agg(F.count('*').alias(size_name))
        .select(F.col(column).alias(id_name), size_name)
        .orderBy(F.col(size_name).desc(), F.col(id_name).asc())
        .limit(100)
    )
  • Every Product property Neo4j would hold is a column; Products that exist only because an edge points at them have null attributes, so coalesce(p.title, p.asin) behaves as in Cypher (a metadata product with an empty title keeps "").
  • Cypher patterns written -[:ALSO_BUY]- are undirected. GraphFrames walks src -> dst, so cobuy_undirected adds a reversed copy of every edge and does NOT deduplicate: Neo4j treats each stored relationship as its own match in an undirected pattern.
  • Cypher MATCH requires the relationships of one pattern to be distinct but lets nodes repeat. Where that matters (variable-length paths Q9/Q22/Q23, the 4-cycle Q11) it is spelled out at the query.
  • Where Cypher's ORDER BY leaves ties, a secondary key on the returned identifiers is added so the 200/100 rows are reproducible. It never changes which rows a tie-free ordering would return.
  • Q16/Q18 attach :SeedProduct / :TopReviewer labels in Neo4j. DataFrames are immutable, so Q16/Q18 return their ranking only (the SET is dropped), and Q1/Q15/Q30 recompute that ranking inline -- as the TigerGraph translation does, with the same tie-break (count DESC, key ASC).
  • Q6/Q7 (BFS) and Q26-Q29 (graph algorithms) run eagerly, so they are built inside the runner's measured job group, not here; their sections below document the translation and are replaced at load time.
1. Neighborhood
2. Path
3. Pattern
4. Aggregation
5. Analytics
6. Hybrid
7. Others
No queries in this category.

Q1. People Who Bought This Also Bought (Seed Expansion)

Subcategory: T1.1 One-Hop Neighbors

This query expands one hop from the seed products through ALSO_BUY edges, excludes the seeds themselves, and ranks neighboring products by how many seed products link to them.

python 
seed_ids = seed_products.select(F.col("id").alias("seed_id"))
q1 = (
    cobuy.find("(seed)-[e]->(other)")
    .join(seed_ids, F.col("seed.id") == F.col("seed_id"), "left_semi")
    .join(seed_ids, F.col("other.id") == F.col("seed_id"), "left_anti")
    .groupBy(F.col("other.asin").alias("ASIN"),
             display_title("other").alias("CoBuyProduct"))
    .agg(F.countDistinct("seed.id").alias("LinkedSeeds"))
    .orderBy(F.col("LinkedSeeds").desc(), F.col("ASIN").asc())
    .limit(200)
)

GraphFrames notes

(seed:SeedProduct)-[:ALSO_BUY]->(other) WHERE NOT other:SeedProduct

Equivalent query in GQL standard:

gql 
MATCH (seed:SeedProduct)-[:ALSO_BUY]->(other:Product)
WHERE NOT other:SeedProduct
RETURN other.asin AS ASIN,
    coalesce(other.title, other.asin) AS CoBuyProduct,
    count(DISTINCT seed) AS LinkedSeeds
ORDER BY LinkedSeeds DESC
LIMIT 200;

Q2. Two-Hop Co-Buy Recommendations

Subcategory: T1.2 k-Hop Neighborhood

This query expands from a selected product through two outgoing ALSO_BUY hops and returns distinct second-hop products.

python 
q2 = (
    cobuy.find("(s)-[e1]->(m); (m)-[e2]->(rec)")
    .filter((F.col("s.asin") == SEED_ASIN) & (F.col("rec.asin") != SEED_ASIN))
    .select(F.col("rec.asin").alias("ASIN"), display_title("rec").alias("Product"))
    .distinct()
    .orderBy("Product", "ASIN")
    .limit(200)
)

Equivalent query in GQL standard:

gql 
MATCH (:Product {asin: $asin})-[:ALSO_BUY]->(:Product)-[:ALSO_BUY]->(rec:Product)
WHERE rec.asin <> $asin
RETURN DISTINCT rec.asin AS ASIN, coalesce(rec.title, rec.asin) AS Product
ORDER BY Product
LIMIT $limit;

Q3. Products Exactly Two Co-Buy Hops Away

Subcategory: T1.3 Exact kth-Hop Neighbors

This query returns products exactly two ALSO_BUY hops from a selected product, excluding direct one-hop recommendations.

python 
q3 = (
    cobuy.find("(s)-[e1]->(m); (m)-[e2]->(second); !(s)-[]->(second)")
    .filter((F.col("s.asin") == SEED_ASIN) & (F.col("second.id") != F.col("s.id")))
    .select(F.col("second.asin").alias("ASIN"), display_title("second").alias("Product"))
    .distinct()
    .orderBy("Product", "ASIN")
    .limit(200)
)

GraphFrames notes

NOT (seed)-[:ALSO_BUY]->(second) is a GraphFrames negated edge.

Equivalent query in GQL standard:

gql 
MATCH (seed:Product {asin: $asin})-[:ALSO_BUY]->(:Product)-[:ALSO_BUY]->(second:Product)
WHERE second <> seed
  AND NOT (seed)-[:ALSO_BUY]->(second)
RETURN DISTINCT second.asin AS ASIN, coalesce(second.title, second.asin) AS Product
ORDER BY Product
LIMIT $limit;

Q4. Highly Rated Co-Buy Neighbors

Subcategory: T1.4 Neighborhood with Property Filter

This query filters one-hop co-buy neighbors to products whose review average meets a minimum rating threshold.

python 
q4 = (
    g.find("(s)-[e1]->(p); (r)-[e2]->(p)")
    .filter(
        (F.col("e1.etype") == "also_buy") & (F.col("e2.etype") == "reviews")
        & (F.col("s.type") == "product") & (F.col("p.type") == "product")
        & (F.col("r.type") == "review")
        & (F.col("s.asin") == SEED_ASIN)
        & F.col("r.overall").isNotNull()
    )
    .groupBy(F.col("p.asin").alias("ASIN"), display_title("p").alias("Product"))
    .agg(F.avg("r.overall").alias("_avg"), F.count("r.id").alias("ReviewCount"))
    .filter(F.col("_avg") >= 4.0)
    .select("ASIN", "Product", F.round("_avg", 2).alias("AvgRating"), "ReviewCount")
    .orderBy(F.col("AvgRating").desc(), F.col("ReviewCount").desc(), F.col("ASIN").asc())
    .limit(100)
)

GraphFrames notes

ORDER BY AvgRating refers to the returned, rounded AvgRating.

Equivalent query in GQL standard:

gql 
MATCH (:Product {asin: $asin})-[:ALSO_BUY]->(p:Product)<-[:REVIEWS]-(r:Review)
WHERE r.overall IS NOT NULL
WITH p, avg(r.overall) AS AvgRating, count(r) AS ReviewCount
WHERE AvgRating >= $minRating
RETURN p.asin AS ASIN, coalesce(p.title, p.asin) AS Product, round(AvgRating, 2) AS AvgRating, ReviewCount
ORDER BY AvgRating DESC, ReviewCount DESC
LIMIT $limit;

Q5. High-Rating Helpful Review Records

Subcategory: T2.2 Edge Filtering

Reviews are modeled as nodes between reviewers and products, so this query filters review-edge records by rating and helpful-vote properties.

python 
q5 = (
    g.find("(u)-[w]->(r); (r)-[rv]->(p)")
    .filter(
        (F.col("w.etype") == "wrote") & (F.col("rv.etype") == "reviews")
        & (F.col("u.type") == "reviewer") & (F.col("r.type") == "review")
        & (F.col("p.type") == "product")
        & (F.col("r.overall") >= 4.0)
        & (F.coalesce(F.col("r.vote"), F.lit(0)) >= 10)
    )
    .select(
        F.col("u.reviewerID").alias("ReviewerID"),
        F.col("p.asin").alias("ASIN"),
        F.col("r.overall").alias("Rating"),
        F.col("r.vote").alias("HelpfulVotes"),
    )
    .orderBy(F.col("HelpfulVotes").desc(), F.col("Rating").desc(),
             F.col("ReviewerID").asc(), F.col("ASIN").asc())
    .limit(100)
)

Equivalent query in GQL standard:

gql 
MATCH (u:Reviewer)-[:WROTE]->(r:Review)-[:REVIEWS]->(p:Product)
WHERE r.overall >= $minRating
  AND coalesce(r.vote, 0) >= $minHelpfulVotes
RETURN u.reviewerID AS ReviewerID, p.asin AS ASIN, r.overall AS Rating, r.vote AS HelpfulVotes
ORDER BY HelpfulVotes DESC, Rating DESC
LIMIT $limit;

Q6. Reachability Between Products in the Co-Buy Graph

Subcategory: T2.3 Reachability between Two Nodes

This query tests whether one product can reach another product through up to four ALSO_BUY hops.

python 
def build_q6():
    path = cobuy.bfs(
        fromExpr=f"asin = '{SEED_ASIN}'",
        toExpr=f"asin = '{TARGET_ASIN}'",
        maxPathLength=4,
    ).limit(1)
    return path.agg((F.count("*") > 0).alias("reachable"))

GraphFrames notes

cobuy.bfs(asin = SEED -> asin = TARGET, maxPathLength=4).limit(1), then count(*) > 0 AS reachable. A path of length 1..4 exists iff the shortest one is at most 4 hops, so BFS decides it.

This one runs eagerly (a BFS / graph algorithm), so it is written as a function that the benchmark runner calls inside the measured Spark job group.

Equivalent query in GQL standard:

gql 
MATCH p = ANY (:Product {asin: $sourceAsin})-[:ALSO_BUY]->{1,4}(:Product {asin: $targetAsin})
RETURN true AS reachable;

Q7. Shortest Co-Buy Path Between Products

Subcategory: T2.4 Shortest Path

This query finds the shortest directed co-buy path between two products within a bounded depth.

python 
def build_q7():
    bfs = cobuy.bfs(
        fromExpr=f"asin = '{SEED_ASIN}'",
        toExpr=f"asin = '{TARGET_ASIN}'",
        maxPathLength=6,
    )
    intermediate = sorted(
        [column for column in bfs.columns if re.fullmatch(r"v\d+", column)],
        key=lambda column: int(column[1:]),
    )
    vertex_columns = ["from"] + intermediate + ["to"]
    hop_count = len([column for column in bfs.columns if re.fullmatch(r"e\d+", column)])
    return (
        bfs.select(
            F.array(*[F.col(f"{column}.asin") for column in vertex_columns]).alias("ProductPath"),
            F.lit(hop_count).alias("HopCount"),
        )
        .orderBy("ProductPath")
        .limit(1)
    )

GraphFrames notes

cobuy.bfs(..., maxPathLength=6), path rendered as the array of asins, one path kept (Neo4j's shortestPath also returns one).

This one runs eagerly (a BFS / graph algorithm), so it is written as a function that the benchmark runner calls inside the measured Spark job group.

Equivalent query in GQL standard:

gql 
MATCH p = SHORTEST (:Product {asin: $sourceAsin})-[:ALSO_BUY]->{1,6}(:Product {asin: $targetAsin})
RETURN p;

Q8. Simple Co-Buy Paths up to Three Hops

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

This query enumerates acyclic co-buy paths between two products up to three hops.

python 
def seed_to_target_paths(hops, all_nodes_distinct):
    names = ["a", "b", "c", "d"][: hops + 1]
    pattern = "; ".join(f"({names[i]})-[e{i + 1}]->({names[i + 1]})" for i in range(hops))
    cond = (F.col("a.asin") == SEED_ASIN) & (F.col(f"{names[-1]}.asin") == TARGET_ASIN)
    if all_nodes_distinct:
        for i in range(len(names)):
            for j in range(i + 1, len(names)):
                cond = cond & (F.col(f"{names[i]}.id") != F.col(f"{names[j]}.id"))
    elif hops == 3:
        # Relationship isomorphism: e1 = (a,b) and e3 = (c,d) are the same
        # edge iff a = c and b = d. Adjacent edges could only coincide on a
        # self-loop, and ALSO_BUY has none.
        cond = cond & ~((F.col("a.id") == F.col("c.id")) & (F.col("b.id") == F.col("d.id")))
    return (
        cobuy.find(pattern)
        .filter(cond)
        .select(
            F.array(*[F.col(f"{n}.asin") for n in names]).alias("ProductPath"),
            F.lit(hops).alias("HopCount"),
            sum((F.coalesce(F.col(f"{n}.price_min"), F.lit(0.0)) for n in names),
                F.lit(0.0)).alias("PathPrice"),
        )
    )


q8 = (
    seed_to_target_paths(1, True)
    .unionByName(seed_to_target_paths(2, True))
    .unionByName(seed_to_target_paths(3, True))
    .select("ProductPath", "HopCount")
    .orderBy("HopCount", "ProductPath")
    .limit(200)
)

GraphFrames notes

Paths of each length are separate motifs; the WHERE clause demands every node of the path be distinct.

Equivalent query in GQL standard:

gql 
MATCH p = (src:Product {asin: $sourceAsin})-[:ALSO_BUY]->{1,3}(dst:Product {asin: $targetAsin})
RETURN p
LIMIT $limit;

Q9. Top Weighted Co-Buy Paths by Product Price

Subcategory: T2.6 top-K Weighted Paths

This query ranks bounded co-buy paths by the sum of product prices along each path.

python 
q9 = (
    seed_to_target_paths(1, False)
    .unionByName(seed_to_target_paths(2, False))
    .unionByName(seed_to_target_paths(3, False))
    .orderBy(F.col("PathPrice").desc(), F.col("HopCount").asc(), F.col("ProductPath").asc())
    .limit(100)
)

GraphFrames notes

No node-distinctness clause here, so only relationship isomorphism applies.

It reuses seed_to_target_paths from Q8.

Equivalent query in GQL standard:

gql 
MATCH p = (:Product {asin: $sourceAsin})-[:ALSO_BUY]->{1,3}(:Product {asin: $targetAsin})
RETURN p
LIMIT $k;

Q10. Co-Buy Product Triangles

Subcategory: T3.1 Triangle Detection

This query finds directed triangles in the product co-buy graph.

python 
q10 = (
    cobuy.find("(a)-[e1]->(b); (b)-[e2]->(c); (c)-[e3]->(a)")
    .filter((F.col("a.asin") < F.col("b.asin")) & (F.col("b.asin") < F.col("c.asin")))
    .select(F.col("a.asin").alias("Product1"), F.col("b.asin").alias("Product2"),
            F.col("c.asin").alias("Product3"))
    .limit(200)
)

Equivalent query in GQL standard:

gql 
MATCH (p1:Product)-[:ALSO_BUY]->(p2:Product)-[:ALSO_BUY]->(p3:Product)-[:ALSO_BUY]->(p1)
WHERE p1.asin < p2.asin AND p2.asin < p3.asin
RETURN p1.asin AS Product1, p2.asin AS Product2, p3.asin AS Product3
LIMIT $limit;

Q11. Four-Product Co-Buy Cycles

Subcategory: T3.2 Square Detection

This query searches for four-product cycles formed by directed ALSO_BUY relationships, where the second and fourth products must be different.

python 
q11 = (
    cobuy.find("(a)-[e1]->(b); (b)-[e2]->(c); (c)-[e3]->(d); (d)-[e4]->(a)")
    .filter(
        (F.col("a.asin") < F.col("b.asin"))
        & (F.col("a.asin") < F.col("c.asin"))
        & (F.col("a.asin") < F.col("d.asin"))
        & (F.col("b.id") != F.col("d.id"))
    )
    .select(F.col("a.asin").alias("Product1"), F.col("b.asin").alias("Product2"),
            F.col("c.asin").alias("Product3"), F.col("d.asin").alias("Product4"))
    .limit(200)
)

GraphFrames notes

p1 < p2, p3, p4 and p2 <> p4. Without the last clause Cypher would accept a->b->c->b->a (four distinct relationships, p2 = p4); the workload now excludes it, so all four products are distinct (p1 <> p3 follows from p1 < p3, and adjacent ones would need a self-loop).

Equivalent query in GQL standard:

gql 
MATCH (p1:Product)-[:ALSO_BUY]->(p2:Product)-[:ALSO_BUY]->(p3:Product)-[:ALSO_BUY]->(p4:Product)-[:ALSO_BUY]->(p1)
WHERE p1.asin < p2.asin AND p1.asin < p3.asin AND p1.asin < p4.asin AND p2 <> p4
RETURN p1.asin AS Product1, p2.asin AS Product2, p3.asin AS Product3, p4.asin AS Product4
LIMIT $limit;

Q12. Products with Many Co-Buy Recommendations

Subcategory: T3.3 Star Pattern Query

This query finds star-like products that point to many co-buy recommendation neighbors.

python 
q12 = (
    cobuy.find("(center)-[e]->(neighbor)")
    .groupBy(F.col("center.asin").alias("ASIN"), display_title("center").alias("Product"))
    .agg(F.countDistinct("neighbor.id").alias("RecommendationCount"))
    .filter(F.col("RecommendationCount") >= 50)
    .orderBy(F.col("RecommendationCount").desc(), F.col("ASIN").asc())
    .limit(100)
)

Equivalent query in GQL standard:

gql 
MATCH (center:Product)-[:ALSO_BUY]->(neighbor:Product)
WITH center, count(DISTINCT neighbor) AS RecommendationCount
WHERE RecommendationCount >= $minDegree
RETURN center.asin AS ASIN, coalesce(center.title, center.asin) AS Product, RecommendationCount
ORDER BY RecommendationCount DESC
LIMIT $limit;

Q13. Four-Product Co-Buy Cliques

Subcategory: T3.4 4-Clique

This query treats co-buy links as undirected and finds four products where every pair is connected by an ALSO_BUY relationship.

python 
q13 = (
    cobuy_undirected.find(
        "(a)-[e1]->(b); (a)-[e2]->(c); (a)-[e3]->(d); "
        "(b)-[e4]->(c); (b)-[e5]->(d); (c)-[e6]->(d)"
    )
    .filter(
        (F.col("a.asin") < F.col("b.asin"))
        & (F.col("b.asin") < F.col("c.asin"))
        & (F.col("c.asin") < F.col("d.asin"))
    )
    .select(F.col("a.asin").alias("Product1"), F.col("b.asin").alias("Product2"),
            F.col("c.asin").alias("Product3"), F.col("d.asin").alias("Product4"))
    .limit(200)
)

Equivalent query in GQL standard:

gql 
MATCH (p1:Product)-[:ALSO_BUY]-(p2:Product),
      (p1)-[:ALSO_BUY]-(p3:Product),
      (p1)-[:ALSO_BUY]-(p4:Product),
      (p2)-[:ALSO_BUY]-(p3),
      (p2)-[:ALSO_BUY]-(p4),
      (p3)-[:ALSO_BUY]-(p4)
WHERE p1.asin < p2.asin AND p2.asin < p3.asin AND p3.asin < p4.asin
RETURN p1.asin AS Product1, p2.asin AS Product2, p3.asin AS Product3, p4.asin AS Product4
LIMIT $limit;

Q14. Reviewers Who Reviewed the Same Product

Subcategory: T3.5 Two-hop Join Pattern

This query joins reviewers through products they have both reviewed.

python 
q14_reviewer_products = (
    g.find("(u)-[w]->(r); (r)-[rv]->(p)")
    .filter(
        (F.col("w.etype") == "wrote") & (F.col("rv.etype") == "reviews")
        & (F.col("u.type") == "reviewer") & (F.col("r.type") == "review")
        & (F.col("p.type") == "product")
    )
    .select(F.col("u.reviewerID").alias("reviewerID"), F.col("p.id").alias("pid"))
    .distinct()
)
q14 = (
    q14_reviewer_products.alias("x")
    .join(q14_reviewer_products.alias("y"), "pid")
    .filter(F.col("x.reviewerID") < F.col("y.reviewerID"))
    .groupBy(F.col("x.reviewerID").alias("Reviewer1"), F.col("y.reviewerID").alias("Reviewer2"))
    .agg(F.countDistinct("pid").alias("SharedProductCount"))
    .orderBy(F.col("SharedProductCount").desc(), F.col("Reviewer1").asc(), F.col("Reviewer2").asc())
    .limit(100)
)

GraphFrames notes

count(DISTINCT p) per reviewer pair only depends on which products each reviewer reviewed, so the (reviewer, product) pairs are deduplicated before the self-join; the result is the same as joining every review path.

Equivalent query in GQL standard:

gql 
MATCH (u1:Reviewer)-[:WROTE]->(:Review)-[:REVIEWS]->(p:Product)<-[:REVIEWS]-(:Review)<-[:WROTE]-(u2:Reviewer)
WHERE u1.reviewerID < u2.reviewerID
RETURN u1.reviewerID AS Reviewer1, u2.reviewerID AS Reviewer2, count(DISTINCT p) AS SharedProductCount
ORDER BY SharedProductCount DESC
LIMIT $limit;

Q15. Reviewers Who Write the Longest Reviews

Subcategory: T4.1 Ungrouped Node Degree Statistics by Label

This query takes the top reviewers identified earlier and ranks them by the average length of their review text, returning reviewer identity, review count, and average review length.

python 
q15 = (
    g.find("(u)-[w]->(r)")
    .filter((F.col("w.etype") == "wrote") & (F.col("u.type") == "reviewer")
            & (F.col("r.type") == "review"))
    .join(top_reviewers.select(F.col("id").alias("top_id")),
          F.col("u.id") == F.col("top_id"), "left_semi")
    .groupBy(
        F.col("u.reviewerID").alias("ReviewerID"),
        F.coalesce(F.col("u.name"), F.col("u.reviewerID")).alias("ReviewerName"),
    )
    .agg(F.count("r.id").alias("ReviewCount"),
         F.avg(F.length("r.reviewText")).alias("AvgReviewLength"))
    .orderBy(F.col("AvgReviewLength").desc(), F.col("ReviewerID").asc())
    .limit(200)
)

GraphFrames notes

size(string) in Cypher and length() in Spark both count characters.

Equivalent query in GQL standard:

gql 
MATCH (u:TopReviewer)-[:WROTE]->(r:Review)
RETURN u.reviewerID AS ReviewerID,
    coalesce(u.name, u.reviewerID) AS ReviewerName,
    count(r) AS ReviewCount,
    avg(size(r.reviewText)) AS AvgReviewLength
ORDER BY AvgReviewLength DESC
LIMIT 200;

Q16. Products with the Most Reviews

Subcategory: T4.2 Grouped Node Degree Statistics by Label

This query finds the 200 products with the highest number of reviews, marks them as seed products, and returns each product's identifier, display title, and review count.

python 
q16 = (
    seed_products
    .select(F.col("asin").alias("ASIN"),
            F.coalesce(F.col("title"), F.col("asin")).alias("Product"),
            "ReviewCount")
    .orderBy(F.col("ReviewCount").desc(), F.col("ASIN").asc())
)

Equivalent query in GQL standard:

gql 
MATCH (p:Product)<-[:REVIEWS]-(r:Review)
WITH p, count(r) AS ReviewCount
ORDER BY ReviewCount DESC
LIMIT 200
SET p:SeedProduct
RETURN p.asin AS ASIN, coalesce(p.title, p.asin) AS Product, ReviewCount;

Q17. Highly Rated Products with Many Reviews

Subcategory: T4.2 Grouped Node Degree Statistics by Label

This query finds products with at least 50 reviews and an average rating of at least 4.0, returning the most-reviewed high-quality products first.

python 
q17 = (
    g.find("(r)-[e]->(p)")
    .filter(
        (F.col("e.etype") == "reviews") & (F.col("r.type") == "review")
        & (F.col("p.type") == "product") & F.col("r.overall").isNotNull()
    )
    .groupBy(F.col("p.asin").alias("ASIN"), display_title("p").alias("Product"))
    .agg(F.avg("r.overall").alias("_avg"), F.count("r.id").alias("ReviewCount"))
    .filter((F.col("ReviewCount") >= 50) & (F.col("_avg") >= 4.0))
    .select("ASIN", "Product", F.round("_avg", 2).alias("AvgRating"), "ReviewCount")
    .orderBy(F.col("ReviewCount").desc(), F.col("AvgRating").desc(), F.col("ASIN").asc())
    .limit(200)
)

Equivalent query in GQL standard:

gql 
MATCH (p:Product)<-[:REVIEWS]-(r:Review)
WHERE r.overall IS NOT NULL
WITH p, avg(r.overall) AS AvgRating, count(r) AS ReviewCount
WHERE ReviewCount >= 50 AND AvgRating >= 4.0
RETURN p.asin AS ASIN,
    coalesce(p.title, p.asin) AS Product,
    round(AvgRating, 2) AS AvgRating,
    ReviewCount
ORDER BY ReviewCount DESC, AvgRating DESC
LIMIT 200;

Q18. Most Active Reviewers

Subcategory: T4.2 Grouped Node Degree Statistics by Label

This query identifies the 200 reviewers who wrote the most reviews, marks them as top reviewers, and returns their IDs, names, and review counts.

python 
q18 = (
    top_reviewers
    .select(F.col("reviewerID").alias("ReviewerID"),
            F.coalesce(F.col("name"), F.col("reviewerID")).alias("ReviewerName"),
            "ReviewCount")
    .orderBy(F.col("ReviewCount").desc(), F.col("ReviewerID").asc())
)

Equivalent query in GQL standard:

gql 
MATCH (u:Reviewer)-[:WROTE]->(r:Review)
WITH u, count(r) AS ReviewCount
ORDER BY ReviewCount DESC
LIMIT 200
SET u:TopReviewer
RETURN u.reviewerID AS ReviewerID,
    coalesce(u.name, u.reviewerID) AS ReviewerName,
    ReviewCount
ORDER BY ReviewCount DESC;

Q19. Products with the Most Images

Subcategory: T4.2 Grouped Node Degree Statistics by Label

This query counts product image neighbors for each product and returns the products with the largest number of attached images.

python 
q19 = (
    g.find("(p)-[e]->(i)")
    .filter(
        (F.col("e.etype") == "has_product_image")
        & (F.col("p.type") == "product") & (F.col("i.type") == "image")
    )
    .groupBy(F.col("p.asin").alias("ASIN"), display_title("p").alias("Product"))
    .agg(F.count("i.id").alias("ImageCount"))
    .orderBy(F.col("ImageCount").desc(), F.col("ASIN").asc())
    .limit(200)
)

Equivalent query in GQL standard:

gql 
MATCH (p:Product)-[:HAS_PRODUCT_IMAGE]->(i:Image)
WITH p.asin AS ASIN, coalesce(p.title, p.asin) AS Product, count(i) AS ImageCount
RETURN ASIN, Product, ImageCount
ORDER BY ImageCount DESC
LIMIT 200;

Q20. Review Edge Rating Summary

Subcategory: T4.3 Ungrouped Edge Count and Property Statistics

Reviews are modeled as nodes between reviewers and products, so this query summarizes review-edge records and their rating properties.

python 
q20 = (
    g.find("(u)-[w]->(r); (r)-[rv]->(p)")
    .filter(
        (F.col("w.etype") == "wrote") & (F.col("rv.etype") == "reviews")
        & (F.col("u.type") == "reviewer") & (F.col("r.type") == "review")
        & (F.col("p.type") == "product")
    )
    .agg(
        F.count("r.id").alias("ReviewCount"),
        F.avg("r.overall").alias("AvgRating"),
        F.min("r.overall").alias("MinRating"),
        F.max("r.overall").alias("MaxRating"),
    )
)

Equivalent query in GQL standard:

gql 
MATCH (:Reviewer)-[:WROTE]->(r:Review)-[:REVIEWS]->(:Product)
RETURN count(r) AS ReviewCount,
       avg(r.overall) AS AvgRating,
       min(r.overall) AS MinRating,
       max(r.overall) AS MaxRating;

Q21. Category-Level Review Rating Summary

Subcategory: T4.4 Grouped Edge Count and Property Statistics

This query groups review records by product category and summarizes ratings within each category.

python 
q21 = (
    g.find("(p)-[ic]->(c); (r)-[rv]->(p)")
    .filter(
        (F.col("ic.etype") == "in_category") & (F.col("rv.etype") == "reviews")
        & (F.col("p.type") == "product") & (F.col("c.type") == "category")
        & (F.col("r.type") == "review")
    )
    .groupBy(F.col("c.name").alias("Category"))
    .agg(F.count("r.id").alias("ReviewCount"),
         F.round(F.avg("r.overall"), 2).alias("AvgRating"))
    .orderBy(F.col("ReviewCount").desc(), F.col("Category").asc())
    .limit(100)
)

Equivalent query in GQL standard:

gql 
MATCH (c:Category)<-[:IN_CATEGORY]-(p:Product)<-[:REVIEWS]-(r:Review)
RETURN c.name AS Category,
       count(r) AS ReviewCount,
       round(avg(r.overall), 2) AS AvgRating
ORDER BY ReviewCount DESC
LIMIT $limit;

Q22. Co-Buy Path Counts by Length

Subcategory: T4.5 Path Counts by Length

This query counts bounded ALSO_BUY paths from a selected product and groups them by path length.

python 
def seed_paths(hops):
    names = ["a", "b", "c", "d"][: hops + 1]
    pattern = "; ".join(f"({names[i]})-[e{i + 1}]->({names[i + 1]})" for i in range(hops))
    cond = F.col("a.asin") == SEED_ASIN
    if hops == 3:
        cond = cond & ~((F.col("a.id") == F.col("c.id")) & (F.col("b.id") == F.col("d.id")))
    return (
        cobuy.find(pattern)
        .filter(cond)
        .select(F.lit(hops).alias("PathLength"), F.col(f"{names[-1]}.id").alias("target_id"))
    )


q22_paths = seed_paths(1).unionByName(seed_paths(2)).unionByName(seed_paths(3))
q22 = (
    q22_paths
    .groupBy("PathLength")
    .agg(F.count("*").alias("PathCount"),
         F.countDistinct("target_id").alias("DistinctTargetCount"))
    .orderBy("PathLength")
)

GraphFrames notes

Every ALSO_BUY path of 1..3 hops from the seed, under relationship isomorphism (see seed_to_target_paths). count(path) counts paths, count(DISTINCT target) their endpoints.

Equivalent query in GQL standard:

gql 
MATCH path = (:Product {asin: $asin})-[:ALSO_BUY]->{1,3}(target:Product)
RETURN length(path) AS PathLength,
       count(path) AS PathCount,
       count(DISTINCT target) AS DistinctTargetCount
ORDER BY PathLength;

Q23. Co-Buy Path Counts by Length and Endpoint Category

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

This query groups co-buy paths by path length and by the category of the endpoint product.

python 
q23_in_category = edges_of("in_category").select(
    F.col("src").alias("target_id"), F.col("dst").alias("category_id"))
q23_categories = g.vertices.filter(F.col("type") == "category").select(
    F.col("id").alias("category_id"), F.col("name").alias("EndpointCategory"))
q23 = (
    q22_paths
    .join(q23_in_category, "target_id")
    .join(q23_categories, "category_id")
    .groupBy("PathLength", "EndpointCategory")
    .agg(F.count("*").alias("PathCount"),
         F.countDistinct("target_id").alias("EndpointProductCount"))
    .orderBy(F.col("PathLength").asc(), F.col("PathCount").desc(),
             F.col("EndpointCategory").asc())
    .limit(200)
)

GraphFrames notes

Each path is extended by every IN_CATEGORY edge of its endpoint.

It reuses q22_paths from Q22.

Equivalent query in GQL standard:

gql 
MATCH path = (:Product {asin: $asin})-[:ALSO_BUY]->{1,3}(target:Product)-[:IN_CATEGORY]->(c:Category)
RETURN length(path) AS PathLength,
       c.name AS EndpointCategory,
       count(path) AS PathCount,
       count(DISTINCT target) AS EndpointProductCount
ORDER BY PathLength, PathCount DESC
LIMIT $limit;

Q24. Products with Both Many Reviews and Images

Subcategory: T4.7 Union and Intersection

This query intersects products with many reviews and products with multiple images to find items with both strong review and media signals.

python 
q24_reviewed = product_review_counts.filter(F.col("ReviewCount") >= 50).select(
    F.col("id").alias("reviewed_id"))
q24 = (
    g.find("(p)-[e]->(i)")
    .filter(
        (F.col("e.etype") == "has_product_image")
        & (F.col("p.type") == "product") & (F.col("i.type") == "image")
    )
    .join(q24_reviewed, F.col("p.id") == F.col("reviewed_id"), "left_semi")
    .groupBy(F.col("p.asin").alias("ASIN"), display_title("p").alias("Product"))
    .agg(F.count("i.id").alias("ImageCount"))
    .filter(F.col("ImageCount") >= 5)
    .orderBy(F.col("ImageCount").desc(), F.col("ASIN").asc())
    .limit(100)
)

Equivalent query in GQL standard:

gql 
MATCH (p:Product)<-[:REVIEWS]-(r:Review)
WITH p, count(r) AS ReviewCount
WHERE ReviewCount >= $minReviews
WITH collect(p) AS ReviewedProducts
MATCH (p:Product)-[:HAS_PRODUCT_IMAGE]->(i:Image)
WITH ReviewedProducts, p, count(i) AS ImageCount
WHERE p IN ReviewedProducts AND ImageCount >= $minImages
RETURN p.asin AS ASIN, coalesce(p.title, p.asin) AS Product, ImageCount
ORDER BY ImageCount DESC
LIMIT $limit;

Q25. Top Products by Co-Buy Degree Centrality

Subcategory: T5.1 Degree Centrality

This query computes degree centrality directly by counting products connected through incoming or outgoing co-buy edges.

python 
q25 = (
    cobuy_undirected.find("(p)-[e]->(neighbor)")
    .groupBy(F.col("p.asin").alias("ASIN"), display_title("p").alias("Product"))
    .agg(F.countDistinct("neighbor.id").alias("CoBuyDegree"))
    .orderBy(F.col("CoBuyDegree").desc(), F.col("ASIN").asc())
    .limit(100)
)

Equivalent query in GQL standard:

gql 
MATCH (p:Product)-[:ALSO_BUY]-(neighbor:Product)
RETURN p.asin AS ASIN,
       coalesce(p.title, p.asin) AS Product,
       count(DISTINCT neighbor) AS CoBuyDegree
ORDER BY CoBuyDegree DESC
LIMIT $limit;

Q26. PageRank on the Co-Buy Product Graph

Subcategory: T5.2 PageRank

This query projects the product co-buy graph and runs PageRank to rank influential products.

python 
def build_q26():
    ranked = cobuy.pageRank(resetProbability=0.15, maxIter=20)
    return (
        ranked.vertices
        .select(F.col("asin").alias("ASIN"), F.col("title").alias("Product"),
                F.col("pagerank").alias("score"))
        .orderBy(F.col("score").desc(), F.col("ASIN").asc())
        .limit(100)
    )

GraphFrames notes

cobuy.pageRank(resetProbability=0.15, maxIter=20) on the directed (NATURAL) graph -- Neo4j GDS defaults dampingFactor 0.85 and maxIterations 20, written out. Neo4j returns the raw title (no coalesce). GraphX scales scores differently from GDS, so compare rankings, not values.

This one runs eagerly (a BFS / graph algorithm), so it is written as a function that the benchmark runner calls inside the measured Spark job group.

Note: Cannot be expressed in standard GQL because Graph Data Science (GDS) procedures and PageRank are not supported by the current GQL standard.

Q27. Weakly Connected Components of the Co-Buy Graph

Subcategory: T5.3 Connected Components / Weakly Connected Components

This query treats co-buy links as undirected and reports the largest weakly connected components.

python 
def build_q27():
    spark.sparkContext.setCheckpointDir(CHECKPOINT_DIR)
    return component_sizes(cobuy.connectedComponents(), "component",
                           "componentId", "ComponentSize")

GraphFrames notes

cobuy.connectedComponents() (inherently undirected), component sizes, top 100.

This one runs eagerly (a BFS / graph algorithm), so it is written as a function that the benchmark runner calls inside the measured Spark job group.

Note: Cannot be expressed in standard GQL because Graph Data Science (GDS) procedures and connected-components algorithms are not supported by the current GQL standard.

Q28. Strongly Connected Components of the Co-Buy Graph

Subcategory: T5.4 Strongly Connected Components

This query runs strongly connected components on the directed co-buy product graph.

python 
def build_q28():
    return component_sizes(cobuy.stronglyConnectedComponents(maxIter=20), "component",
                           "componentId", "ComponentSize")

GraphFrames notes

cobuy.stronglyConnectedComponents(maxIter=20) on the directed graph, component sizes, top 100.

This one runs eagerly (a BFS / graph algorithm), so it is written as a function that the benchmark runner calls inside the measured Spark job group.

Note: Cannot be expressed in standard GQL because Graph Data Science (GDS) procedures and strongly connected component algorithms are not supported by the current GQL standard.

Q29. Louvain Communities in the Co-Buy Graph

Subcategory: T5.5 Community Detection

This query detects communities of products in the undirected co-buy graph using Louvain.

python 
def build_q29():
    return component_sizes(cobuy_undirected.labelPropagation(maxIter=20), "label",
                           "communityId", "CommunitySize")

GraphFrames notes

Core GraphFrames has no Louvain; as in the Yelp workload, labelPropagation(maxIter=20) on the undirected graph substitutes and is recorded as implementation=label_propagation_substitute.

This one runs eagerly (a BFS / graph algorithm), so it is written as a function that the benchmark runner calls inside the measured Spark job group.

Note: Cannot be expressed in standard GQL because Graph Data Science (GDS) procedures and community detection algorithms are not supported by the current GQL standard.

Q30. Second-Hop Co-Buy Influence (Ripple Effect)

Subcategory: T2.1 Node Filtering (Single Node Paths); T3.6 Line Pattern; T4.1 Ungrouped Node Degree Statistics by Label

This query starts from the most-reviewed seed product, follows two ALSO_BUY hops, and ranks second-hop products by how many first-hop neighbors connect back to them.

python 
q30_top_seed = seed_products.orderBy(F.col("ReviewCount").desc(), F.col("asin").asc()) \
    .limit(1).select(F.col("id").alias("top_seed_id"))
q30 = (
    cobuy.find("(seed)-[e1]->(first); (first)-[e2]->(second)")
    .join(q30_top_seed, F.col("seed.id") == F.col("top_seed_id"), "left_semi")
    .join(seed_ids, F.col("first.id") == F.col("seed_id"), "left_anti")
    .filter(F.col("second.id") != F.col("seed.id"))
    .groupBy(F.col("second.asin").alias("ASIN"),
             display_title("second").alias("SecondHopProduct"))
    .agg(F.countDistinct("first.id").alias("SharedNeighbors"))
    .orderBy(F.col("SharedNeighbors").desc(), F.col("ASIN").asc())
    .limit(200)
)

GraphFrames notes

The single most-reviewed seed, its first hops that are not seeds, and their second hops other than the seed.

It reuses seed_ids from Q1.

Equivalent query in GQL standard:

gql 
MATCH (seed:SeedProduct)<-[:REVIEWS]-(review:Review)
WITH seed, count(review) AS ReviewCount
ORDER BY ReviewCount DESC
LIMIT 1
MATCH (seed)-[:ALSO_BUY]->(firstHop:Product)
WHERE NOT firstHop:SeedProduct
MATCH (firstHop)-[:ALSO_BUY]->(secondHop:Product)
WHERE secondHop <> seed
RETURN secondHop.asin AS ASIN,
    coalesce(secondHop.title, secondHop.asin) AS SecondHopProduct,
    count(DISTINCT firstHop) AS SharedNeighbors
ORDER BY SharedNeighbors DESC
LIMIT 200;

Q31. Are Reviews with More Images More Helpful?

Subcategory: T2.1 Node Filtering (Single Node Paths); T4.2 Grouped Node Degree Statistics by Label

This query finds highly helpful reviews, counts any images attached to each review, and returns reviewer, rating, helpful-vote, and image-count fields for comparison.

python 
q31_images = edges_of("has_review_image").select(
    F.col("src").alias("review_id_v"), F.col("dst").alias("ri_id"))
q31 = (
    reviews.filter(F.col("vote").isNotNull())
    .select(F.col("id").alias("review_id_v"), "reviewerID", "overall", "vote")
    .join(q31_images, "review_id_v", "left")
    .groupBy(
        F.col("reviewerID").alias("ReviewerID"),
        F.coalesce(F.col("reviewerID"), F.lit("Unknown")).alias("ReviewerName"),
        F.col("overall").alias("Rating"),
        F.col("vote").alias("HelpfulVotes"),
    )
    .agg(F.count("ri_id").alias("ImageCount"))
    .orderBy(F.col("HelpfulVotes").desc(), F.col("ReviewerID").asc(),
             F.col("Rating").desc(), F.col("ImageCount").desc())
    .limit(200)
)

GraphFrames notes

The WITH groups by (reviewerID, name, rating, votes), not by review, so reviews sharing all four are pooled and their images counted together. OPTIONAL MATCH becomes a left join; count(ri) skips the unmatched nulls.

Equivalent query in GQL standard:

gql 
MATCH (r:Review)
WHERE r.vote IS NOT NULL
OPTIONAL MATCH (r)-[:HAS_IMAGE]->(ri:ReviewImage)
WITH r.reviewerID AS ReviewerID,
    coalesce(r.reviewerID, "Unknown") AS ReviewerName,
    r.overall AS Rating,
    r.vote AS HelpfulVotes,
    count(ri) AS ImageCount
RETURN ReviewerID, ReviewerName, Rating, HelpfulVotes, ImageCount
ORDER BY HelpfulVotes DESC
LIMIT 200;