Distributed temporal graph analytics with GRADOOP

Graphs are simple, yet powerful data structures to model and analyze relations between real-world data objects. The analysis of graph data has recently gained increasing interest, e.g., for web information systems, social networks [23], business intelligence [60, 69, 81] or in life science applications [20, 51]. A powerful class of graphs are so-called knowledge graphs, such as in the current CovidGraph project¹, to semantically represent heterogeneous entities (gene mutations, publications, patients, etc.) and their relations from different data sources, e.g., to provide consolidated and integrated knowledge to improve data analysis. There is a large spectrum of analysis forms for graph data, ranging from graph queries to find certain patterns (e.g., biological pathways), over graph mining (e.g., to rank websites or detect communities in social graphs) to machine learning on graph data, e.g., to predict new relations. Graphs are often large and heterogeneous with millions or billions of vertices and edges of different types making the efficient implementation and execution of graph algorithms challenging [42, 73]. Furthermore, the structure and contents of graphs and networks mostly change over time making it necessary to continuously evolve the graph data and support temporal graph analysis instead of being limited to the analysis of static graph data snapshots. Using such time information for temporal graph queries and analysis is valuable in numerous applications and domains, e.g., to determine which people have been in the same department for a certain duration of time or to find out how collaborations between universities or friendships in a social network evolve, how an infectious disease spreads over time, etc. Like in bitemporal databases [37, 38], time support for graph data should include both valid time (also known as application time) and transaction time (also known as system time), to differentiate when something has occurred or changed in the real world and when such changes have been recorded and thus became visible to the system. This management of graph data along two timelines allows one to keep a complete history of the past database states as well as to track the provenance of information with full governance and immutability. It also allows to query across both valid and system time axes, e.g., to answer questions like “What friends did Alice have on last August 1st as we knew it on September 1st?”.

Two major categories of systems focus on the management and analysis of graph data: graph database systems and distributed graph processing systems [42]. A closer look at both categories and their strengths and weaknesses is given in Sect 2. So graph database systems are typically less suited for high-volume data analysis and graph mining [31, 53, 75] as they often do not support distributed processing on partitioned graphs which limits the maximum graph size and graph analysis to the resources of a single machine. By contrast, distributed graph processing systems support high scalability and parallel graph processing but typically lack an expressive graph data model and declarative query support [13, 42]. In particular, the latter makes it difficult for users to formulate complex analytical tasks as this requires profound programming and system knowledge. While a single graph can be processed and modeled, the support for storing and analyzing many distinct graphs is usually missing in these systems. Further, both categories have typically neither native support for a temporal graph data model [49, 74] nor for temporal graph analysis and querying.

To overcome the limitations of these system approaches and to combine their strengths, we started in 2015 already the development of a new open-source² distributed graph analysis platform called Gradoop (Graph Analytics on Hadoop) that has continuously been extended in the last years [29, 39‐41, 43, 65, 66]. Gradoop is a distributed platform to achieve high scalability and parallel graph processing. It is based on an extended property graph model supporting the processing both of single graphs and of collections of graphs, as well as an extensible set of declarative graph operators and graph mining algorithms. Graph operators are not limited to a common query functionality such as pattern matching but also include novel operators for graph transformation or grouping. With the help of a domain-specific language called GrALa, these operators and algorithms can be easily combined within dataflow programs to implement data integration and graph analysis. While the initial focus has been on the creation and analysis of static graphs, we have recently added support for bitemporal graphs making Gradoop a distributed platform for temporal graph analysis.

In this work, we present a complete system overview of Gradoop with a focus on the latest extensions for temporal property graphs. This addition required adjustments in all components of the system, as well as the integration of analytical operators tailored to temporal graphs, for example, a new version of the pattern matching and grouping operators as well as support for temporal graph queries. We also outline the implementation of these operators and evaluate their performance. We also reflect on lessons learnt from the Gradoop effort.

