Schema Diagram and Graph Statistics
Since GraphFrames has no support for printing a graph schema, we reuse the one output from Neo4j.
Scale factor 100 of the LDBC Social Network Benchmark, Business Intelligence workload, loaded
from the same CSV files as the Neo4j and TigerGraph implementations. Only the
initial_snapshot is loaded; the inserts and deletes
update streams are not applied, because this implementation runs read-only queries.
A GraphFrame has a single vertex DataFrame and a single edge DataFrame, so every vertex carries
its concrete type in a type column and every edge its relationship in an
etype column. The Neo4j label hierarchy is flattened into concrete types:
Comment and Post are the Message nodes,
City, Country and Continent the Place nodes,
and University and Company the Organisation nodes.
Node Statistics
| Node Label | Count |
|---|---|
Comment |
185,495,476 |
Post |
64,029,217 |
Forum |
4,611,436 |
Person |
473,001 |
Tag |
16,080 |
University |
6,380 |
Company |
1,575 |
City |
1,343 |
Country |
111 |
TagClass |
71 |
Continent |
6 |
Edge Statistics
| Relationship | Count |
|---|---|
(:Place)-[:IS_PART_OF]->(:Place) |
1,454 |
(:TagClass)-[:IS_SUBCLASS_OF]->(:TagClass) |
70 |
(:Organisation)-[:IS_LOCATED_IN]->(:Place) |
7,955 |
(:Tag)-[:HAS_TYPE]->(:TagClass) |
16,080 |
(:Comment)-[:HAS_CREATOR]->(:Person) |
185,495,476 |
(:Comment)-[:IS_LOCATED_IN]->(:Country) |
185,495,476 |
(:Comment)-[:REPLY_OF]->(:Comment) |
85,829,276 |
(:Comment)-[:REPLY_OF]->(:Post) |
99,666,200 |
(:Forum)-[:CONTAINER_OF]->(:Post) |
64,029,217 |
(:Forum)-[:HAS_MEMBER]->(:Person) |
303,838,931 |
(:Forum)-[:HAS_MODERATOR]->(:Person) |
4,611,436 |
(:Forum)-[:HAS_TAG]->(:Tag) |
15,195,472 |
(:Person)-[:HAS_INTEREST]->(:Tag) |
11,057,039 |
(:Person)-[:IS_LOCATED_IN]->(:City) |
473,001 |
(:Person)-[:KNOWS]->(:Person) |
18,655,515 |
(:Person)-[:LIKES]->(:Comment) |
184,325,690 |
(:Person)-[:LIKES]->(:Post) |
98,423,296 |
(:Post)-[:HAS_CREATOR]->(:Person) |
64,029,217 |
(:Comment)-[:HAS_TAG]->(:Tag) |
238,074,593 |
(:Post)-[:HAS_TAG]->(:Tag) |
78,380,259 |
(:Post)-[:IS_LOCATED_IN]->(:Country) |
64,029,217 |
(:Person)-[:STUDY_AT]->(:University) |
378,582 |
(:Person)-[:WORK_AT]->(:Company) |
1,029,492 |
Every count above matches the Neo4j load of the same dataset exactly, table by table.
Total nodes: 254,634,696 · Total relationships: 1,703,042,944
Disk space: 41.9 GB of Delta tables (Parquet), 125.6 GB on HDFS with 3-way replication
Total import time: 379.0 s (6m 19s) to read the CSV files and write the 34 Delta tables, one after another
Data Loading
Overview
Data import is handled in four parts. First the raw CSV files are read into DataFrames. Then each node and edge type is defined as its own DataFrame, and each is written to HDFS as a Delta table registered in the Hive Metastore. Finally, the queries build the graph from those tables. Storing the tables once means no query pays for parsing 115 GB of CSV, and the same tables can be read by any engine that speaks Delta Lake.
Spark reads the very same files as the Neo4j and TigerGraph imports: |-delimited CSV
with no header row, one directory of Spark output shards per entity, read here
from shared storage that every worker mounts. The schema of each file is given explicitly. Rows
that do not fit it fail the read (FAILFAST, the counterpart of
--bad-tolerance=0 in Neo4j), and empty fields become nulls, as they do with
Neo4j’s --ignore-empty-strings.
GraphFrames needs one id space across all vertices, but LDBC ids collide between entities:
tags, places and organisations are all numbered from 0. Each vertex id is therefore a string
"<family>:<LDBC id>", with the original number kept in
ldbc_id. The prefix follows the Neo4j label family rather than the concrete type,
so Comment and Post share message:: their ids never
collide in the source, and an edge can then point at a message without knowing which kind it
is. creationDate is a TIMESTAMP that keeps the milliseconds of the
source, and birthday a DATE.
DataFrames
We defined a few helper functions to read the source directories and to build the vertex ids, timestamps and edge DataFrames.
TS_FORMAT = "yyyy-MM-dd'T'HH:mm:ss.SSSXXX"
def csv_dir(entity):
return CSV_ROOT / entity
def csv_bytes(entity):
return sum(p.stat().st_size for p in csv_dir(entity).glob('part-*.csv'))
def read_ldbc(entity, schema):
df = (spark.read.format('csv').schema(schema)
.option('sep', '|').option('header', 'false').option('mode', 'FAILFAST')
.option('pathGlobFilter', 'part-*.csv')
.load(csv_dir(entity).as_uri()))
return df.limit(ROW_LIMIT) if ROW_LIMIT else df
def vertex_id(family, column):
return F.concat(F.lit(f'{family}:'), F.col(column).cast('string'))
def ts(column):
return F.to_timestamp(F.col(column), TS_FORMAT)
def edges(entity, etype, src_family, dst_family, dynamic=True, extra=()):
cols = (['creationDate STRING'] if dynamic else []) + ['src BIGINT', 'dst BIGINT'] + \
[f'{name} {typ}' for name, typ in extra]
raw = read_ldbc(entity, ', '.join(cols))
out = [vertex_id(src_family, 'src').alias('src'), vertex_id(dst_family, 'dst').alias('dst'),
F.lit(etype).alias('etype')]
if dynamic:
out.append(ts('creationDate').alias('creationDate'))
out += [F.col(name) for name, _ in extra]
return raw.select(*out)
Define Nodes
First we define the City, Country and Continent nodes. They come from one source directory, whose :LABEL column gives the concrete type, and share the place: id prefix.
PLACE_SCHEMA = 'id BIGINT, name STRING, url STRING, label STRING'
place_raw = read_ldbc('static/Place', PLACE_SCHEMA)
def place_vertices(label):
return place_raw.filter(F.col('label') == label).select(
vertex_id('place', 'id').alias('id'), F.lit(label.lower()).alias('type'),
F.col('id').alias('ldbc_id'), 'name', 'url')
city_vertices_df = place_vertices('City')
country_vertices_df = place_vertices('Country')
continent_vertices_df = place_vertices('Continent')
city_vertices_df.printSchema(); city_vertices_df.show(3, truncate=80)
Then we define the Company and University nodes, which share the organisation: prefix.
ORGANISATION_SCHEMA = 'id BIGINT, label STRING, name STRING, url STRING'
organisation_raw = read_ldbc('static/Organisation', ORGANISATION_SCHEMA)
def organisation_vertices(label):
return organisation_raw.filter(F.col('label') == label).select(
vertex_id('organisation', 'id').alias('id'), F.lit(label.lower()).alias('type'),
F.col('id').alias('ldbc_id'), 'name', 'url')
company_vertices_df = organisation_vertices('Company')
university_vertices_df = organisation_vertices('University')
university_vertices_df.printSchema(); university_vertices_df.show(3, truncate=80)
Then we define the TagClass node.
TAGCLASS_SCHEMA = 'id BIGINT, name STRING, url STRING'
tagclass_raw = read_ldbc('static/TagClass', TAGCLASS_SCHEMA)
tagclass_vertices_df = tagclass_raw.select(
vertex_id('tagclass', 'id').alias('id'), F.lit('tagclass').alias('type'),
F.col('id').alias('ldbc_id'), 'name', 'url')
tagclass_vertices_df.printSchema(); tagclass_vertices_df.show(3, truncate=80)
Then we define the Tag node.
TAG_SCHEMA = 'id BIGINT, name STRING, url STRING'
tag_raw = read_ldbc('static/Tag', TAG_SCHEMA)
tag_vertices_df = tag_raw.select(
vertex_id('tag', 'id').alias('id'), F.lit('tag').alias('type'),
F.col('id').alias('ldbc_id'), 'name', 'url')
tag_vertices_df.printSchema(); tag_vertices_df.show(3, truncate=80)
Then we define the Forum node.
FORUM_SCHEMA = 'creationDate STRING, id BIGINT, title STRING'
forum_raw = read_ldbc('dynamic/Forum', FORUM_SCHEMA)
forum_vertices_df = forum_raw.select(
vertex_id('forum', 'id').alias('id'), F.lit('forum').alias('type'),
F.col('id').alias('ldbc_id'), ts('creationDate').alias('creationDate'), 'title')
forum_vertices_df.printSchema(); forum_vertices_df.show(3, truncate=80)
Then we define the Person node. speaks and email are ;-separated lists in the source and become arrays.
PERSON_SCHEMA = ('creationDate STRING, id BIGINT, firstName STRING, lastName STRING, gender STRING, '
'birthday STRING, locationIP STRING, browserUsed STRING, speaks STRING, email STRING')
person_raw = read_ldbc('dynamic/Person', PERSON_SCHEMA)
person_vertices_df = person_raw.select(
vertex_id('person', 'id').alias('id'), F.lit('person').alias('type'),
F.col('id').alias('ldbc_id'), ts('creationDate').alias('creationDate'),
'firstName', 'lastName', 'gender', F.to_date('birthday', 'yyyy-MM-dd').alias('birthday'),
'locationIP', 'browserUsed', F.split('speaks', ';').alias('speaks'),
F.split('email', ';').alias('email'))
person_vertices_df.printSchema(); person_vertices_df.show(3, truncate=80)
Then we define the Comment node.
COMMENT_SCHEMA = 'creationDate STRING, id BIGINT, locationIP STRING, browserUsed STRING, content STRING, length INT'
comment_raw = read_ldbc('dynamic/Comment', COMMENT_SCHEMA)
comment_vertices_df = comment_raw.select(
vertex_id('message', 'id').alias('id'), F.lit('comment').alias('type'),
F.col('id').alias('ldbc_id'), ts('creationDate').alias('creationDate'),
'locationIP', 'browserUsed', 'content', 'length')
comment_vertices_df.printSchema(); comment_vertices_df.show(3, truncate=80)
Finally we define the Post node. A post has either an imageFile or content; the empty one is read as null.
POST_SCHEMA = ('creationDate STRING, id BIGINT, imageFile STRING, locationIP STRING, browserUsed STRING, '
'language STRING, content STRING, length INT')
post_raw = read_ldbc('dynamic/Post', POST_SCHEMA)
post_vertices_df = post_raw.select(
vertex_id('message', 'id').alias('id'), F.lit('post').alias('type'),
F.col('id').alias('ldbc_id'), ts('creationDate').alias('creationDate'),
'imageFile', 'locationIP', 'browserUsed', 'language', 'content', 'length')
post_vertices_df.printSchema(); post_vertices_df.show(3, truncate=80)
Define Edges
Each of the 23 source relationship groups becomes one edge DataFrame, with src,
dst and etype (the Neo4j relationship type in lower case), plus
creationDate for the dynamic edges and classYear or
workFrom for STUDY_AT and WORK_AT.
First we define the IS_PART_OF edge from Place to Place.
place_ispartof_place_edges_df = edges('static/Place_isPartOf_Place', 'is_part_of', 'place', 'place', dynamic=False)
place_ispartof_place_edges_df.printSchema(); place_ispartof_place_edges_df.show(3, truncate=80)
Then we define the IS_SUBCLASS_OF edge from TagClass to TagClass.
tagclass_issubclassof_tagclass_edges_df = edges('static/TagClass_isSubclassOf_TagClass', 'is_subclass_of', 'tagclass', 'tagclass', dynamic=False)
tagclass_issubclassof_tagclass_edges_df.printSchema(); tagclass_issubclassof_tagclass_edges_df.show(3, truncate=80)
Then we define the IS_LOCATED_IN edge from Organisation to Place.
organisation_islocatedin_place_edges_df = edges('static/Organisation_isLocatedIn_Place', 'is_located_in', 'organisation', 'place', dynamic=False)
organisation_islocatedin_place_edges_df.printSchema(); organisation_islocatedin_place_edges_df.show(3, truncate=80)
Then we define the HAS_TYPE edge from Tag to TagClass.
tag_hastype_tagclass_edges_df = edges('static/Tag_hasType_TagClass', 'has_type', 'tag', 'tagclass', dynamic=False)
tag_hastype_tagclass_edges_df.printSchema(); tag_hastype_tagclass_edges_df.show(3, truncate=80)
Then we define the HAS_CREATOR edge from Comment to Person.
comment_hascreator_person_edges_df = edges('dynamic/Comment_hasCreator_Person', 'has_creator', 'message', 'person')
comment_hascreator_person_edges_df.printSchema(); comment_hascreator_person_edges_df.show(3, truncate=80)
Then we define the IS_LOCATED_IN edge from Comment to Country.
comment_islocatedin_country_edges_df = edges('dynamic/Comment_isLocatedIn_Country', 'is_located_in', 'message', 'place')
comment_islocatedin_country_edges_df.printSchema(); comment_islocatedin_country_edges_df.show(3, truncate=80)
Then we define the REPLY_OF edge from Comment to Comment.
comment_replyof_comment_edges_df = edges('dynamic/Comment_replyOf_Comment', 'reply_of', 'message', 'message')
comment_replyof_comment_edges_df.printSchema(); comment_replyof_comment_edges_df.show(3, truncate=80)
Then we define the REPLY_OF edge from Comment to Post.
comment_replyof_post_edges_df = edges('dynamic/Comment_replyOf_Post', 'reply_of', 'message', 'message')
comment_replyof_post_edges_df.printSchema(); comment_replyof_post_edges_df.show(3, truncate=80)
Then we define the CONTAINER_OF edge from Forum to Post.
forum_containerof_post_edges_df = edges('dynamic/Forum_containerOf_Post', 'container_of', 'forum', 'message')
forum_containerof_post_edges_df.printSchema(); forum_containerof_post_edges_df.show(3, truncate=80)
Then we define the HAS_MEMBER edge from Forum to Person.
forum_hasmember_person_edges_df = edges('dynamic/Forum_hasMember_Person', 'has_member', 'forum', 'person')
forum_hasmember_person_edges_df.printSchema(); forum_hasmember_person_edges_df.show(3, truncate=80)
Then we define the HAS_MODERATOR edge from Forum to Person.
forum_hasmoderator_person_edges_df = edges('dynamic/Forum_hasModerator_Person', 'has_moderator', 'forum', 'person')
forum_hasmoderator_person_edges_df.printSchema(); forum_hasmoderator_person_edges_df.show(3, truncate=80)
Then we define the HAS_TAG edge from Forum to Tag.
forum_hastag_tag_edges_df = edges('dynamic/Forum_hasTag_Tag', 'has_tag', 'forum', 'tag')
forum_hastag_tag_edges_df.printSchema(); forum_hastag_tag_edges_df.show(3, truncate=80)
Then we define the HAS_INTEREST edge from Person to Tag.
person_hasinterest_tag_edges_df = edges('dynamic/Person_hasInterest_Tag', 'has_interest', 'person', 'tag')
person_hasinterest_tag_edges_df.printSchema(); person_hasinterest_tag_edges_df.show(3, truncate=80)
Then we define the IS_LOCATED_IN edge from Person to City.
person_islocatedin_city_edges_df = edges('dynamic/Person_isLocatedIn_City', 'is_located_in', 'person', 'place')
person_islocatedin_city_edges_df.printSchema(); person_islocatedin_city_edges_df.show(3, truncate=80)
Then we define the KNOWS edge from Person to Person.
person_knows_person_edges_df = edges('dynamic/Person_knows_Person', 'knows', 'person', 'person')
person_knows_person_edges_df.printSchema(); person_knows_person_edges_df.show(3, truncate=80)
Then we define the LIKES edge from Person to Comment.
person_likes_comment_edges_df = edges('dynamic/Person_likes_Comment', 'likes', 'person', 'message')
person_likes_comment_edges_df.printSchema(); person_likes_comment_edges_df.show(3, truncate=80)
Then we define the LIKES edge from Person to Post.
person_likes_post_edges_df = edges('dynamic/Person_likes_Post', 'likes', 'person', 'message')
person_likes_post_edges_df.printSchema(); person_likes_post_edges_df.show(3, truncate=80)
Then we define the HAS_CREATOR edge from Post to Person.
post_hascreator_person_edges_df = edges('dynamic/Post_hasCreator_Person', 'has_creator', 'message', 'person')
post_hascreator_person_edges_df.printSchema(); post_hascreator_person_edges_df.show(3, truncate=80)
Then we define the HAS_TAG edge from Comment to Tag.
comment_hastag_tag_edges_df = edges('dynamic/Comment_hasTag_Tag', 'has_tag', 'message', 'tag')
comment_hastag_tag_edges_df.printSchema(); comment_hastag_tag_edges_df.show(3, truncate=80)
Then we define the HAS_TAG edge from Post to Tag.
post_hastag_tag_edges_df = edges('dynamic/Post_hasTag_Tag', 'has_tag', 'message', 'tag')
post_hastag_tag_edges_df.printSchema(); post_hastag_tag_edges_df.show(3, truncate=80)
Then we define the IS_LOCATED_IN edge from Post to Country.
post_islocatedin_country_edges_df = edges('dynamic/Post_isLocatedIn_Country', 'is_located_in', 'message', 'place')
post_islocatedin_country_edges_df.printSchema(); post_islocatedin_country_edges_df.show(3, truncate=80)
Then we define the STUDY_AT edge from Person to University.
person_studyat_university_edges_df = edges('dynamic/Person_studyAt_University', 'study_at', 'person', 'organisation', extra=(('classYear', 'INT'),))
person_studyat_university_edges_df.printSchema(); person_studyat_university_edges_df.show(3, truncate=80)
Finally we define the WORK_AT edge from Person to Company.
person_workat_company_edges_df = edges('dynamic/Person_workAt_Company', 'work_at', 'person', 'organisation', extra=(('workFrom', 'INT'),))
person_workat_company_edges_df.printSchema(); person_workat_company_edges_df.show(3, truncate=80)
Delta Tables on HDFS
The 11 vertex and 23 edge DataFrames are collected by name.
vertex_sources = {
'comment': (comment_vertices_df, 'dynamic/Comment'),
'post': (post_vertices_df, 'dynamic/Post'),
'forum': (forum_vertices_df, 'dynamic/Forum'),
'person': (person_vertices_df, 'dynamic/Person'),
'tag': (tag_vertices_df, 'static/Tag'),
'tagclass': (tagclass_vertices_df, 'static/TagClass'),
'city': (city_vertices_df, 'static/Place'),
'country': (country_vertices_df, 'static/Place'),
'continent': (continent_vertices_df, 'static/Place'),
'company': (company_vertices_df, 'static/Organisation'),
'university': (university_vertices_df, 'static/Organisation'),
}
edge_sources = {
'place_ispartof_place': (place_ispartof_place_edges_df, 'static/Place_isPartOf_Place'),
'tagclass_issubclassof_tagclass': (tagclass_issubclassof_tagclass_edges_df, 'static/TagClass_isSubclassOf_TagClass'),
'organisation_islocatedin_place': (organisation_islocatedin_place_edges_df, 'static/Organisation_isLocatedIn_Place'),
'tag_hastype_tagclass': (tag_hastype_tagclass_edges_df, 'static/Tag_hasType_TagClass'),
'comment_hascreator_person': (comment_hascreator_person_edges_df, 'dynamic/Comment_hasCreator_Person'),
'comment_islocatedin_country': (comment_islocatedin_country_edges_df, 'dynamic/Comment_isLocatedIn_Country'),
'comment_replyof_comment': (comment_replyof_comment_edges_df, 'dynamic/Comment_replyOf_Comment'),
'comment_replyof_post': (comment_replyof_post_edges_df, 'dynamic/Comment_replyOf_Post'),
'forum_containerof_post': (forum_containerof_post_edges_df, 'dynamic/Forum_containerOf_Post'),
'forum_hasmember_person': (forum_hasmember_person_edges_df, 'dynamic/Forum_hasMember_Person'),
'forum_hasmoderator_person': (forum_hasmoderator_person_edges_df, 'dynamic/Forum_hasModerator_Person'),
'forum_hastag_tag': (forum_hastag_tag_edges_df, 'dynamic/Forum_hasTag_Tag'),
'person_hasinterest_tag': (person_hasinterest_tag_edges_df, 'dynamic/Person_hasInterest_Tag'),
'person_islocatedin_city': (person_islocatedin_city_edges_df, 'dynamic/Person_isLocatedIn_City'),
'person_knows_person': (person_knows_person_edges_df, 'dynamic/Person_knows_Person'),
'person_likes_comment': (person_likes_comment_edges_df, 'dynamic/Person_likes_Comment'),
'person_likes_post': (person_likes_post_edges_df, 'dynamic/Person_likes_Post'),
'post_hascreator_person': (post_hascreator_person_edges_df, 'dynamic/Post_hasCreator_Person'),
'comment_hastag_tag': (comment_hastag_tag_edges_df, 'dynamic/Comment_hasTag_Tag'),
'post_hastag_tag': (post_hastag_tag_edges_df, 'dynamic/Post_hasTag_Tag'),
'post_islocatedin_country': (post_islocatedin_country_edges_df, 'dynamic/Post_isLocatedIn_Country'),
'person_studyat_university': (person_studyat_university_edges_df, 'dynamic/Person_studyAt_University'),
'person_workat_company': (person_workat_company_edges_df, 'dynamic/Person_workAt_Company'),
}
print(len(vertex_sources), 'vertex tables,', len(edge_sources), 'edge tables')
Each one is written to HDFS as a Delta table and registered in the Hive Metastore as
snb100.snb100_<name>. The number of partitions follows the size of the source
CSV, about one per 256 MB, and rows are hash-partitioned and sorted by id for
vertices and by src for edges.
MAX_RECORDS_PER_FILE = 5_000_000
TIMING_COLUMNS = ['table', 'kind', 'rows', 'seconds', 'partitions', 'num_files', 'delta_bytes',
'source_csv_bytes', 'started_at', 'ended_at']
with open(TIMING_FILE, 'w') as fh:
fh.write('\t'.join(TIMING_COLUMNS) + '\n')
IMPORT_STARTED = time.perf_counter()
IMPORT_STARTED_AT = dt.datetime.now(dt.timezone.utc).isoformat(timespec='seconds')
def partitions_for(entity):
if SMOKE:
return 1
return max(1, math.ceil(csv_bytes(entity) / (256 * 1024 ** 2)))
def write_delta_hdfs_source(name):
if not WRITE_DELTA:
print(f'Skipped {name}; set WRITE_DELTA=True first.')
return
kind = 'vertex' if name in vertex_sources else 'edge'
dataframe, entity = (vertex_sources if kind == 'vertex' else edge_sources)[name]
sort_columns = ['id'] if kind == 'vertex' else ['src', 'dst']
table_name = f'{DATABASE}.{DATABASE}_{name}'
table_path = f'{OUTPUT_ROOT}/{name}'
partitions = partitions_for(entity)
started_at = dt.datetime.now(dt.timezone.utc).isoformat(timespec='seconds')
t0 = time.perf_counter()
(dataframe.repartition(partitions, sort_columns[0]).sortWithinPartitions(*sort_columns)
.write.format('delta').mode('overwrite').option('overwriteSchema', 'true')
.option('maxRecordsPerFile', str(MAX_RECORDS_PER_FILE)).save(table_path))
seconds = time.perf_counter() - t0
ended_at = dt.datetime.now(dt.timezone.utc).isoformat(timespec='seconds')
spark.sql(f'CREATE DATABASE IF NOT EXISTS {DATABASE}')
spark.sql(f'DROP TABLE IF EXISTS {table_name}')
escaped_path = table_path.replace("'", "''")
spark.sql(f"CREATE TABLE {table_name} USING DELTA LOCATION '{escaped_path}'")
detail = spark.sql(f'DESCRIBE DETAIL {table_name}').select('numFiles', 'sizeInBytes').first()
rows = spark.table(table_name).count()
record = dict(table=name, kind=kind, rows=rows, seconds=f'{seconds:.3f}', partitions=partitions,
num_files=detail['numFiles'], delta_bytes=detail['sizeInBytes'],
source_csv_bytes=csv_bytes(entity), started_at=started_at, ended_at=ended_at)
with open(TIMING_FILE, 'a') as fh:
fh.write('\t'.join(str(record[c]) for c in TIMING_COLUMNS) + '\n')
print(f'Written {table_name}: {rows:,} rows in {seconds:,.1f} s, '
f'{detail["numFiles"]} files, {detail["sizeInBytes"] / 1e9:.2f} GB -> {table_path}')
write_delta_hdfs_source('comment')
write_delta_hdfs_source('post')
write_delta_hdfs_source('forum')
write_delta_hdfs_source('person')
write_delta_hdfs_source('tag')
write_delta_hdfs_source('tagclass')
write_delta_hdfs_source('city')
write_delta_hdfs_source('country')
write_delta_hdfs_source('continent')
write_delta_hdfs_source('company')
write_delta_hdfs_source('university')
write_delta_hdfs_source('place_ispartof_place')
write_delta_hdfs_source('tagclass_issubclassof_tagclass')
write_delta_hdfs_source('organisation_islocatedin_place')
write_delta_hdfs_source('tag_hastype_tagclass')
write_delta_hdfs_source('comment_hascreator_person')
write_delta_hdfs_source('comment_islocatedin_country')
write_delta_hdfs_source('comment_replyof_comment')
write_delta_hdfs_source('comment_replyof_post')
write_delta_hdfs_source('forum_containerof_post')
write_delta_hdfs_source('forum_hasmember_person')
write_delta_hdfs_source('forum_hasmoderator_person')
write_delta_hdfs_source('forum_hastag_tag')
write_delta_hdfs_source('person_hasinterest_tag')
write_delta_hdfs_source('person_islocatedin_city')
write_delta_hdfs_source('person_knows_person')
write_delta_hdfs_source('person_likes_comment')
write_delta_hdfs_source('person_likes_post')
write_delta_hdfs_source('post_hascreator_person')
write_delta_hdfs_source('comment_hastag_tag')
write_delta_hdfs_source('post_hastag_tag')
write_delta_hdfs_source('post_islocatedin_country')
write_delta_hdfs_source('person_studyat_university')
write_delta_hdfs_source('person_workat_company')
After the writes, every table’s row count is checked against the Neo4j load, every timestamp and date column is checked for nulls (a value that fails to parse becomes null silently), and the union of all vertex ids is checked for duplicates. All three checks pass.
Define Graph
Now that the tables are stored, the queries build the graph from them: all vertex tables are unioned into a single vertex DataFrame and all edge tables into a single edge DataFrame. The plans stay lazy and uncached, so one query cannot warm another.
_eid is a deterministic edge identity, used to keep the edges of a motif distinct
where Cypher’s relationship-isomorphic matching would. It is built from
etype, src and dst rather than with
monotonically_increasing_id(): a non-deterministic column would stop Spark from
pushing an etype filter down to the Delta files, so every query would scan all
1.7 billion edges.
from functools import reduce
VERTEX_TABLES = [f'snb100.snb100_{t}' for t in (
'comment', 'post', 'forum', 'person', 'tag', 'tagclass',
'city', 'country', 'continent', 'company', 'university')]
EDGE_TABLES = [f'snb100.snb100_{t}' for t in (
'place_ispartof_place', 'tagclass_issubclassof_tagclass', 'organisation_islocatedin_place',
'tag_hastype_tagclass', 'comment_hascreator_person', 'comment_islocatedin_country',
'comment_replyof_comment', 'comment_replyof_post', 'forum_containerof_post',
'forum_hasmember_person', 'forum_hasmoderator_person', 'forum_hastag_tag',
'person_hasinterest_tag', 'person_islocatedin_city', 'person_knows_person',
'person_likes_comment', 'person_likes_post', 'post_hascreator_person', 'comment_hastag_tag',
'post_hastag_tag', 'post_islocatedin_country', 'person_studyat_university',
'person_workat_company')]
def union_all(dataframes):
return reduce(lambda a, b: a.unionByName(b, allowMissingColumns=True), dataframes)
vertices_df = union_all([spark.table(t) for t in VERTEX_TABLES])
edges_df = union_all([spark.table(t) for t in EDGE_TABLES]).withColumn(
'_eid', F.concat_ws('|', 'etype', 'src', 'dst'))
vertices_df.printSchema()
edges_df.printSchema()
Queries
The 41 queries are those of the Neo4j implementation, with the same numbering and the same
parameter values. Each query builds its own GraphFrame from only the node and edge types it
touches, with only the columns it reads: with 254.6 million vertices, a motif over the full
graph would read every message body. Graph operations (find, bfs,
shortestPaths, aggregateMessages, triangleCount,
pageRank, connectedComponents,
stronglyConnectedComponents, labelPropagation) are used wherever
GraphFrames has a corresponding primitive; grouping, ranking and set logic use DataFrame
operations.
The queries share a few helper functions. message_roots() finds the post at the
root of every message’s thread, the REPLY_OF*0.. pattern of six BI queries, by pointer
jumping: each round replaces a comment’s ancestor by that ancestor’s current
ancestor, so it finishes in a logarithmic number of rounds in the depth of the deepest thread.
bellman_ford() computes weighted shortest distances with
aggregateMessages for the three weighted-path queries, which GraphFrames has no
primitive for. The helpers define functions only and are run before each query.
import re
from functools import reduce
from pyspark.sql import Window
from pyspark.storagelevel import StorageLevel
from graphframes import GraphFrame
from graphframes.lib import AggregateMessages as AM
def V(*types):
return vertices_df.filter(F.col('type').isin(*types))
def E(*etypes):
return edges_df.filter(F.col('etype').isin(*etypes))
def T(value):
return F.to_timestamp(F.lit(value))
def pid(ldbc_id):
return f'person:{ldbc_id}'
def both_directions(edges):
rev = (edges.withColumnRenamed('src', '_src').withColumnRenamed('dst', 'src')
.withColumnRenamed('_src', 'dst'))
return edges.unionByName(rev)
def persons(*cols):
return V('person').select('id', 'ldbc_id', *cols)
def knows_graph(*person_cols, undirected=False):
edges = E('knows').select('src', 'dst', '_eid', 'creationDate')
return GraphFrame(persons(*person_cols), both_directions(edges) if undirected else edges)
def persons_in_country(country_name):
g = GraphFrame(
V('person', 'city', 'country').select('id', 'type', 'name'),
E('is_located_in', 'is_part_of').filter(~F.col('src').startswith('message:'))
.select('src', 'dst', 'etype'))
return (g.find('(p)-[e1]->(ci); (ci)-[e2]->(co)')
.filter((F.col('p.type') == 'person') & (F.col('e1.etype') == 'is_located_in') &
(F.col('ci.type') == 'city') & (F.col('e2.etype') == 'is_part_of') &
(F.col('co.type') == 'country') & (F.col('co.name') == country_name))
.select(F.col('p.id').alias('person'), F.col('ci.id').alias('city'),
F.col('ci.name').alias('city_name')))
def tag_id(name):
return V('tag').filter(F.col('name') == name).select(F.col('id').alias('tag'))
def creators():
return E('has_creator').select(F.col('src').alias('message'), F.col('dst').alias('person'))
def message_roots():
parent = E('reply_of').select(F.col('src').alias('message'), F.col('dst').alias('anc'))
is_comment = V('comment').select(F.col('id').alias('anc'), F.lit(True).alias('anc_is_comment'))
cur = parent.join(is_comment, 'anc', 'left').localCheckpoint()
while True:
active = cur.filter(F.col('anc_is_comment').isNotNull())
if active.isEmpty():
break
lookup = cur.select(F.col('message').alias('anc'), F.col('anc').alias('next_anc'),
F.col('anc_is_comment').alias('next_is_comment'))
stepped = active.join(lookup, 'anc').select(
'message', F.col('next_anc').alias('anc'), F.col('next_is_comment').alias('anc_is_comment'))
cur = cur.filter(F.col('anc_is_comment').isNull()).unionByName(stepped).localCheckpoint()
posts = V('post').select(F.col('id').alias('message'), F.col('id').alias('root'))
return cur.select('message', F.col('anc').alias('root')).unionByName(posts)
def bellman_ford(vertex_ids, edges, sources, max_rounds=1000):
dist = sources.select('id', F.col('dist').cast('double')).localCheckpoint()
frontier = dist
vertices = vertex_ids.select('id')
for _ in range(max_rounds):
g = GraphFrame(vertices.join(frontier.select('id', F.col('dist').alias('d')), 'id', 'left'),
edges.select('src', 'dst', 'weight'))
msgs = g.aggregateMessages(F.min(AM.msg).alias('cand'),
sendToDst=AM.src['d'] + AM.edge['weight'])
improved = (msgs.filter(F.col('cand').isNotNull()).join(dist, 'id', 'left')
.filter(F.col('dist').isNull() | (F.col('cand') < F.col('dist')))
.select('id', F.col('cand').alias('dist'))).localCheckpoint()
if improved.isEmpty():
break
dist = (dist.join(improved.withColumnRenamed('dist', 'new'), 'id', 'full')
.select('id', F.coalesce('new', 'dist').alias('dist'))).localCheckpoint()
frontier = improved
return dist
Select a query below to view its logic and description.
Q1. One-hop friends of a given person
Return the direct KNOWS neighbours of one person. This is the baseline local traversal: a one-edge motif filtered to a single start vertex. GraphFrames has no index, so finding the start person is a filtered scan of the Person vertices.
q1_g = knows_graph('firstName', 'lastName')
q1_result = (
q1_g.find('(a)-[e]-(friend)')
.filter(F.col('a.id') == pid(100316))
.select(F.col('friend.ldbc_id').alias('personId'),
F.col('friend.firstName').alias('firstName'),
F.col('friend.lastName').alias('lastName'))
.orderBy('personId')
.limit(100)
)
q1_result.show(truncate=False)
Q2. Persons within three friendship hops
Return every distinct person reachable within one to three KNOWS hops of a start person. The frontier grows quickly on a social graph, so this measures expansion cost and duplicate elimination across overlapping paths.
q2_g = knows_graph('firstName', 'lastName')
q2_start = pid(100316)
q2_paths = [
q2_g.find('(a)-[e1]-(n)').filter(F.col('a.id') == q2_start),
q2_g.find('(a)-[e1]-(b); (b)-[e2]-(n)').filter(
(F.col('a.id') == q2_start) & (F.col('e1._eid') != F.col('e2._eid'))),
q2_g.find('(a)-[e1]-(b); (b)-[e2]-(c); (c)-[e3]-(n)').filter(
(F.col('a.id') == q2_start) & (F.col('e1._eid') != F.col('e2._eid')) &
(F.col('e1._eid') != F.col('e3._eid')) & (F.col('e2._eid') != F.col('e3._eid'))),
]
q2_result = (
reduce(lambda x, y: x.unionByName(y), [p.select('n') for p in q2_paths])
.filter(F.col('n.id') != q2_start)
.select(F.col('n.ldbc_id').alias('personId'),
F.col('n.firstName').alias('firstName'),
F.col('n.lastName').alias('lastName'))
.distinct()
.orderBy('personId')
.limit(100)
)
q2_result.show(truncate=False)
Q3. Persons exactly three friendship hops away
Return the persons reached by paths of length exactly three, rather than everything within three hops. Only the start person is excluded, so a person who is also reachable in fewer hops still appears if a three-hop path to them exists.
q3_g = knows_graph('firstName', 'lastName')
q3_start = pid(100316)
q3_result = (
q3_g.find('(a)-[e1]-(b); (b)-[e2]-(c); (c)-[e3]-(n)')
.filter((F.col('a.id') == q3_start) &
(F.col('e1._eid') != F.col('e2._eid')) & (F.col('e1._eid') != F.col('e3._eid')) &
(F.col('e2._eid') != F.col('e3._eid')) & (F.col('n.id') != q3_start))
.select(F.col('n.ldbc_id').alias('personId'),
F.col('n.firstName').alias('firstName'),
F.col('n.lastName').alias('lastName'))
.distinct()
.orderBy('personId')
.limit(100)
)
q3_result.show(truncate=False)
Q4. Persons within two hops speaking a given language
Expand one to two KNOWS hops and keep only the persons who speak a given language and were born on or after a given date. The speaks property is an array, so the filter is an array_contains test, applied to the union of the one- and two-hop motif results.
q4_g = knows_graph('firstName', 'lastName', 'speaks', 'birthday')
q4_start = pid(100316)
q4_paths = [
q4_g.find('(a)-[e1]-(n)').filter(F.col('a.id') == q4_start),
q4_g.find('(a)-[e1]-(b); (b)-[e2]-(n)').filter(
(F.col('a.id') == q4_start) & (F.col('e1._eid') != F.col('e2._eid'))),
]
q4_result = (
reduce(lambda x, y: x.unionByName(y), [p.select('n') for p in q4_paths])
.filter((F.col('n.id') != q4_start) &
F.array_contains('n.speaks', 'en') &
(F.col('n.birthday') >= F.lit('1980-01-01').cast('date')))
.select(F.col('n.ldbc_id').alias('personId'),
F.col('n.firstName').alias('firstName'),
F.col('n.lastName').alias('lastName'),
F.col('n.birthday').alias('birthday'))
.distinct()
.orderBy(F.desc('birthday'), 'personId')
.limit(100)
)
q4_result.show(truncate=False)
Q5. Persons speaking a given language who registered in a period
Filter Person vertices on an array property and a creationDate range with no traversal at all. This is the single-node path case: a filter over the vertex DataFrame, measuring predicate evaluation over a scan in isolation.
q5_g = knows_graph('firstName', 'lastName', 'speaks', 'creationDate')
q5_result = (
q5_g.vertices
.filter(F.array_contains('speaks', 'en') &
(F.col('creationDate') >= T('2010-07-01T00:00:00.000+00:00')) &
(F.col('creationDate') <= T('2010-07-08T00:00:00.000+00:00')))
.select(F.col('ldbc_id').alias('personId'), 'firstName', 'lastName', 'creationDate')
.orderBy(F.desc('creationDate'), 'personId')
.limit(100)
)
q5_result.show(truncate=False)
Q6. Friendship edges created within a given window
Filter KNOWS edges by their creationDate and return both endpoints. GraphFrames has no edge index, so this is a scan of the KNOWS edges with a range predicate, followed by the joins to the two endpoint vertices.
q6_g = knows_graph()
q6_result = (
q6_g.find('(p1)-[k]->(p2)')
.filter((F.col('k.creationDate') >= T('2010-07-01T00:00:00.000+00:00')) &
(F.col('k.creationDate') <= T('2010-07-08T00:00:00.000+00:00')))
.select(F.col('p1.ldbc_id').alias('person1Id'),
F.col('p2.ldbc_id').alias('person2Id'),
F.col('k.creationDate').alias('creationDate'))
.orderBy('creationDate', 'person1Id')
.limit(100)
)
q6_result.show(truncate=False)
Q7. Whether two persons are connected within four friendship hops
Return a single boolean saying whether a bounded KNOWS path exists between two persons. bfs runs breadth-first from the first person over the undirected KNOWS graph and stops at the first level that reaches the second, so this measures how early the search can stop rather than full path enumeration.
q7_g = knows_graph(undirected=True)
q7_paths = q7_g.bfs(fromExpr=f"id = '{pid(10006)}'", toExpr=f"id = '{pid(2199023261649)}'",
maxPathLength=4)
q7_result = q7_paths.limit(1).agg((F.count(F.lit(1)) > 0).alias('reachable'))
q7_result.show(truncate=False)
Q8. Shortest friendship path between two persons
Return the shortest KNOWS path between two persons, bounded at six hops, as the list of person ids along it. bfs returns every shortest path at the first level that reaches the target; one of them is kept, as Cypher’s shortestPath returns one. The weighted variants of this problem are Q10, Q11 and Q12.
q8_g = knows_graph(undirected=True)
q8_paths = q8_g.bfs(fromExpr=f"id = '{pid(10006)}'", toExpr=f"id = '{pid(2199023261649)}'",
maxPathLength=6)
q8_nodes = ['from'] + sorted([c for c in q8_paths.columns if re.fullmatch(r'v\d+', c)],
key=lambda c: int(c[1:])) + ['to']
q8_result = (
q8_paths
.select(F.array(*[F.col(f'{c}.ldbc_id') for c in q8_nodes]).alias('personIds'),
F.lit(len(q8_nodes) - 1).alias('hopCount'))
.limit(1)
)
q8_result.show(truncate=False)
Q9. All simple friendship paths between two persons up to length three
Enumerate every path between two persons up to three hops, keeping only simple paths. One motif per length matches the paths, and inequality filters on the intermediate vertices remove those that revisit a person. Unlike reachability or shortest path this materialises every result.
q9_g = knows_graph()
q9_a, q9_b = pid(10006), pid(2199023261649)
q9_len1 = (q9_g.find('(a)-[e1]-(b)')
.filter((F.col('a.id') == q9_a) & (F.col('b.id') == q9_b))
.select(F.array('a.ldbc_id', 'b.ldbc_id').alias('personIds'), F.lit(1).alias('hopCount')))
q9_len2 = (q9_g.find('(a)-[e1]-(x); (x)-[e2]-(b)')
.filter((F.col('a.id') == q9_a) & (F.col('b.id') == q9_b) &
(F.col('x.id') != q9_a) & (F.col('x.id') != q9_b))
.select(F.array('a.ldbc_id', 'x.ldbc_id', 'b.ldbc_id').alias('personIds'),
F.lit(2).alias('hopCount')))
q9_len3 = (q9_g.find('(a)-[e1]-(x); (x)-[e2]-(y); (y)-[e3]-(b)')
.filter((F.col('a.id') == q9_a) & (F.col('b.id') == q9_b) &
(F.col('x.id') != q9_a) & (F.col('x.id') != q9_b) &
(F.col('y.id') != q9_a) & (F.col('y.id') != q9_b) & (F.col('x.id') != F.col('y.id')))
.select(F.array('a.ldbc_id', 'x.ldbc_id', 'y.ldbc_id', 'b.ldbc_id').alias('personIds'),
F.lit(3).alias('hopCount')))
q9_result = (
q9_len1.unionByName(q9_len2).unionByName(q9_len3)
.orderBy('hopCount')
.limit(100)
)
q9_result.show(truncate=False)
Q10. Weighted interaction paths
Find the most trusted path between two persons. Edge weight is derived from how strongly the two interacted inside forums created in a given window (a reply counts 1.0 or 0.5, and the weight is 1/(w+1)). The weights are computed into the edge DataFrame of a query-local GraphFrame, and the cheapest path cost comes from Bellman–Ford relaxation with aggregateMessages, which yields the same minimum as Dijkstra on non-negative weights. A message’s forum is found through its thread root, computed by message_roots().
q10_g = knows_graph(undirected=True)
q10_forums = (V('forum')
.filter((F.col('creationDate') >= T('2010-04-16T00:00:00.000+00:00')) &
(F.col('creationDate') <= T('2010-05-08T00:00:00.000+00:00')))
.select(F.col('id').alias('forum')))
q10_posts = (E('container_of').join(q10_forums, F.col('src') == F.col('forum'))
.select(F.col('dst').alias('root')))
q10_threads = message_roots().join(F.broadcast(q10_posts), 'root')
q10_is_post = V('post').select(F.col('id').alias('p'), F.lit(1.0).alias('post_weight'))
q10_replies = (E('reply_of').select(F.col('src').alias('c'), F.col('dst').alias('p'))
.join(q10_threads.select(F.col('message').alias('c')), 'c')
.join(q10_is_post, 'p', 'left')
.withColumn('w', F.coalesce('post_weight', F.lit(0.5))))
q10_creator = creators()
q10_pairs = (q10_replies
.join(q10_creator.select(F.col('message').alias('c'), F.col('person').alias('pc')), 'c')
.join(q10_creator.select(F.col('message').alias('p'), F.col('person').alias('pp')), 'p')
.select('pc', 'pp', 'w'))
q10_w = (q10_pairs.select(F.col('pc').alias('src'), F.col('pp').alias('dst'), 'w')
.unionByName(q10_pairs.select(F.col('pp').alias('src'), F.col('pc').alias('dst'), 'w'))
.groupBy('src', 'dst').agg(F.sum('w').alias('w')))
q10_edges = (q10_g.edges.join(q10_w, ['src', 'dst'], 'left')
.select('src', 'dst', (1.0 / (F.coalesce('w', F.lit(0.0)) + 1.0)).alias('weight')))
q10_dist = bellman_ford(q10_g.vertices, q10_edges,
spark.createDataFrame([(pid(10006), 0.0)], 'id string, dist double'))
q10_result = (q10_dist.filter(F.col('id') == pid(2199023261649))
.select(F.col('dist').alias('totalCost'))
.unionByName(spark.createDataFrame([(-1.0,)], 'totalCost double'))
.agg(F.max('totalCost').alias('totalCost')))
q10_result.show(truncate=False)
Q11. Interaction path between cities
Find the cheapest weighted path between persons living in two given cities. Edge weight falls as reply interaction rises (round(40-sqrt(numInteractions)), floored at 1), so heavily interacting pairs sit closer together. One multi-source Bellman–Ford run from every person in the first city gives the global minimum; a run back from each target at that minimum then recovers every pair that attains it.
q11_g = knows_graph(undirected=True)
q11_creator = creators()
q11_inter = (E('reply_of').select(F.col('src').alias('c'), F.col('dst').alias('p'))
.join(q11_creator.select(F.col('message').alias('c'), F.col('person').alias('pc')), 'c')
.join(q11_creator.select(F.col('message').alias('p'), F.col('person').alias('pp')), 'p'))
q11_counts = (q11_inter.select(F.col('pc').alias('src'), F.col('pp').alias('dst'))
.unionByName(q11_inter.select(F.col('pp').alias('src'), F.col('pc').alias('dst')))
.groupBy('src', 'dst').agg(F.count(F.lit(1)).alias('n')))
q11_edges = (q11_g.edges.select('src', 'dst').join(q11_counts, ['src', 'dst'])
.select('src', 'dst',
F.greatest(F.round(40 - F.sqrt('n')), F.lit(1.0)).alias('weight'))
.localCheckpoint())
q11_located = E('is_located_in').filter(F.col('src').startswith('person:'))
q11_city1 = q11_located.filter(F.col('dst') == 'place:702').select(F.col('src').alias('id'))
q11_city2 = q11_located.filter(F.col('dst') == 'place:967').select(F.col('src').alias('id'))
q11_from_city1 = bellman_ford(q11_g.vertices, q11_edges, q11_city1.withColumn('dist', F.lit(0.0)))
q11_best = q11_from_city1.join(q11_city2, 'id').agg(F.min('dist').alias('d')).first()['d']
if q11_best is None:
q11_result = spark.createDataFrame([], 'person1Id bigint, person2Id bigint, totalWeight double')
else:
q11_targets = [r['id'] for r in q11_from_city1.join(q11_city2, 'id')
.filter(F.col('dist') == q11_best).collect()]
q11_pairs = []
for t in q11_targets:
back = bellman_ford(q11_g.vertices, q11_edges,
spark.createDataFrame([(t, 0.0)], 'id string, dist double'))
q11_pairs.append(back.join(q11_city1, 'id').filter(F.col('dist') == q11_best)
.select(F.col('id').alias('p1'), F.lit(t).alias('p2'), 'dist'))
q11_ids = V('person').select('id', 'ldbc_id')
q11_result = (
reduce(lambda x, y: x.unionByName(y), q11_pairs)
.join(q11_ids.select(F.col('id').alias('p1'), F.col('ldbc_id').alias('person1Id')), 'p1')
.join(q11_ids.select(F.col('id').alias('p2'), F.col('ldbc_id').alias('person2Id')), 'p2')
.select('person1Id', 'person2Id', F.col('dist').alias('totalWeight'))
.orderBy('person1Id', 'person2Id')
)
q11_result.show(truncate=False)
Q12. Recruitment
From every current employee of a target company, find the cheapest weighted path to a given candidate. Edge weight is the smallest difference in enrolment year between two persons who studied at the same university (min(abs(classYear difference)) + 1). The weighted graph is symmetric, so a single Bellman–Ford run from the candidate gives every employee’s distance.
q12_g = knows_graph(undirected=True)
q12_study = E('study_at').select('src', F.col('dst').alias('uni'), 'classYear')
q12_w = (q12_g.edges.select('src', 'dst')
.join(q12_study.select(F.col('src'), 'uni', F.col('classYear').alias('ya')), 'src')
.join(q12_study.select(F.col('src').alias('dst'), 'uni', F.col('classYear').alias('yb')),
['dst', 'uni'])
.groupBy('src', 'dst').agg((F.min(F.abs(F.col('ya') - F.col('yb'))) + 1).alias('weight'))
.localCheckpoint())
q12_target = pid(13194139540317)
q12_company = V('company').filter(F.col('name') == 'National_Airlines_(5M)').select(F.col('id').alias('dst'))
q12_workers = (E('work_at').join(q12_company, 'dst').select(F.col('src').alias('id')).distinct()
.filter(F.col('id') != q12_target))
q12_dist = bellman_ford(q12_g.vertices, q12_w,
spark.createDataFrame([(q12_target, 0.0)], 'id string, dist double'))
q12_reached = q12_dist.join(q12_workers, 'id')
q12_result = (
q12_reached
.join(q12_reached.agg(F.min('dist').alias('dist')), 'dist')
.join(V('person').select('id', 'ldbc_id'), 'id')
.select(F.col('ldbc_id').alias('person1Id'), F.col('dist').alias('totalWeight'))
.orderBy('person1Id')
.limit(20)
)
q12_result.show(truncate=False)
Q13. Friend triangles
Count friendship triangles inside one country where all three KNOWS edges were created within a given window. The a.id < b.id < c.id ordering deduplicates, so each triangle is counted once rather than six times.
q13_g = GraphFrame(
persons_in_country('India').select(F.col('person').alias('id')).distinct(),
E('knows').filter((F.col('creationDate') >= T('2012-09-28T00:00:00.000+00:00')) &
(F.col('creationDate') <= T('2013-01-10T00:00:00.000+00:00'))).select('src', 'dst'))
q13_g = GraphFrame(q13_g.vertices,
q13_g.edges.join(q13_g.vertices.select(F.col('id').alias('src')), 'src')
.join(q13_g.vertices.select(F.col('id').alias('dst')), 'dst'))
q13_result = (q13_g.triangleCount(storage_level=StorageLevel.MEMORY_AND_DISK)
.agg((F.sum('count') / 3).cast('long').alias('count')))
q13_result.show(truncate=False)
Q14. Friendship squares around a given person
Find four-node cycles through a given person: the person, two of their friends, and a fourth person who knows both. All six inequality constraints are stated explicitly so that degenerate matches collapsing two roles onto one person are excluded.
q14_g = knows_graph()
q14_result = (
q14_g.find('(a)-[e1]-(b); (b)-[e2]-(c); (c)-[e3]-(d); (d)-[e4]-(a)')
.filter((F.col('a.id') == pid(100316)) &
(F.col('a.id') != F.col('b.id')) & (F.col('a.id') != F.col('c.id')) &
(F.col('a.id') != F.col('d.id')) & (F.col('b.id') != F.col('c.id')) &
(F.col('b.id') != F.col('d.id')) & (F.col('c.id') != F.col('d.id')))
.select(F.col('b.ldbc_id').alias('firstFriendId'),
F.col('c.ldbc_id').alias('bridgeFriendId'),
F.col('d.ldbc_id').alias('secondFriendId'))
.distinct()
.limit(100)
)
q14_result.show(truncate=False)
Q15. Forums with the most members among those created after a date
Find hub forums: those created after a given date whose membership exceeds a threshold, ranked by member count. HAS_MEMBER is the largest relationship type in the graph at 303.8 million edges and forum membership is heavily skewed, so this is the star pattern under a realistic degree distribution.
q15_g = GraphFrame(V('forum', 'person').select('id', 'type', 'ldbc_id', 'title', 'creationDate'),
E('has_member').select('src', 'dst'))
q15_result = (
q15_g.find('(forum)-[m]->(person)')
.filter((F.col('forum.type') == 'forum') &
(F.col('forum.creationDate') > T('2012-06-01T00:00:00.000+00:00')))
.groupBy(F.col('forum.ldbc_id').alias('forumId'), F.col('forum.title').alias('title'))
.agg(F.countDistinct('person.id').alias('memberCount'))
.filter(F.col('memberCount') >= 100)
.orderBy(F.desc('memberCount'), 'forumId')
.limit(100)
)
q15_result.show(truncate=False)
Q16. Groups of four mutually acquainted persons around a person
Find 4-cliques containing a given person, where all six pairs among the four are connected by KNOWS. The b.id < c.id < d.id ordering reports each clique once instead of six times. It is the densest pattern in the set and the most demanding on join ordering.
q16_g = knows_graph()
q16_result = (
q16_g.find('(a)-[e1]-(b); (a)-[e2]-(c); (a)-[e3]-(d); (b)-[e4]-(c); (b)-[e5]-(d); (c)-[e6]-(d)')
.filter((F.col('a.id') == pid(100316)) &
(F.col('b.ldbc_id') < F.col('c.ldbc_id')) & (F.col('c.ldbc_id') < F.col('d.ldbc_id')))
.select(F.col('b.ldbc_id').alias('person2Id'),
F.col('c.ldbc_id').alias('person3Id'),
F.col('d.ldbc_id').alias('person4Id'))
.limit(100)
)
q16_result.show(truncate=False)
Q17. Friend recommendation
For two persons who share an interest in a tag but are not yet connected by KNOWS, count their mutual friends and rank by that count. This is the common-neighbour join T3.5 describes, combined with a negative pattern (NOT (person1)-[:KNOWS]-(person2)).
q17_g = GraphFrame(persons(), E('knows').select('src', 'dst', '_eid'))
q17_interested = (E('has_interest').join(tag_id('Thomas_More'), F.col('dst') == F.col('tag'))
.select(F.col('src').alias('id'), F.lit(True).alias('interested')).distinct())
q17_g = GraphFrame(q17_g.vertices.join(q17_interested, 'id', 'left'), q17_g.edges)
q17_pairs = (
q17_g.find('(p1)-[e1]-(m); (m)-[e2]-(p2)')
.filter(F.col('p1.interested') & F.col('p2.interested') & (F.col('p1.id') != F.col('p2.id')))
.select(F.col('p1.id').alias('p1'), F.col('p2.id').alias('p2'),
F.col('p1.ldbc_id').alias('person1Id'), F.col('p2.ldbc_id').alias('person2Id'),
F.col('m.id').alias('mutual'))
)
q17_knows = (q17_g.edges.select(F.col('src').alias('p1'), F.col('dst').alias('p2'))
.unionByName(q17_g.edges.select(F.col('dst').alias('p1'), F.col('src').alias('p2'))))
q17_result = (
q17_pairs.join(q17_knows, ['p1', 'p2'], 'left_anti')
.groupBy('person1Id', 'person2Id')
.agg(F.countDistinct('mutual').alias('mutualFriendCount'))
.orderBy(F.desc('mutualFriendCount'), 'person1Id', 'person2Id')
.limit(20)
)
q17_result.show(truncate=False)
Q18. Popular topics in a country
A long chain running Country → City → Person → Forum → Post → Message → Tag → TagClass, with a variable-length REPLY_OF*0.. segment walking the reply tree in the middle, counting the messages per forum that hit a given TagClass.
The REPLY_OF*0.. segment, from a message to the post at the root of its thread, comes from message_roots() in the shared helpers.
q18_g = GraphFrame(V('forum', 'person').select('id', 'type', 'ldbc_id', 'title', 'creationDate'),
E('has_moderator').select('src', 'dst'))
q18_forums = (
q18_g.find('(forum)-[hm]->(person)')
.filter(F.col('forum.type') == 'forum')
.select(F.col('forum.id').alias('forum'), F.col('person.id').alias('person'),
F.col('forum.ldbc_id').alias('forum.id'), F.col('forum.title').alias('forum.title'),
F.col('forum.creationDate').alias('forum.creationDate'), F.col('person.ldbc_id').alias('person.id'))
.join(persons_in_country('India').select('person'), 'person')
)
q18_posts = E('container_of').join(q18_forums.select('forum'), F.col('src') == F.col('forum')).select(
'forum', F.col('dst').alias('root'))
q18_tagclass = V('tagclass').filter(F.col('name') == 'Saint').select(F.col('id').alias('dst'))
q18_tags = E('has_type').join(q18_tagclass, 'dst').select(F.col('src').alias('tag'))
q18_messages = (E('has_tag').filter(F.col('src').startswith('message:'))
.join(F.broadcast(q18_tags), F.col('dst') == F.col('tag'))
.select(F.col('src').alias('message')).distinct())
q18_result = (
message_roots().join(q18_messages, 'message').join(q18_posts, 'root')
.groupBy('forum').agg(F.countDistinct('message').alias('messageCount'))
.join(q18_forums, 'forum')
.select('`forum.id`', '`forum.title`', '`forum.creationDate`', '`person.id`', 'messageCount')
.orderBy(F.desc('messageCount'), F.col('`forum.id`'))
.limit(20)
)
q18_result.show(truncate=False)
Q19. Related Topics
Follow a three-hop line from a tag through its messages and their replies to the other tags those replies carry, excluding replies that are themselves tagged with the starting tag. The exclusion is a negative pattern over the same three-hop shape.
q19_g = GraphFrame(V('comment', 'post', 'tag').select('id', 'type', 'name'),
E('has_tag', 'reply_of').filter(F.col('src').startswith('message:'))
.select('src', 'dst', 'etype'))
q19_tag = 'tag:' + str(V('tag').filter(F.col('name') == 'Jacques_Chirac').first()['ldbc_id'])
q19_matches = (
q19_g.find('(m)-[ht]->(t); (c)-[r]->(m); (c)-[ht2]->(rt)')
.filter((F.col('t.id') == q19_tag) & (F.col('ht.etype') == 'has_tag') &
(F.col('r.etype') == 'reply_of') & (F.col('c.type') == 'comment') &
(F.col('ht2.etype') == 'has_tag') & (F.col('rt.type') == 'tag'))
.select(F.col('c.id').alias('comment'), F.col('rt.name').alias('relatedTag.name'))
)
q19_tagged = E('has_tag').filter(F.col('dst') == q19_tag).select(F.col('src').alias('comment'))
q19_result = (
q19_matches.join(q19_tagged, 'comment', 'left_anti')
.groupBy('`relatedTag.name`').agg(F.countDistinct('comment').alias('count'))
.orderBy(F.desc('count'), F.col('`relatedTag.name`'))
.limit(100)
)
q19_result.show(truncate=False)
Q20. International dialog
For friend pairs split across two countries, score their interaction: a reply scores 4 or 1 and a like scores 10 or 1. Each of the four cases is a line running person1 → message → message → person2 or person1 → message → person2; each is computed once as a set of (person, person) pairs and joined onto the friend pairs. Only the highest-scoring pair per city is kept, using a window ranking.
q20_g = GraphFrame(persons(), E('knows').select('src', 'dst'))
q20_p1 = persons_in_country('India').select(F.col('person').alias('p1'), 'city', 'city_name')
q20_p2 = persons_in_country('Netherlands').select(F.col('person').alias('p2'))
q20_pairs = (q20_g.find('(a)-[k]-(b)')
.select(F.col('a.id').alias('p1'), F.col('b.id').alias('p2'))
.join(q20_p1, 'p1').join(q20_p2, 'p2').distinct())
q20_people = q20_p1.select(F.col('p1').alias('person')).unionByName(
q20_p2.select(F.col('p2').alias('person'))).distinct()
q20_creator = creators().join(F.broadcast(q20_people), 'person')
q20_reply = (E('reply_of').select(F.col('src').alias('c'), F.col('dst').alias('m'))
.join(q20_creator.select(F.col('message').alias('c'), F.col('person').alias('pc')), 'c')
.join(q20_creator.select(F.col('message').alias('m'), F.col('person').alias('pm')), 'm')
.select('pc', 'pm').distinct())
q20_likes = (E('likes').select(F.col('src').alias('liker'), F.col('dst').alias('m'))
.join(F.broadcast(q20_people.select(F.col('person').alias('liker'))), 'liker')
.join(q20_creator.select(F.col('message').alias('m'), F.col('person').alias('pm')), 'm')
.select('liker', 'pm').distinct())
def q20_flag(df, a, b, name, points):
return df.select(F.col(a).alias('p1'), F.col(b).alias('p2'), F.lit(points).alias(name))
q20_scored = (
q20_pairs
.join(q20_flag(q20_reply, 'pc', 'pm', 's1', 4), ['p1', 'p2'], 'left')
.join(q20_flag(q20_reply, 'pm', 'pc', 's2', 1), ['p1', 'p2'], 'left')
.join(q20_flag(q20_likes, 'liker', 'pm', 's3', 10), ['p1', 'p2'], 'left')
.join(q20_flag(q20_likes, 'pm', 'liker', 's4', 1), ['p1', 'p2'], 'left')
.select('p1', 'p2', 'city', 'city_name',
sum(F.coalesce(c, F.lit(0)) for c in ('s1', 's2', 's3', 's4')).alias('score'))
.join(V('person').select(F.col('id').alias('p1'), F.col('ldbc_id').alias('person1Id')), 'p1')
.join(V('person').select(F.col('id').alias('p2'), F.col('ldbc_id').alias('person2Id')), 'p2')
)
q20_rank = Window.partitionBy('city').orderBy(F.desc('score'), 'person1Id', 'person2Id')
q20_result = (
q20_scored.withColumn('rk', F.row_number().over(q20_rank)).filter(F.col('rk') == 1)
.select(F.col('person1Id').alias('top.person1Id'), F.col('person2Id').alias('top.person2Id'),
F.col('city_name').alias('city1.name'), F.col('score').alias('top.score'))
.orderBy(F.desc('`top.score`'), F.col('`top.person1Id`'), F.col('`top.person2Id`'))
.limit(100)
)
q20_result.show(truncate=False)
Q21. Most active Posters of a given Topic
For each person posting about a tag, aggregate three per-node degrees — messages created, replies received and likes received — with the latter two counted per message and left-joined, then group by person and rank by 1×messages + 2×replies + 10×likes.
q21_g = GraphFrame(V('comment', 'post', 'person', 'tag').select('id', 'type', 'ldbc_id', 'name'),
E('has_tag', 'has_creator').select('src', 'dst', 'etype'))
q21_messages = (
q21_g.find('(m)-[ht]->(t); (m)-[hc]->(p)')
.filter((F.col('t.type') == 'tag') & (F.col('t.name') == 'Republic_of_New_Granada') &
(F.col('ht.etype') == 'has_tag') & (F.col('hc.etype') == 'has_creator'))
.select(F.col('m.id').alias('message'), F.col('p.ldbc_id').alias('person.id'))
)
q21_likes = E('likes').groupBy(F.col('dst').alias('message')).agg(F.count(F.lit(1)).alias('likeCount'))
q21_replies = E('reply_of').groupBy(F.col('dst').alias('message')).agg(F.count(F.lit(1)).alias('replyCount'))
q21_result = (
q21_messages.join(q21_likes, 'message', 'left').join(q21_replies, 'message', 'left')
.groupBy('`person.id`')
.agg(F.sum(F.coalesce('replyCount', F.lit(0))).alias('replyCount'),
F.sum(F.coalesce('likeCount', F.lit(0))).alias('likeCount'),
F.count('message').alias('messageCount'))
.withColumn('score', F.col('messageCount') + 2 * F.col('replyCount') + 10 * F.col('likeCount'))
.orderBy(F.desc('score'), F.col('`person.id`'))
.limit(100)
)
q21_result.show(truncate=False)
Q22. How many persons have a given number of posts
Compute the distribution of how many persons created N qualifying messages, where a message must have content, be shorter than a threshold, be newer than a start date, and belong to a post in one of the given languages. This is a per-person degree histogram over HAS_CREATOR.
The REPLY_OF*0.. segment, from a message to the post at the root of its thread, comes from message_roots() in the shared helpers.
q22_g = GraphFrame(V('person').select('id'), E('has_creator').select('src', 'dst'))
q22_posts = (V('post').filter(F.col('language').isin('es', 'pt', 'zh'))
.select(F.col('id').alias('root')))
q22_messages = (V('comment', 'post')
.filter(F.col('content').isNotNull() & (F.col('length') < 135) &
(F.col('creationDate') > T('2012-05-25T00:00:00.000+00:00')))
.select(F.col('id').alias('message')))
q22_counts = (message_roots().join(q22_messages, 'message').join(q22_posts, 'root')
.join(creators(), 'message')
.groupBy('person').agg(F.count(F.lit(1)).alias('messageCount')))
q22_result = (
q22_g.vertices.join(q22_counts.withColumnRenamed('person', 'id'), 'id', 'left')
.select(F.coalesce('messageCount', F.lit(0)).alias('messageCount'))
.groupBy('messageCount').agg(F.count(F.lit(1)).alias('personCount'))
.orderBy(F.desc('personCount'), F.desc('messageCount'))
)
q22_result.show(truncate=False)
Q23. Enrolment-year statistics over all study relationships
Count every STUDY_AT relationship and summarise its numeric classYear property with a global average, minimum and maximum. No grouping and no node properties are involved, so it isolates relationship scanning and aggregation.
q23_g = GraphFrame(V('person', 'university').select('id', 'type'),
E('study_at').select('src', 'dst', 'classYear'))
q23_result = (
q23_g.find('(p)-[s]->(u)')
.filter(F.col('u.type') == 'university')
.agg(F.count('s').alias('edgeCount'), F.avg('s.classYear').alias('avgClassYear'),
F.min('s.classYear').alias('minClassYear'), F.max('s.classYear').alias('maxClassYear'))
)
q23_result.show(truncate=False)
Q24. Enrolment-year statistics by university and gender
The grouped counterpart of the previous query: the same classYear statistics, grouped by a property of the target node (the university name) and one of the source node (the person’s gender), ordered by edge count. It measures multidimensional grouping and top-K materialisation.
q24_g = GraphFrame(V('person', 'university').select('id', 'type', 'name', 'gender'),
E('study_at').select('src', 'dst', 'classYear'))
q24_result = (
q24_g.find('(p)-[s]->(u)')
.filter((F.col('p.type') == 'person') & (F.col('u.type') == 'university'))
.groupBy(F.col('u.name').alias('universityName'), F.col('p.gender').alias('gender'))
.agg(F.count('s').alias('edgeCount'), F.avg('s.classYear').alias('avgClassYear'),
F.min('s.classYear').alias('minClassYear'), F.max('s.classYear').alias('maxClassYear'))
.orderBy(F.desc('edgeCount'), 'universityName', 'gender')
.limit(100)
)
q24_result.show(truncate=False)
Q25. Friendship path counts by length and gender reached
Enumerate KNOWS paths up to three hops from one person and report, for each combination of path length and the gender of the person reached, how many paths there are and how many distinct persons they reach. Grouping by an endpoint property on top of path length is what separates this from a plain path-count query.
q25_g = knows_graph('gender')
q25_start = pid(100316)
q25_paths = [
q25_g.find('(a)-[e1]-(n)').filter(F.col('a.id') == q25_start)
.select(F.lit(1).alias('pathLength'), 'n'),
q25_g.find('(a)-[e1]-(b); (b)-[e2]-(n)')
.filter((F.col('a.id') == q25_start) & (F.col('e1._eid') != F.col('e2._eid')))
.select(F.lit(2).alias('pathLength'), 'n'),
q25_g.find('(a)-[e1]-(b); (b)-[e2]-(c); (c)-[e3]-(n)')
.filter((F.col('a.id') == q25_start) & (F.col('e1._eid') != F.col('e2._eid')) &
(F.col('e1._eid') != F.col('e3._eid')) & (F.col('e2._eid') != F.col('e3._eid')))
.select(F.lit(3).alias('pathLength'), 'n'),
]
q25_result = (
reduce(lambda x, y: x.unionByName(y), q25_paths)
.groupBy('pathLength', F.col('n.gender').alias('gender'))
.agg(F.count(F.lit(1)).alias('pathCount'), F.countDistinct('n.id').alias('distinctPersonCount'))
.orderBy('pathLength', F.desc('pathCount'))
.limit(100)
)
q25_result.show(truncate=False)
Q26. Persons interested in both of two given tags
Return the persons who have a HAS_INTEREST edge to each of two named tags. A single motif with two HAS_INTEREST edges from the same person expresses the intersection.
q26_g = GraphFrame(V('person', 'tag').select('id', 'type', 'ldbc_id', 'name', 'firstName', 'lastName'),
E('has_interest').select('src', 'dst'))
q26_result = (
q26_g.find('(p)-[i1]->(ta); (p)-[i2]->(tb)')
.filter((F.col('ta.type') == 'tag') & (F.col('ta.name') == 'Thomas_More') &
(F.col('tb.type') == 'tag') & (F.col('tb.name') == 'Michelangelo'))
.select(F.col('p.ldbc_id').alias('personId'), F.col('p.firstName').alias('firstName'),
F.col('p.lastName').alias('lastName'))
.distinct()
.orderBy('personId')
.limit(100)
)
q26_result.show(truncate=False)
Q27. Degree centrality over the friendship graph
Rank persons by how many friends they have on the undirected KNOWS graph. It is the cheapest of the analytics queries — a single pass over the adjacency structure with no iteration — and gives a baseline against which PageRank’s cost can be read.
q27_g = knows_graph('firstName', 'lastName')
q27_result = (
q27_g.degrees.join(q27_g.vertices, 'id')
.select(F.col('ldbc_id').alias('personId'), 'firstName', 'lastName',
F.col('degree').cast('long').alias('degree'))
.orderBy(F.desc('degree'))
.limit(10)
)
q27_result.show(truncate=False)
Q28. PageRank over the friendship graph
Rank persons by PageRank on the undirected KNOWS graph to find the most structurally influential accounts. The algorithm iterates until the scores converge or the iteration limit is reached.
q28_g = knows_graph('firstName', 'lastName', undirected=True)
q28_result = (
q28_g.pageRank(resetProbability=0.15, maxIter=20).vertices
.select(F.col('ldbc_id').alias('personId'), 'firstName', 'lastName', F.col('pagerank'))
.orderBy(F.desc('pagerank'))
.limit(10)
)
q28_result.show(truncate=False)
Q29. Weakly connected components of the friendship graph
Find the connected regions of the undirected KNOWS graph and report the largest by size. On a generated social network almost every person falls in one giant component, so the result also shows how many isolated accounts the generator produced.
q29_g = knows_graph()
q29_result = (
q29_g.connectedComponents()
.groupBy(F.col('component').alias('componentId')).agg(F.count(F.lit(1)).alias('componentSize'))
.orderBy(F.desc('componentSize'))
.limit(10)
)
q29_result.show(truncate=False)
Q30. Strongly connected components of the directed friendship graph
Find the maximal sets of persons that can all reach each other following KNOWS edges in their stored direction, with stronglyConnectedComponents on the directed KNOWS graph. SNB stores each friendship once, as a single directed edge, so mutual reachability is rare and the components are expected to be mostly singletons.
q30_g = knows_graph()
q30_result = (
q30_g.stronglyConnectedComponents(maxIter=1000)
.groupBy(F.col('component').alias('componentId')).agg(F.count(F.lit(1)).alias('componentSize'))
.orderBy(F.desc('componentSize'))
.limit(10)
)
q30_result.show(truncate=False)
stronglyConnectedComponents (GraphX), cannot process this graph: every KNOWS edge points from the smaller to the larger person id, so the directed KNOWS graph is a deep acyclic graph, and GraphX trims it one layer per pass in a loop that is not checkpointed and that maxIter does not bound. After about 50 minutes and 100,000 stages, tasks failed with StackOverflowError and the driver ran out of memory, which stopped the Spark session. The query is therefore recorded as not supported. Because the graph is acyclic, the correct answer is 473,001 single-person components.Q31. Louvain communities of the friendship graph
Partition the undirected KNOWS graph into densely connected communities with Louvain and report the largest. The generator correlates friendship with location and interests, so the communities it finds are not arbitrary.
q31_g = knows_graph()
q31_result = (
q31_g.labelPropagation(maxIter=10)
.groupBy(F.col('label').alias('communityId')).agg(F.count(F.lit(1)).alias('communitySize'))
.orderBy(F.desc('communitySize'))
.limit(10)
)
q31_result.show(truncate=False)
Q32. Top thread initiators
Count the posts each person started within a window, together with the messages in the whole reply tree beneath them (REPLY_OF*0..) that also fall in the window. It combines two workloads: the bounded-length path enumeration of T4.5, and the ungrouped per-node count of T4.1 applied to the thread initiator, since the result is a count of distinct posts and distinct replies per person rather than a breakdown by path length.
The REPLY_OF*0.. segment, from a message to the post at the root of its thread, comes from message_roots() in the shared helpers.
q32_g = GraphFrame(V('post', 'person').select('id', 'type', 'ldbc_id', 'firstName', 'lastName', 'creationDate'),
E('has_creator').select('src', 'dst'))
q32_start, q32_end = T('2012-09-05T00:00:00.000+00:00'), T('2012-12-04T00:00:00.000+00:00')
q32_posts = (q32_g.find('(post)-[hc]->(person)')
.filter((F.col('post.type') == 'post') &
(F.col('post.creationDate') >= q32_start) & (F.col('post.creationDate') <= q32_end))
.select(F.col('post.id').alias('root'), F.col('person.ldbc_id').alias('person.id'),
F.col('person.firstName').alias('person.firstName'),
F.col('person.lastName').alias('person.lastName')))
q32_replies = (V('comment', 'post')
.filter((F.col('creationDate') >= q32_start) & (F.col('creationDate') <= q32_end))
.select(F.col('id').alias('message')))
q32_result = (
message_roots().join(q32_replies, 'message').join(q32_posts, 'root')
.groupBy('`person.id`', '`person.firstName`', '`person.lastName`')
.agg(F.countDistinct('root').alias('threadCount'), F.countDistinct('message').alias('messageCount'))
.orderBy(F.desc('messageCount'), F.col('`person.id`'))
.limit(100)
)
q32_result.show(truncate=False)
Q33. Posting summary
Build a histogram over every message created before a given instant, grouped by year, by whether the message is a comment, and by length bucket (<40, <80, <160, rest), reporting count, average length, total length and share of all messages. It does not match a defined type because it summarises a node property (length) with no traversal at all, whereas the aggregation types cover degree statistics, edge-property statistics and path counts.
q33_g = GraphFrame(V('comment', 'post').select('id', 'type', 'creationDate', 'content', 'length'),
E('reply_of').select('src', 'dst'))
q33_cutoff = T('2010-08-01T12:12:48.387+00:00')
q33_before = q33_g.vertices.filter(F.col('creationDate') < q33_cutoff)
q33_total = q33_before.agg(F.count(F.lit(1)).cast('double').alias('totalMessageCount'))
q33_result = (
q33_before.filter(F.col('content').isNotNull())
.select(F.year('creationDate').alias('year'), (F.col('type') == 'comment').alias('isComment'),
F.when(F.col('length') < 40, 0).when(F.col('length') < 80, 1)
.when(F.col('length') < 160, 2).otherwise(3).alias('lengthCategory'),
'length')
.groupBy('year', 'isComment', 'lengthCategory')
.agg(F.count(F.lit(1)).alias('messageCount'),
(F.sum('length') / F.count(F.lit(1))).alias('averageMessageLength'),
F.sum('length').alias('sumMessageLength'))
.crossJoin(q33_total)
.withColumn('percentageOfMessages', F.col('messageCount') / F.col('totalMessageCount'))
.drop('totalMessageCount')
.orderBy(F.desc('year'), 'isComment', 'lengthCategory')
)
q33_result.show(truncate=False)
Q34. Tag evolution
For every tag of a given TagClass, compare how many messages referenced it in two adjacent 100-day windows and rank by the difference. The distinguishing work is the comparison of two independent time windows and the arithmetic between them, which no defined type covers.
q34_g = GraphFrame(V('tag', 'tagclass').select('id', 'type', 'name'), E('has_type').select('src', 'dst'))
q34_d0 = T('2010-06-19T00:00:00.000+00:00')
q34_d1 = q34_d0 + F.expr('INTERVAL 100 DAYS')
q34_d2 = q34_d0 + F.expr('INTERVAL 200 DAYS')
q34_tags = (q34_g.find('(tag)-[ht]->(tc)')
.filter((F.col('tc.type') == 'tagclass') & (F.col('tc.name') == 'SoccerPlayer'))
.select(F.col('tag.id').alias('tag'), F.col('tag.name').alias('tag.name')))
q34_hits = (E('has_tag').filter(F.col('src').startswith('message:')).select('src', 'dst')
.join(F.broadcast(q34_tags.select('tag')), F.col('dst') == F.col('tag'))
.join(V('comment', 'post').select(F.col('id').alias('src'), 'creationDate'), 'src')
.groupBy('tag')
.agg(F.sum(((F.col('creationDate') >= q34_d0) & (F.col('creationDate') < q34_d1)).cast('long'))
.alias('countWindow1'),
F.sum(((F.col('creationDate') >= q34_d1) & (F.col('creationDate') < q34_d2)).cast('long'))
.alias('countWindow2')))
q34_result = (
q34_tags.join(q34_hits, 'tag', 'left')
.select('`tag.name`', F.coalesce('countWindow1', F.lit(0)).alias('countWindow1'),
F.coalesce('countWindow2', F.lit(0)).alias('countWindow2'))
.withColumn('diff', F.abs(F.col('countWindow1') - F.col('countWindow2')))
.orderBy(F.desc('diff'), F.col('`tag.name`'))
.limit(100)
)
q34_result.show(truncate=False)
Q35. Top message creators in a country
First pick the 100 forums with the most members, then aggregate the message output of those forums’ members, using UNION ALL to keep members with zero messages in the result. It is a two-phase query whose second phase consumes the top-K output of the first, so neither phase alone characterises it.
The REPLY_OF*0.. segment, from a message to the post at the root of its thread, comes from message_roots() in the shared helpers.
q35_g = GraphFrame(V('forum', 'person', 'city', 'country').select('id', 'type', 'ldbc_id', 'creationDate'),
E('has_member', 'is_located_in', 'is_part_of')
.filter(~F.col('src').startswith('message:') & ~F.col('src').startswith('organisation:'))
.select('src', 'dst', 'etype'))
q35_counts = (
q35_g.find('(forum)-[m]->(person); (person)-[l]->(city); (city)-[p]->(country)')
.filter((F.col('forum.type') == 'forum') & (F.col('m.etype') == 'has_member') &
(F.col('l.etype') == 'is_located_in') & (F.col('p.etype') == 'is_part_of') &
(F.col('forum.creationDate') > T('2010-01-29T00:00:00.000+00:00')))
.groupBy(F.col('forum.id').alias('forum'), F.col('forum.ldbc_id').alias('forumId'),
F.col('country.ldbc_id').alias('countryId'))
.agg(F.count(F.lit(1)).alias('numberOfMembers'))
)
q35_top = (q35_counts.groupBy('forum', 'forumId').agg(F.max('numberOfMembers').alias('best'))
.orderBy(F.desc('best'), 'forumId').limit(100).select('forum').localCheckpoint())
q35_members = (E('has_member').join(F.broadcast(q35_top), F.col('src') == F.col('forum'))
.select(F.col('dst').alias('person')).distinct())
q35_posts = (E('container_of').join(F.broadcast(q35_top), F.col('src') == F.col('forum'))
.select(F.col('dst').alias('root')))
q35_counted = (message_roots().join(q35_posts, 'root').join(creators(), 'message')
.join(q35_members, 'person')
.groupBy('person').agg(F.countDistinct('message').alias('messageCount')))
q35_result = (
q35_members.join(q35_counted, 'person', 'left')
.join(V('person').select(F.col('id').alias('person'), 'ldbc_id', 'firstName', 'lastName', 'creationDate'),
'person')
.select(F.col('ldbc_id').alias('personId'), F.col('firstName').alias('personFirstName'),
F.col('lastName').alias('personLastName'), F.col('creationDate').alias('personCreationDate'),
F.coalesce('messageCount', F.lit(0)).alias('messageCount'))
.orderBy(F.desc('messageCount'), 'personId')
.limit(100)
)
q35_result.show(truncate=False)
Q36. Most authoritative users on a given topic
A two-level authority score: person1’s score comes from the person2 who liked their message, and person2’s weight in turn depends on how many likes person2’s own messages received. count(DISTINCT like) ensures each person2’s popularity is added only once per person1. It propagates score through the graph like PageRank but at a fixed depth of two rather than iterating to convergence, so it is neither a degree statistic nor an iterative algorithm.
q36_g = GraphFrame(V('comment', 'post', 'person', 'tag').select('id', 'type', 'ldbc_id', 'name'),
E('has_tag', 'has_creator').select('src', 'dst', 'etype'))
q36_person1 = (
q36_g.find('(m1)-[ht]->(t); (m1)-[hc]->(p1)')
.filter((F.col('t.type') == 'tag') & (F.col('t.name') == 'All_by_Myself') &
(F.col('ht.etype') == 'has_tag') & (F.col('hc.etype') == 'has_creator'))
.select(F.col('m1.id').alias('message1'), F.col('p1.id').alias('person1'),
F.col('p1.ldbc_id').alias('person1.id'))
)
q36_likes = E('likes').select(F.col('src').alias('liker'), F.col('dst').alias('message'))
q36_person2 = (q36_likes.join(q36_person1.select(F.col('message1').alias('message'), 'person1'), 'message')
.select('person1', F.col('liker').alias('person2')).distinct())
q36_popularity = (q36_likes.join(creators(), 'message')
.join(q36_person2.select(F.col('person2').alias('person')).distinct(), 'person')
.groupBy(F.col('person').alias('person2')).agg(F.count(F.lit(1)).alias('popularity')))
q36_result = (
q36_person1.select('person1', '`person1.id`').distinct()
.join(q36_person2.join(q36_popularity, 'person2').groupBy('person1')
.agg(F.sum('popularity').alias('authorityScore')), 'person1', 'left')
.select('`person1.id`', F.coalesce('authorityScore', F.lit(0)).alias('authorityScore'))
.orderBy(F.desc('authorityScore'), F.col('`person1.id`'))
.limit(100)
)
q36_result.show(truncate=False)
Q37. Central Person for a Tag
Score the followers and posters of a tag (100 points for an interest edge, 1 point per message in the window), then add up the same score across each person’s friends and return the highest combined total. The score is a weighted mix of two edge types that is then aggregated over a one-hop neighbourhood, which does not reduce to any single defined type.
q37_g = GraphFrame(persons(), E('knows').select('src', 'dst'))
q37_tag = tag_id('Vladimir_Putin')
q37_interest = (E('has_interest').join(q37_tag, F.col('dst') == F.col('tag'))
.select(F.col('src').alias('person'), F.lit(100).alias('points')))
q37_posts = (E('has_tag').join(q37_tag, F.col('dst') == F.col('tag')).select(F.col('src').alias('message'))
.join(V('comment', 'post')
.filter((F.col('creationDate') > T('2010-06-12T00:00:00.000+00:00')) &
(F.col('creationDate') < T('2010-06-23T00:00:00.000+00:00')))
.select(F.col('id').alias('message')), 'message')
.join(creators(), 'message').select('person', F.lit(1).alias('points')))
q37_score = (q37_interest.unionByName(q37_posts).groupBy('person').agg(F.sum('points').alias('score'))
.localCheckpoint())
q37_g = GraphFrame(q37_g.vertices.join(q37_score.withColumnRenamed('person', 'id'), 'id', 'left')
.withColumn('score', F.coalesce('score', F.lit(0))), q37_g.edges)
q37_friends = q37_g.aggregateMessages(F.sum(AM.msg).alias('friendsScore'),
sendToSrc=AM.dst['score'], sendToDst=AM.src['score'])
q37_result = (
q37_g.vertices.join(q37_score.withColumnRenamed('person', 'id').select('id'), 'id')
.join(q37_friends, 'id', 'left')
.select(F.col('ldbc_id').alias('person.id'), 'score',
F.coalesce('friendsScore', F.lit(0)).alias('friendsScore'))
.orderBy(F.desc(F.col('score') + F.col('friendsScore')), F.col('`person.id`'))
.limit(100)
)
q37_result.show(truncate=False)
Q38. Experts in social circle
Among the persons whose KNOWS distance from a start person falls in the ring [minPathDistance, maxPathDistance], find those who live in a given country and have posted about a given TagClass. shortestPaths with the start person as the only landmark gives every hop distance in one pass, and the ring is the persons at distance 3 or 4. The ring is then joined against a location pattern and a tag pattern.
q38_g = knows_graph(undirected=True)
q38_start = pid(100316)
q38_dist = q38_g.shortestPaths(landmarks=[q38_start]).select(
'id', F.col('distances').getItem(q38_start).alias('d'))
q38_candidates = (q38_dist.filter(F.col('d').between(3, 4)).select(F.col('id').alias('person'))
.join(persons_in_country('Australia').select('person'), 'person'))
q38_tagclass = V('tagclass').filter(F.col('name') == 'NascarDriver').select(F.col('id').alias('dst'))
q38_class_tags = E('has_type').join(q38_tagclass, 'dst').select(F.col('src').alias('tag'))
q38_hastag = E('has_tag').filter(F.col('src').startswith('message:')).select(
F.col('src').alias('message'), F.col('dst').alias('tag'))
q38_messages = (creators().join(F.broadcast(q38_candidates), 'person')
.join(q38_hastag.join(F.broadcast(q38_class_tags), 'tag').select('message').distinct(),
'message'))
q38_result = (
q38_messages.join(q38_hastag, 'message')
.join(V('tag').select(F.col('id').alias('tag'), F.col('name').alias('tag.name')), 'tag')
.join(V('person').select(F.col('id').alias('person'), F.col('ldbc_id').alias('expertCandidatePerson.id')),
'person')
.groupBy('`expertCandidatePerson.id`', '`tag.name`')
.agg(F.countDistinct('message').alias('messageCount'))
.orderBy(F.desc('messageCount'), F.col('`tag.name`'), F.col('`expertCandidatePerson.id`'))
.limit(100)
)
q38_result.show(truncate=False)
Q39. Zombies in a country
Select the zombie accounts of a country by a rate test (fewer than one message per month over the account’s lifetime), then compute for each zombie the share of its likes that came from other zombies. The query first derives a subpopulation and then measures a ratio within it, which no defined type describes.
q39_g = GraphFrame(V('person').select('id', 'creationDate'), E('likes').select('src', 'dst'))
q39_end = T('2012-11-11T00:00:00.000+00:00')
q39_people = (persons_in_country('India').select(F.col('person').alias('id'))
.join(V('person').select('id', 'ldbc_id', 'creationDate'), 'id')
.filter(F.col('creationDate') < q39_end))
q39_created = (creators().join(V('comment', 'post').filter(F.col('creationDate') < q39_end)
.select(F.col('id').alias('message')), 'message')
.join(q39_people.select(F.col('id').alias('person')), 'person')
.groupBy('person').agg(F.count(F.lit(1)).alias('messageCount')))
q39_zombies = (q39_people.join(q39_created.withColumnRenamed('person', 'id'), 'id', 'left')
.withColumn('months', 12 * (F.lit(2012) - F.year('creationDate'))
+ (F.lit(11) - F.month('creationDate')) + 1)
.filter(F.coalesce('messageCount', F.lit(0)) < F.col('months'))
.select('id', 'ldbc_id').localCheckpoint())
q39_received = (creators().join(q39_zombies.select(F.col('id').alias('person')), 'person')
.join(q39_g.edges.select(F.col('src').alias('liker'), F.col('dst').alias('message')), 'message')
.join(q39_g.vertices.select(F.col('id').alias('liker'), F.col('creationDate').alias('likerCreated')),
'liker')
.join(q39_zombies.select(F.col('id').alias('liker'), F.lit(1).alias('likerIsZombie')),
'liker', 'left')
.groupBy('person')
.agg(F.count('likerIsZombie').alias('zombieLikeCount'),
F.sum((F.col('likerCreated') < q39_end).cast('long')).alias('totalLikeCount')))
q39_result = (
q39_zombies.join(q39_received.withColumnRenamed('person', 'id'), 'id', 'left')
.select(F.col('ldbc_id').alias('zombie.id'),
F.coalesce('zombieLikeCount', F.lit(0)).alias('zombieLikeCount'),
F.coalesce('totalLikeCount', F.lit(0)).alias('totalLikeCount'))
.withColumn('zombieScore', F.when(F.col('totalLikeCount') == 0, F.lit(0.0))
.otherwise(F.col('zombieLikeCount') / F.col('totalLikeCount')))
.orderBy(F.desc('zombieScore'), F.col('`zombie.id`'))
.limit(100)
)
q39_result.show(truncate=False)
Q40. Fake news detection
For each of two (tag, date) combinations, find the persons who posted about that tag on that day and who have at most maxKnowsLimit friends doing the same, then keep only the persons present in both. The same parameterised computation runs twice and the two results are joined on the person.
q40_g = GraphFrame(persons(), E('knows').select('src', 'dst'))
def q40_side(tag_name, day, letter):
authored = (E('has_tag').join(tag_id(tag_name), F.col('dst') == F.col('tag')).select(F.col('src').alias('message'))
.join(V('comment', 'post').filter(F.to_date('creationDate') == F.lit(day).cast('date'))
.select(F.col('id').alias('message')), 'message')
.join(creators(), 'message').localCheckpoint())
friends = (q40_g.find('(a)-[k]-(b)')
.select(F.col('a.id').alias('person'), F.col('b.id').alias('person2'))
.join(F.broadcast(authored.select('person').distinct()), 'person')
.join(F.broadcast(authored.select(F.col('person').alias('person2')).distinct()), 'person2')
.groupBy('person').agg(F.countDistinct('person2').alias('cp2')))
return (authored.groupBy('person').agg(F.countDistinct('message').alias(f'messageCount{letter}'))
.join(friends, 'person', 'left')
.filter(F.coalesce('cp2', F.lit(0)) <= 5).drop('cp2'))
q40_result = (
q40_side('Marc_Gicquel', '2010-12-02', 'A').join(q40_side('Vladimir_Putin', '2012-11-13', 'B'), 'person')
.join(V('person').select(F.col('id').alias('person'), F.col('ldbc_id').alias('person1.id')), 'person')
.select('`person1.id`', 'messageCountA', 'messageCountB')
.orderBy(F.desc(F.col('messageCountA') + F.col('messageCountB')), F.col('`person1.id`'))
.limit(20)
)
q40_result.show(truncate=False)
Q41. Information propagation analysis
Structurally the heaviest query in the set: it spans two distinct forums, three persons and two variable-length REPLY_OF*0.. segments, then requires message2 to be more than delta hours later than message1 and excludes person1’s membership of forum2. The thread roots come from message_roots(); the reply edge between the comment and message2 is a motif over the tagged messages, and the rest is a multi-way join with a temporal predicate and an anti-join.
q41_g = GraphFrame(V('comment', 'post').select('id'), E('reply_of').select('src', 'dst'))
q41_tag = tag_id('Koda_Kumi')
q41_tagged = (E('has_tag').join(q41_tag, F.col('dst') == F.col('tag')).select(F.col('src').alias('message'))
.join(V('comment', 'post').select(F.col('id').alias('message'), 'creationDate'), 'message')
.join(creators(), 'message')
.join(message_roots(), 'message')
.join(E('container_of').select(F.col('src').alias('forum'), F.col('dst').alias('root')), 'root')
.select('message', 'person', 'creationDate', 'forum')
.localCheckpoint())
q41_member = E('has_member').select(F.col('src').alias('forum'), F.col('dst').alias('person'))
q41_g = GraphFrame(q41_tagged.select(F.col('message').alias('id'), 'person', 'creationDate', 'forum'),
q41_g.edges)
q41_replies = (q41_g.find('(comment)-[r]->(message2)')
.select(F.col('comment.person').alias('person2'), F.col('message2.id').alias('message2'),
F.col('message2.person').alias('person3'), F.col('message2.creationDate').alias('date2'),
F.col('message2.forum').alias('forum2')))
q41_forum1 = (q41_replies.filter(F.col('person2') != F.col('person3'))
.join(q41_member.select(F.col('forum').alias('forum1'), F.col('person').alias('person2')), 'person2')
.join(q41_member.select(F.col('forum').alias('forum1'), F.col('person').alias('person3')),
['forum1', 'person3'])
.filter(F.col('forum1') != F.col('forum2')))
q41_matches = (q41_forum1
.join(q41_tagged.select(F.col('forum').alias('forum1'), F.col('person').alias('person1'),
F.col('creationDate').alias('date1')), 'forum1')
.filter(F.col('date2') > F.col('date1') + F.expr('INTERVAL 10 HOURS'))
.join(q41_member.select(F.col('forum').alias('forum2'), F.col('person').alias('person1')),
['forum2', 'person1'], 'left_anti'))
q41_result = (
q41_matches.groupBy('person1').agg(F.countDistinct('message2').alias('messageCount'))
.join(V('person').select(F.col('id').alias('person1'), F.col('ldbc_id').alias('person1.id')), 'person1')
.select('`person1.id`', 'messageCount')
.orderBy(F.desc('messageCount'), F.col('`person1.id`'))
.limit(10)
)
q41_result.show(truncate=False)