Recently, large scale graph data management, querying and processing have experienced a renaissance in several timely application domains (e.g., social networks, bibliographical networks and knowledge graphs). However, these applications still introduce new challenges with large-scale graph processing. Therefore, recently, we have witnessed a remarkable growth in the prevalence of work on graph processing in both academia and industry. Querying and processing large graphs is an interesting and challenging task. Recently, several centralized/distributed large-scale graph processing frameworks have been developed. However, they mainly focus on batch graph analytics. On the other hand, the state-of-the-art graph databases can't sustain for distributed eficient querying for large graphs with complex queries. In particular, online large scale graph querying engines are still limited. In this paper, we present a research plan shipped with the stateof-the-art techniques for large-scale property graph querying and processing. We present our goals and initial results for querying and processing large property graphs based on the emerging and promising Apache Spark framework, a defacto standard platform for big data processing. In principle, the design of this research plan is revolving around two main goals. The first goal focuses on designing an adequate and eficient graph-based storage backend that can be integrated with the Apache Spark framework. The second goal focuses on developing various Graph-aware optimization techniques (e.g., graph indexing, graph materialized views), and extending the default relational Spark Catalyst optimizer with Graph-aware cost-based optimizations. Achieving these contributions can significantly enhance the performance of executing graph queries on top of Apache Spark.
Graphs are everywhere. They are intuitive and rich data
models that can represent strong connectivity within the data. Due
to their rich expressivity, graphs are widely used in several
application domains including the Internet of Things (IoT), social
networks, knowledge graphs, transportation networks, Semantic
Web, and Linked Open Data (LOD) among many others [
In general, graphs can be represented in diferent data
models [
Recently, several graph query languages have been proposed
to support querying diferent kinds of graph data models [
Problem Statement: With the continuous increase in graph
data, processing large graphs introduces several challenges and
detrimental issues to the performance of graph-based
applications [
In this paper, we provide an overview of the current state-ofthe-art eforts in solving the large scale graph querying along side with their limitations (Section 2). We present our planned contributions based on one of the emerging distributed processing platforms for querying large graph data, Morpheus11 (Section 3). We present our initial results in Section 4, before we conclude the paper in Section 5. 2
Distributed processing frameworks can be utilized to solve the
graph scalability issues with query evaluation. Apache Spark
represents the defacto standard for distributed big data
processing [
Spark-SQL is a high-level library for processing structured as
well as semi-structured large datasets. It enables querying these
datasets stored in DataFrames abstraction using SQL. Spark-SQL
acts as a distributed SQL query engine over large structured
datasets. In addition, SparkSQL ofers a Catalyst optimizer in its
6https://s3.amazonaws.com/artifacts.opencypher.org/website/ocim1/slides/
Graph+Pattern+Matching+in+SAP+HANA.pdf
7https://oss.redislabs.com/redisgraph/
8https://bitnine.net/agensgraph/
9https://memgraph.com/
10https://github.com/opencypher/morpheus
11https://github.com/opencypher/morpheus
core for improving query executions [
GraphX extends the low level RDD abstraction, and introduces a new abstraction called Resilient Distributed Graphs (RDG). In a graph, RDG relates records with vertices and edges and produces an expressive computational primitives’ collection. GraphX chains the benefits of graph-parallel and data-parallel systems. However, GraphX is not currently actively maintained. Besides, GraphX is based on the low level RDGs. Thus, it cannot exploit the Spark 2’s Catalyst query optimizer that supports only Spark DataFrames API. Moreover, GraphX is only available to Scala users.
GraphFrames is a graph package built on DataFrames [
In practice, by default, Spark does not support processing and querying of property graph data model, despite is widespread use. To this extent, the Morpheus project has come to the scene. In particular, the Morpheus project has been designed to enable the evaluation of Cypher over large property graphs using DataFrames on top of Apache Spark framework. In practice, this framework enables combining the scalability of the Spark framework with the features and capabilities of Neo4j by enabling the Cypher language to be integrated into the Spark analytics pipeline. Interestingly, graph processing and querying can be then easily interwoven with other Spark processing analytics libraries such as Spark GraphX, Spark ML or Spark-SQL. Moreover, this enables easy merging of graphs from Morpheus into Neo4j. Besides more advanced capabilities of Morpheus such as the ability to handle multiple graphs (i.e. graph Composability) from diferent data sources even if they are not graph sources (i.e relational data sources), it has the ability to create graph views on the data as well.
Figure 2 illustrates the architecture of Morpheus framework. In Morpheus, Cypher queries are translated into Abstract Syntax Tree (AST ). Then, Morpheus core system translates this AST into DataFrame operations with schema and Data-Type handling. It is worth noting that DataFrames in Spark use schema, while Neo4j or generally property graphs optionally use a schema (i.e. schema free data model). Therefore, Morpheus provides a Graph Data Definition Language GDDL for handling schema mapping. Particularly, GDDL expresses property graph types and maps between those types and the relational data sources. Moreover, the Morpheus core system manages importing graph data that can reside in diferent Spark storage backends such as HDFS (i.e. in diferent file formats), Hive, relational databases using JDBC, or Neo4j (i.e. Morpheus Property graph data sources PGDs), and exporting these property graphs directly back to those Spark storage backends. This interestingly means that graph data can be read from these native Spark data sources without altering nor copying the original data sources to Morpheus. Particularly, it is like you plug-in these storage backends to the Morpheus framework as shown in Figure 2. The native Spark Catalyst optimizer is used in Morpheus pipeline for making various core query optimizations for the generated relational plan of operations. Last but not least, Morpheus runs these optimized query plans on the Spark cluster using distributed Spark runtime environment. 3
In general, Morpheus has been designed to enable executing Cypher queries on top of Spark. However, on the backend, the property graphs are represented and maintained using Spark relational DataFrames. Therefore, Cypher graph-based queries are internally translated into relational operations over these DataFrames. Therefore, Spark still performs operations on the property graph as tabular data and views with specified schema. Thus, adding a graph-aware optimization layer for Spark can significantly enhance the performance of graph query execution on the property graph data. In this research plan, we are focusing on two main aspects for enhancing querying and processing large property graphs in the context of the Morpheus project. The ifrst aspect is to design an eficient Spark-based storage backend for persisting property graphs. The other aspect is to provide graph-aware optimizations for query processing inside Morpheus such as Graph indexing, Graph Materialized Views, and last but not least graph cost-based optimizations on top of the default Spark Catalyst optimizer. In order to achieve these aspects in our research plan, we focus on answering the following Research Questions (RQs):
better performance ?): Large graphs require well-suited
persistence solutions that are eficient for query evaluations and
processing [
RQ2: Graph Indexes (How can we use Graph Indexing
for Better Performance?): The default method for processing
graph queries is to perform a subgraph matching search against
the graph dataset [
Currently, Morpheus doesn’t use any indexing mechanism for property graphs while executing graph queries. To this end, we aim to build an eficient indexing scheme of the property graphs in an ofline mode (taking into consideration its schema, as well as its storage backend). Then, this index will be used for reducing the search space for the complex graph pattern matching task. Thus, consulting this index for executing the graph query workload is better than exhaustive vertex-to-vertex correspondence checking from the query to the graph which involves a lot of expensive join operations in the relational representation. In our plan, we don’t consider the overhead of updating the built index, as Morpheus is currently supporting only read operations, and thus no insertions and deletions happen to the already generated property graph.
databases tend to compute each query from scratch without
being aware of the previous query workloads [
It is worthy to note that, graph materialization has its side
efects regarding the memory space that you need to sacrifice
for keeping those views. This also comes with another challenge
concerning the selection of best proper views (i.e. graph
substructures and properties of interest) to materialize and keep in
memory [
graph CBO for better performance?): In general, Spark SQL
uses the Catalyst optimizer to optimize all the queries written
both in explicit SQL or in a DataFrame Domain Specific Language
(DSL). Basically, Catalyst is a Spark library built as a
relationalbased optimization engine. Each rule in the rule-based part of
Catalyst focuses on a specific optimization. Catalyst can also
apply various relational cost-based optimizations for improving
the quality of multiple alternative query physical execution plans.
Although there are several eforts for optimizing the cost-based
techniques in Spark-SQL such the work proposed recently in [
To tackle this challenge, we aim to provide a graph-aware query planner which will be implemented as a layer on top of the default Spark Catalyst for providing a GCBO query plan, taking into account the statistics of the property graph that resides in Morpheus storage backend. Particularly, the new graph planner/optimizer can select the best join of tables order based on selectivity and cardinality estimations of graph patterns in the graph query for filter and join operators. Therefore, at the query time, the new GCBO can suggest a more optimized query plan for the Catalyst to follow. 4
In this section, we describe our initial experimental results for answering RQ1. In particular, we have designed a set of micro and macro benchmarking experiments for evaluating the performance of diferent Spark storage backends supported by the Morpheus framework. These storage backends are: Neo4j, Hive, and HDFS with its diferent file formats (CSV,Parquet, and ORC). Notably, we don’t copy data from these Spark storage backends, we only evaluate Morpheus performance with the data already resides in them. We have used the Cypher LDBC Social Network Benchmark (SNB) BI benchmark query workload12. Our selected queries are read only queries (i.e. no updates are supported by Morpheus).
have been performed on a Desktop PC running a Cloudera Virtual Machine (VM) v.5.13 with Centos v7.3 Linux system, running 12https://github.com/ldbc/ldbc_snb_implementations/tree/master/cypher/queries on Intel(R) Core(TM) i5-8250U 1.60 GHzX64-based CPU and 24 GB DDR3 of physical memory. We also used a 64GB virtual hard drive for our VM. We used Spark V2.3 parcel on Cloudera VM to fully support Spark-SQL capabilities. We used the already installed Hive service on Cloudera VM (version:hive1.1.0+cdh5.16.1+1431), and neo4j V3.5.8.
Benchmark Datasets: Using the LDBC SNB data generator13, we generated a graph data set (in CSV format) of Scale Factor (SF=1). We used this data to create a property graph in Neo4j using Neo4j import tool14. The generated property graph has more than 3M nodes, and more than 17M relationships. We also created a graph of tables and views of the same schema inside Hive. Further, we used Morpheus Framework to read this property graph either from Hive or Neo4j to store the same graph into HDFS into Morpheus supported file formats (CSV, ORC, Parquet).
For both experiments (Micro and Macro Benchmarking), we run the experiments for all queries five times (excluding the first cold-start run time, to avoid the warm-up bias, and computed an average of the other four run times). Notably, we take the ( ln) function of average run times in the Macro-Benchmark experiment15.
ing experiment, we selected 21 BI queries (i.e. which are valid to run in the current Morpheus Cypher constructs)16. The results of Figure 4 show that Hive has the lowest performance in general for running most of the queries even those that are not complex with 70% of low performance than others. HDFS backends in general outperform Neo4j and Hive with 100% of better performance. In particular, Parquet format in HDFS has the best performance. It outperforms ORC and CSV format in most cases of running the queries with 42%. While both CSV and ORC achieve only 28.5% of higher performance.
Morpheus Micro-Benchmark: In our Micro-benchmark experiment, we run 18 Atomic/micro level BI queries17. The results of Figure 5 show that Neo4j has the lowest performance in general for running the first 12 queries with 66% of low performance than others. Hive starts to perform worse than Neo4j (and all other systems) only when the number of joins increase and sorting being applied on queries from Q13 to Q18. While, HDFS backends in general outperform Neo4j and Hive with 94.4% of better performance. In particular, Parquet format in HDFS has the best performance, it outperforms ORC and CSV in most queries with 55%. While CSV and ORC only outperform with 22.2% and 16.6%, respectively. 5
We are living in an era of continuous huge growth of more and more connected data. Querying and processing large graphs is an interesting and challenging task. The Morpheus framework aims at integrating Cypher query language to work as a graph query language on top of Spark. Morpheus translates Cypher queries into relational Dataframes operations that can fit in the SparkSQL environment. Morpheus depends mainly on default Spark Catalyst optimizer for optimizing those relational operators. No graph indexing nor graph materialized views are maintained in Morpheus or Spark SQL framework for optimizing property 13https://github.com/ldbc/ldbc_snb_datagen 14https://neo4j.com/docs/operations-manual/current/tools/import/ 15The code and results of our initial experiments is available on https://github.com/ DataSystemsGroupUT/MorephusStorageBenchmarking 16http://bit.ly/2W5b01N Macro LDBC SNB BI Queries 17http://bit.ly/2Pa4TrF Micro/Atomic level queries graph data querying and processing. In this PhD work, we focus on tackling these challenges by designing an eficient storage backend for persisting property graphs for Morpheus. In addition, we aim at providing graph-aware techniques (e.g., indexes, materialized views) for Spark to optimize the graph queries, in addition to other graph-aware CBO for Spark Catalyst Optiomizer. We believe that achieving these contributions as our future research plan can significantly enhance the performance of executing graph queries using the Morpheus framework.