The main contributions are thus as follows:After an overview of the current graph system landscape (Sect. 2), the architecture of the Gradoop framework is described in Sect. 3. The bitemporal graph data model including an outline of all available operators and a detailed description of selected ones is given in Sect. 4. After explaining implementation details (Sect. 5), selected operators are evaluated in Sect 6. In Sect. 7, we discuss lessons learned, related projects and ongoing work. We summarize our work in Sect. 8.

Gradoop relates to graph data processing and temporal databases, two areas with a huge amount of previous research. We will thus mostly focus on the discussion of temporal extensions of property graphs and its query languages and their support within graph databases and distributed graph processing systems. We also point out how Gradoop differs from previous approaches.

With Gradoop, we provide a framework for scalable management and analytics of large, semantically expressive temporal graphs. To achieve horizontal scalability of storage and processing capacity, Gradoop runs on shared-nothing clusters and utilizes existing open source frameworks for distributed data storage and processing. The difficulties of distributing data and computation are hidden beneath a graph abstraction allowing the user to focus on the problem domain.

Figure 1 presents an overview of the Gradoop architecture. Analytical programs are defined within our Graph Analytical Language (GrALa), which is a domain-specific language for the Temporal Property Graph Model (TPGM). GrALa contains operators for accessing static and temporal graphs in the underlying storage as well as for applying graph operations and analytical graph algorithms to them. Operators and algorithms are executed by the distributed execution engine which distributes the computation across the available machines. When the computation of an analytical program is completed, results may either be written back to the storage layer or presented to the user. In the following, we briefly explain the main components, some of which are described in more detail in later sections. We will also discuss data integration support to combine different data sources into a Gradoop graph.

Distributed storage Gradoop supports several ways to store TPGM compliant graph data. To abstract the specific storage, GrALa offers two interfaces: DataSource to read and DataSink to write graphs. An analytical program typically starts with one or more data sources and ends in at least one data sink. A few basic data sources and sinks are built into Gradoop and are always available, e.g., a file-based storage like CSV that allows reading and writing graphs from the local file system or the Apache Hadoop Distributed File System (HDFS) [76]. Gradoop in addition supports Apache HBase [2] and Apache Accumulo [1] as built-in storage, which provides database capabilities on top of the HDFS. Data distribution and replication as well as error handling in the case of cluster failures are handled by HDFS. Due to the available interfaces, Gradoop is not limited to the predefined storages. Other systems, e.g., a native graph database like Neo4j or a relational database, can be used as graph storage by implementing the DataSource and DataSink interfaces. Storage formats will be discussed in Sect. 5.5.

Distributed Execution Engine Within Gradoop, the TPGM and GrALa provide an abstraction for the analyst to work with graphs. However, the actual implementation of the data model and its operators is transparent to the user and hidden within the distributed execution engine. Generally, this can be an arbitrary data management system that allows implementing graph operators. Gradoop uses Apache Flink, a distributed batch and stream processing framework that allows executing arbitrary dataflow programs in a data parallel and distributed manner [8, 16]. Apache Flink handles data distribution along with HDFS, load balancing and failure management. From an analytical perspective, Flink provides several libraries that can be combined and integrated within a Gradoop program, e.g., for graph processing, machine learning and SQL. Apache Flink is further described in Section 5.1.

Temporal property graph model The TPGM [66] describes how graphs and their evolution are represented in Gradoop. It is an extension of the extended property graph model (EPGM) [40] which is based on the widely accepted property graph model [11, 63]. To handle the evolution of the graph and its elements (vertices and edges), the model uses concepts of bitemporal data management [37, 38] by adding two time intervals to the graph and to each of its elements. To facilitate integration of heterogeneous data, the TPGM does not enforce any kind of schema, but the graph elements can have different type labels and attributes. The latter are exposed to the analyst and can be accessed within graph operators. For enhanced analytical expressiveness, the TPGM supports handling of multiple, possibly overlapping graphs within a single analytical program. Graphs, as well as vertices and edges, are first-class citizens of the data model and can have their own properties. Furthermore, graphs are the input and output of analytical operators which enables operator composition. Section 4.1 describes the TPGM model in more detail.

Graph analytical language Programs are specified using declarative GrALa operators. These operators can be composed as they are closed over the TPGM, i.e., take graphs as input and produce graphs. There are I/O operators to read and write graph data and analytical operators to transform or analyze graphs. Table 1 shows a subset of frequently used analytical operators and graph algorithms categorized by their input [40, 66]. There are specific operators for temporal graphs to determine graph snapshots or the difference between two snapshots as well as temporal versions of more general operators such as pattern matching [41] and graph grouping [43, 66]. Furthermore, there are dedicated transformation operators to support data integration [46]. Each category contains auxiliary operators, e.g., to apply unary graph operators on each graph in a graph collection or to call external algorithms. GrALa already integrates well-known graph algorithms (e.g., page rank or connected components), which can be seamlessly integrated into a program. Graph operators will be further described in Sect. 4.2.

Programming interfaces Gradoop provides two options to implement an analytical program. The most comprehensive approach is the Java API containing the TPGM abstraction including all operators defined within GrALa. Here, the analyst has the highest flexibility of interacting with other Flink and Java libraries as well as of implementing custom logic for GrALa operators. For a user-friendly visual definition of Gradoop programs and a visualization of graph results, we have incorporated Gradoop into the data pipelining tool KNIME Analytics Platform [14]. This extension makes it possible to use selected GrALa operators within KNIME analysis workflows and to execute the resulting workflows on a remote cluster [67, 68]. KNIME and the Gradoop extension offer built-in visualization capabilities that can be leveraged for customizable result and graph visualization.

Data integration support Gradoop aims at the analysis of integrated data, e.g., knowledge graphs, originating from different heterogeneous sources. This can be achieved by first translating the individual sources into a Gradoop representation and then performing data integration for the different graphs. Gradoop provides several specific data transformation operators to support this kind of data integration, e.g., to achieve similarly structured graphs (see Sect. 4.2.6). Furthermore, we provide extensive support for entity resolution and entity clustering within a dedicated framework called FAMER [70‐72] which is based on Gradoop and Apache Flink. FAMER determines matching entities from two or more (graph) sources and clusters them together. Such clusters of matching entities can then be fused to single entities (with a Fusion operator) for use in an integrated Gradoop graph. In future work, we will provide a closer integration of Gradoop and FAMER to achieve a unified data transformation and integration for heterogeneous graph data and the construction and evolution of knowledge graphs [59].

In this section, we present the Temporal Property Graph Model (TPGM) as the graph model of Gradoop that allows the representation of evolving graph data and its analysis. We first describe the structural part of TPGM to represent temporal graph data and then discuss the graph operators as part of GrALa. The last subsection briefly discusses the graph algorithms currently available in Gradoop.

In this chapter, we will describe the implementation of the TPGM and GrALa on top of a distributed system. Since Gradoop programs model a dataflow where one or multiple temporal graphs are sequentially processed by chaining graph operators, the utilization of distributed dataflow systems such as Apache Spark [84] and Apache Flink [16] is especially promising. These systems offer, in contrast to MapReduce [24], a wider range of dataflow operators and the ability to keep data in main memory between the execution of those operators. The major challenges of implementing graph operators in these systems are identifying an appropriate graph representation and an efficient combination of the primitive dataflow operators to express graph operator logic.

As discussed in Sect. 2, the most recent approaches to large-scale graph analytics are libraries on top of such distributed dataflow frameworks, e.g., GraphX [83] on Apache Spark or Gelly [3] on Apache Flink. These libraries are well suited for executing iterative algorithms on distributed graphs in combination with general data transformation operators provided by the underlying frameworks. However, the implemented graph data models have no support for temporal graphs, collections and are generic, which means arbitrary user-defined data can be attached to vertices and edges. In consequence, model-specific operators, i.e., such based on labels, properties or the time attributes, need to be user-defined, too. Hence, using those libraries to solve complex analytical problems becomes a laborious programming task.

We thus implemented Gradoop on top of Apache Flink to provide new features for flexible and general-purpose graph analytics and to benefit from existing capabilities to large-scale data and graph processing at the same time. The majority of graph algorithms listed in Table 1 are available in Flink Gelly. Gradoop adds automatic transformation from TPGM graphs into Gelly graphs and vice versa, as later described. In this section, we will briefly introduce Flink and its programming concepts. We will further show how the TPGM graph representation and a subset of the introduced operators, including graph algorithms, are mapped to those concepts. The last section focuses on persistent graph formats.

This section is split into two parts. First we show scalability results for a subset of individual TPGM operators, namely snapshot, difference, time-dependent grouping and temporal pattern matching. In the second part, we analyze the performance evaluation for an analytical program composing several operators. In both cases, we evaluate runtimes and horizontal scalability with respect to increasing data volume and cluster size. The experiments have been run on a cluster with 16 worker nodes. Each worker consists of a E5-2430 6(12) 2.5 Ghz CPU, 48 GB RAM, two 4 TB SATA disks and runs openSuse 13.2, Hadoop 2.7.3 and Flink 1.9.0. On a worker node, a Flink Task Manager is configured with 6 task slots and 40GB memory. The workers are connected via 1 Gigabit Ethernet.

We use two datasets referred to as LDBC and CitiBike in the evaluation. The LDBC dataset generator creates heterogenous social network graphs with a fixed schema and structural characteristics similar to real-world social networks, e.g., p-law distribution [35]. CitiBike is a real-world dataset describing New York City bike rentals since 2013¹¹. The schema of this dataset corresponds to the one in Fig 2 in Sect. 4. Table 5 contains some statistics about the two datasets considering different scaling factors (SF) for LDBC. The largest graph (SF=100) has about 283 million vertices and 1.8 billion edges. The CitiBike graph covers data over almost eight years with up to 20M new edges per year.

The exemplified GrALa workflow for this analysis is shown in Listing 1. The input is the fully evolved CitiBike network as a single logical graph. First, we extract a snapshot containing only bike rentals happened in 2018 and 2019 with operator Snapshot (line 2–3). In lines 4–5, we add a specific cellId calculated from the geographical coordinates in its properties to each vertex (rental station) with the Transformation operator. The Temporal Pattern Matching operator in lines 6–14 uses the enriched snapshot to match all pairs of rentals with a duration of at least 40 or 90 minutes, each where the first trip starts in a specific cell (2883) in NYC and the second trip starts in the destination of the first trip after the end of the first trip. Since the result of the Temporal Pattern Matching operator is a graph collection we use a Reduce operator (line 15) to combine the results to a single logical graph. We further group the combined graph by the vertex properties name and cellId (line 17) and the edge creation per month (line 19) using the time-dependent grouping operator. By applying two aggregation functions on the edges (lines 20–22), we determine both the number of edges represented by a superedge and the average duration of the found rentals. Finally, we apply a Subgraph operator to output only superedges with a count greater than 1 (line 23). The corresponding source code is available online¹².

Figures 10, 11 and 12 show the performance results for both the execution of individual operators (snapshot, difference, grouping and pattern matching (query)) for the LDBC dataset and for the analytical program on the real-world Citi Bike dataset (CBA40 and CBA90). We run each experiment five times and report average execution times. Figure 10 shows the impact of the LDBC data size on the runtime of individual operators. Figures 11 and 12 show runtimes and speedup for different cluster sizes with LDBC SF 100 (for individual operators) and the CitiBike dataset (for the CBA program).

Snapshot and difference Both temporal operators scale well for both increasing data volume (Fig. 10) and cluster size (Figs. 11 and 12). In general, the execution of Difference is slower than Snapshot since it has to evaluate two predicates as intermediate results and add a property to each element.

Figure 12 shows the speedup for LDBC.100 grows nearly linearly until 4 parallel workers and is slightly declining then for more workers. This behavior is typical for distributed dataflow engines since a higher worker count increases the communication overhead over the network (especially for the semi-join performed in the verification step), while the useful work per worker becomes smaller with larger configurations.

Grouping The evaluated Grouping operator on the LDBC datasets uses two grouping key functions: one on label values and one that extracts the week timestamp of the lower bound of the valid time interval. The grouping result thus consists of supervertices and superedges that show the weekly evolution of created entities and their respective relationships over several years. For aggregation, we count the number of new entities and relationships per week.

Figure 10 shows that Grouping also scales nearly linearly for increasing data size (from LDBC.10 to LDBC.100) similar to the two previously discussed operators. The runtime reductions for smaller graphs are limited due to job initialization times. The execution time for LDBC.100 is reduced from 8,097 seconds on a single worker to 967 seconds on 16 workers (Fig. 11) resulting in a speedup of more than 8. Figure 12 shows that the speedup for Grouping is almost linear for up to 8 workers, while more workers result only in modest improvements. These results are similar to the original evaluation of the non-temporal grouping operator in [43].

The shown performance results are influenced by the communication overhead to exchange data, in particular for the join and group-by transformations in our operator implementations. This overhead increases and becomes dominant for more workers. In addition, the usage of a ReduceGroup transformation on the grouped vertices (see Sec 5.3) can lead to an unbalanced workload for skewed group sizes. We already addressed this issue in [43]; however, a more comprehensive evaluation of the influence of skewed graphs is part of future work.

Pattern matching (query) In this experiment, we execute a predefined temporal graph query on the LDBC data on various cluster sizes and two partitioning methods: First, hash-partitioned, which uses Flink’s default partitioning approach combined with Gradoop’s basic GVE Layout and second, label-partitioned, a custom partitioning approach which combines benefits of data locality and Gradoop’s experimental Indexed GVE Layout (see Sect. 5.2.2). The used TemporalGDL query is shown below and determines all persons that like a comment to a post within partly overlapping periods.

Regarding the default hash partitioning, Fig. 10 shows that the execution of this query scales linearly with a larger data size from LDBC.10 (78 s) to LDBC.100 (733 s). Increasing the cluster size for LDBC.100 reduces the query runtime from 6857 s for one worker to 733 seconds using 16 workers (Fig. 11). The speedup behavior (Fig. 12) is perfect for up to 4 workers and levels out for more workers to 7.8 for 16 workers due to increasing communication overhead resulting from semi-joins.

Applying the experimental label-partitioned approach, both the runtimes and the speedup results improve significantly compared to the use of hash partitioning. As shown in Fig. 10, the runtime improvements increase with growing data volume from a factor of 2 for LDBC.10 (37 instead of 78 s) to a factor of 4 for LDBC.100 (178 vs. 733 s).

This significant improvement is mainly achieved by a reduced complexity during semi-join execution, since our Indexed GVE Layout provides an indexed access via type labels. Therefore, Flink is able to optimize the execution pipeline by avoiding complex data shuffling and minimizing intermediate result sets. This is also confirmed by analyzing the impact of a growing number of workers on runtimes (Fig. 11) and speedup (Fig. 12) for LDBC.100. For 16 workers, our label-partitioned approach achieves an excellent speedup of 13.5 compared to only 7.8 with the hash-partitioned GVE Layout.

Overall, the results of the hash-partitioned experiment are similar to the ones of the other operators, while the label-partitioned experiment significantly improves runtimes and speedup by reducing the semi-join complexity and communication effort. A more comprehensive evaluation including selectivity evaluation for different queries and partitioning approaches is left for future work.

CBA The results for the more complex analytical pipelines CBA40 and CBA90 are shown in Figs 11 and 12. Note that both analytical programs contain the same operator pipeline but with different rental durations for the query operator resulting in largely different result sets of 10 M (CBA40) and 150 K (CBA90) matches (i.e., matching subgraphs of the query graph) representing trips with a duration of at least 40 or 90 min.

The more selective CBA90 program could be executed in 223 s on a single worker and only 42 s on 16 workers, while CBA40 takes 1336 and 352 s for 1 and 16 workers, respectively. The sequence of multiple operators within the program leads to smaller speedup value compared to the single operators. This is especially pronounced for CBA40 leading to large intermediate results to be processed by the operators coming after the pattern matching operator. Intermediate results are graphs kept in-memory as input for subsequent operators, and the execution time of these operators is strongly dependent on the size and also the distribution of their input.

In recent years, however, Apache Flink has focused to become a pure stream processing engine so that the DataSet API used by Gradoop is planned to be deprecated. We have therefore begun to evaluate alternate processing frameworks such as the DataStream and Table APIs of Apache Flink. Although a re-implementation of a large part of Gradoop and its operators would be necessary, the reorientation can also provide advantages for improving Gradoop’s performance and feature set: The streaming model of Apache Flink already supports two different notions of time (processing and event time, similar to the bitemporal model of the TPGM), watermarks for out-of-order streams, several window processing features and state backends to materialize intermediate results. Such a major change could thus allow better support for processing, analysis and continuous queries on graph streams [15] that we plan to address in the future.

Logical graphs and collections A unique selling point of Gradoop’s original EPGM and the temporal TPGM models is the introduction of logical graphs as an abstraction of a subgraph that can be easily semantically enhanced without changing the structure of the graph by adding new vertex or edge types or adding redundant properties. Graph collections, i.e., sets of logical graphs, have been found to be a hugely valuable data structure for modeling (possibly overlapping) logical graphs. Graph collections also facilitate the use of binary graph operations, an essential part of graph theory, for property graphs. In addition, they are used as a result of analytical operators that produce multiple graphs, for example, graph pattern matching, where each match represents a logical graph.

Operator concept Another core design decision was the methodology to introduce operators that can be combined to define analytical workflows. Similar to the transformations between datasets in distributed processing engines, we developed single analytical operators that are closed over the model. The internal logic of an operator is a composition of (Flink) transformations and hidden to the analyst, thus providing a top level abstraction of the respective function. The large number of operators provided, which contain both simple analyzes and graph algorithms, can therefore be used as a tool kit for composing complex analysis pipelines. An analyst with programming experience can write new or modify existing operators (which requires knowledge of the implementation details) to extend the functional scope of Gradoop. Further, a typical feature of distributed in-memory systems is a scheduler that translates and optimizes the dataset transformations used in the program to a directed acyclic graph (DAG). Such transformations, represented by vertices in the DAG, are combined and chained to optimize the overall data flow. This, however, results in a disadvantage of the operator concept since the relation between transformations and operators often gets diluted. As a consequence, performance issues with specific operators and its dataset transformations are hard to identify.

Temporal extensions The evolution of entities and relationships is a natural characteristic of many graph datasets that represents a real-world scenario [73]. A static graph model like the PGM or the initial EPGM of Gradoop is in most cases unsuitable for performing analyzes that specifically examine the development of the graph over time. We found that an extension of the data model and the respective operators (including new operators for solely temporal analysis) was covering many requirements of frequently used temporal analysis, for example, the retrieval of a snapshot, without building a completely new model and prototypical implement a new framework. In addition, through the operator concept, it is possible to combine static and temporal operators, for example, to first filter for entities and relationships of interest and then analyzing their evolution with the grouping operator and its temporal aggregations. In future work, we plan to develop alternatives to the GVE data layout without separating vertices and edges so that TPGM integrity conditions can be checked more efficiently. We will also investigate the separation of the newest graph state from its history for increased performance.

Scalability The evaluations so far showed that Gradoop generally scales very well with increasing dataset sizes. On the other hand, the speedup when increasing the number of machines on a fixed dataset reaches its limit relatively fast. (For the considered workloads and datasets, the speedup was mostly lower than 10 with the default hash-based data partitioning.) A main reason for this behavior can be seen in the strong dependency on the underlying Flink system and its optimizer and scheduler that are not tailored to graph data processing. For example, data distribution is by default based on a hash partitioning, in particular for intermediate results in an analytical pipeline that prevents the utilization of data locality for our graph operators but can result in large communication overhead even for traversing edges. For that reason, we prepared a possibility to partition a graph by label to reduce execution complexity for large-scale graphs. However, experience shows that a single partition strategy is not suited for all analytical operators Gradoop provides. Part of our future research will thus be to develop improved data distribution and load balancing techniques to achieve a better speedup behavior for single operators as well as for analysis pipelines.

Usage and acceptance Gradoop is an open-source (Apache License 2.0) research framework that has been co-developed by developers from industrial partners and many students within their bachelor, master and Ph.D. theses, which led to a good number of publications. It has been used by us and others within different industrial collaborations [67] and applications and serves as basis for other research projects, e.g., on knowledge graphs [72].

Furthermore, concepts of Gradoop operators have been adopted by companies. For example, the graph grouping operator from Neo4j’s APOC library was inspired by Gradoop’s grouping operator [56]. The implementation of Gradoop’s pattern matching operator [41] as a proof of concept for the distributed execution of Cypher(-like) queries, directly influenced the development of Neo4j’s Morpheus¹³ project, which provides the OpenCypher grammar for Apache Spark by using its SQL DataFrame API.

To make it easier to start using Gradoop, we deploy the system weekly to the Maven Central Repository¹⁴. Thus, it can be used in own projects by solely adding a dependency without any additional installation effort. We provide further a “getting started” guideline and many example programs in the GitHub repository of Gradoop and in its GitHub wiki [64] to support the usage.

The provided analytical language GrALa can be used in two ways, via Java API or KNIME extensions, to define an analytical program. Consequently, analysts with and without programming skills can use the system for graph analysis. Operators can be configured in many ways, whereby it can be difficult to find out which special configuration should be used for the desired analysis result. Further, many possible combinations of the operators may also represent a challenge for the analyst. We therefore provide example operator configurations and a detailed documentation in the GitHub wiki [64] to assist for finding the right combination and configuration.

Ongoing work In the future, we will investigate how the TPGM and its operator concept can be realized by alternate technologies, e.g., using a distributed streaming model or Actor-like abstractions such as Stateful Functions¹⁵ [7], to process temporal graphs. The continuous analysis of graph streams, where events indicate vertex and edge additions, deletions and modifications, is an emerging research area [17] and will be considered in our future research. Besides, we will integrate machine learning (ML) methods on temporal and streaming graphs, e.g., to identify diminishing and recurring patterns or to predict further evolution steps. Another area that we will work more on is to extend Gradoop to realize temporal knowledge graphs integrating data from different sources and to continuously evolve the integrated data. A further extension is the addition of layouting algorithms for the visualization of logical graphs and graph collections to offload the expensive calculation from frontend application. Finally, we will investigate more in overall performance optimization of Gradoop and add temporal graph algorithms as further operators to the framework.

We provided a system overview of Gradoop, an open-source graph dataflow system for scalable, distributed analytics of static and temporal property graphs. The core of Gradoop is the TPGM, a property graph model with bitemporal versioning and graph collection support, which is formally defined in this work. GrALa, a declarative analytical language, enables data analysts to build complex analytical programs by connecting a broad set of composable graph operators, e.g., time-dependent graph grouping or temporal pattern matching with TemporalGDL as a query language. Analytical programs are executed on a distributed cluster to process large (temporal) graphs with billions of vertices and edges. Several implementation details show how the parts of the framework are realized using Apache Flink as distributed dataflow system. Our experimental analyses on real-world and synthetic graphs demonstrate the horizontal scalability of Gradoop. By applying a suitable custom partitioning, we were able to speed up the performance of our pattern matching operator by a large margin. We finally reflect on several lessons learned during our 6-year experience working on that project. Besides more performance optimizations and graph partitioning, future research directions we consider are: (i) using alternative technologies for Gradoop, its model and operators, (ii) extend our analysis from temporal graphs to graph streams, and (iii) the integration of analysis using graph ML.