(Cypher)-[:ON]-❯(ApacheFlink)❮-[:USING]-(Gradoop)
Junghanns, Martin
Kießling, Max
Graph pattern matching is one of the most interesting and challengingoperations in graph analytics. However, it is primarily supported by graphdatabase systems such as Neo4j but, besides research prototypes, not generallyavailable for distributed (not-only graph) processing frameworks like ApacheFlink or Apache Spark. In our talk, we want to give an overview of our current implementation ofCypher on Apache Flink. Cypher is the Neo4j graph query language and enablesthe intuitive definition of graph patterns including structural and semanticpredicates. As the Neo4j graph data model is not supported out-of-the box byApache Flink, we leverage Gradoop, a Flink-based graph analytics frameworkbased on Apache Flink that already provides an abstraction of schema-freeproperty graphs. We will give a brief overview about the technologies used to implement Cypher,explain our query engine and give a demonstration of the available languagefeatures. Finally, we will discuss open challenges and missing featureshopefully motivating people to contribute. The project is a cooperation between the University of Leipzig and NeoTechnology.
 Hello everyone, my name is Marcel. I'm a PhD student at the University of Lysus. This is Max, he's a master student who's working with me together and also at the University of Lysus in the future. And for me and my crew, we are working on a group. I will explain what it is, it's based on questions. And we will show you how we implement side power inside the system. So for the talk, we thought this was a motivation. Then we will show you what we do for those who don't know what are the changes and those who don't know what kind of this. Then we talk about how side power is realized and implemented in the system and therefore shown in the picture. So let's start with some motivation. So at first, when we talk about graphs, the typical example are social networks and also the example that we choose. In that case, it's the social network of words. So each vertex here we can change the word and the relations between them is the HH, so a H other. And smart analysts would ask something like, who are the closest enemies of each order? So that's the typical question in different fields of the world. And if you want to express such a question in the system, you could do it for example with Neo4j, side power and then you have it very like this. So we have a mesh part, there is an org, we create another org and there are one or two of them. So one or two HHs. Okay, let's have a look at this. How this one would be looking like in plain jelly, which is the graph-presetting frame on top of the patch of the game. So this is the same. It could be nothing but it's expected that it will happen. Because a lot of quotes express the same thing. It's even very soon in. So what's happening here is the messaging between those two. Here we have something like the H-fabers, like we've heard about it before, and here the two means that we get the right twice. So there are a lot of quotes that express the same thing and just express it with the size of the screen. And another problem that we have here is typically these graphs are not homogeneous like this one. So we have different kinds of both, so there's different kinds of action. For example, we have plans and we have the hobbits, for example, and we have not only the H-relationships, we have the plan leader, the plan leader, and the node relationship between the hobbits, for example. And now the enders would ask something like which plan leaders hate each other and one of them knows Kono over one to ten options. So that's also a really good question. Let's look at this how we would say this in the cycle. So this would be the ferry. This would be the ferry. So we have the plan. We have the leader of this plan, this leader against number of which is also a plan leader, and the other plan leader knows Kono in one to ten options. So we have some predicates here. Okay, how would you express this on a craft processing framework? You can try it, but no. So it's kind of our motivation, so we want to make craft queries or craft queries especially usable on those kind of frameworks. So this could be a data flow framework like a project link, for example, such as Spark. So to be honest, of course, each of those systems has a specific use case or they are at some point. So we have here two dimensions, more complex than this, but for this one it's enough. So we have the ease of use, which is like things of how easy is to express the problem in a system, and we have data volume. So it's about scalability, using more machines to compute. So on the one hand of the spectrum, we have craft database like for a company of which age. So you have this declarative languages, easy to express the problem, visualization, a lot of tooling, community, and so on. And on the other hand of this spectrum, you have craft processing systems. The main one is Apache G-Raf, which is really dedicated to compute craft algorithm on this particular cluster, so a shared-computer cluster, but only for this, nothing else. Then we have a bit more extraction here. Those are the craft data flow systems. So like for example, it's also a craft processing system, but on top of a data flow system like Apache Spark or Apache Flink, CraftX for example on Spark, Jelly on Flink. And now you see there's a free area in the right upper corner, and this is where we position our research project, which is called CRADUP, and where we want to combine the expressiveness of craft databases, not as much as craft databases, but a little bit less, but also the scalability. So that's kind of the goal of the project. And I will give you a short overview about the project before we dive into the Cypher implementation. So we built on the ShartNothing cluster like I already said. So we use Hadoop and HDFS. It's our data source, data sync, so that's where we store our crafts. We use Apache Flink as a distributed operator execution layer. So this is where we implement actually our craft operators. And CRADUP itself is just a layer on top of Apache Flink, which uses the batch API of Apache Flink or the stream API. I will talk about this maybe later. And CRADUP, what it provides to you is an abstraction of property graphs on top of this data flow system. So this is the data model that we implement. I will talk about this on the next slide. And we provide a set of graph data flow operators. So one of them is, for example, pattern matching, which we will talk about in detail today. But for those who don't know what Flink is, on the website, they quote, it's an open source streaming data flow engine, and it provides to you data distribution, communication, and fault tolerance. So all the hard parts of parallel and distributed programming are handled by the system. It does this really nice, especially for the streaming use case, but it also has a batch component, which we are currently building on. So what are the basic abstractions that you have in Apache Flink? So one is a data set. A data set is just a distributed collection of data objects, like Pochos, so normal Java objects, or tablets, things like that. And then you have transformations. And transformations are just operations on data sets to produce another data set, like, for example, MAP or reduce. So for example, we have a data set here. A transformation produces a new data set. And a data flow program is just a composition of those transformations. And this is the way how you program in these kind of systems. So you have to translate your problem into a data flow program using those abstractions that the data flow framework provides to you. So what Flink does for you is the parallelization. So it scales out the computation and the data to multiple machines, computes it, and gives you a result. So what are those abstractions? I guess it's a bit small here. It's just categorized in, so this is Flink data set transformations, unary, so which means one data set is input and binary two data sets. And you have the common suspects like MAP and reduce, and these SQL-like things like filter, group by, blah, blah, blah. And we have the binary stuff like join, for example, for a union, co-group. And there are special operators for iterations. So like data iteration or bulk iteration, which allow you to express, yeah, iterative algorithms, graph algorithms, and machine learning algorithms are iterative to express on this framework. We also use it. So Crabboop on the other hand, like I already said, is an abstraction on top of Apache Flink, which provides you not the data set API. You have it underneath, but it provides you already a property graph, abstraction on top of Flink, plus several operators. So the basic data model that we have is a property graph. So we have vertices and directed edges. And we have this concept of so-called logical graphs. So we have not only one graph, but you can have multiple graphs inside your database, which can also overlap. And the concept here is that these graphs are also first-class citizens like vertices and edges, and they can be input and output of operators, of transformations. So this is why they have also identifiers. So they are uniquely identified in the system. You have type labels. Maybe you can't read it, but it's like this is an org. This is a clan. This is an area they live in. It's more doors, so it has attributes. And here we have the hobbits. So we have attributes everywhere and labels. And you can just load this into the system. The abstraction is already there. You don't have to care about how this is mapped to the underlying data set API. So this is what Gradu provides you from the data model point of view. And it's called the extended property graph model because of these logical graphs that you have additionally to the regular property graph model like in Neo4j, for example. So transformations that we provide are not based on data sets, but on logical graphs as input. So a single graph or a collection of graphs as input. We also have unary and binary operators. So for example, what we have here is pattern matching, which we will focus on today. And other things like grouping. It's an OLAP graph, OLAP operator, subgraph, for example. Binary operations, for example, to combine two graphs to a new graph or to compute the overlap between two graphs or the exclusion. Also, graph collection operators. They are kind of relational. So you have a collection of a large collection of graphs. And you want to, for example, filter them so by predicate like selection in relational algebra or also pattern matching on each of these graphs. And there are also these theoretical operators like union of collections or intersection difference and so on. So I want to give you just two examples how these graph data flows look like using the subgraph operator. So the input here is a logical graph. You have different kinds of labels. So the colors are the labels on these vertices and edges. And what you do as a programmer, you just say, okay, I have my logical graph. I load it from HDFS. It's just a method here, but there's a bit more happening. And you say something, okay, mycraft.subcraft. And then you give two predicate. So it's two lambdas, which are just filtering functions for vertices and edges. So what we say here is a vertex is in the output graph. It has a green label and an edge is in the output graph if it has an orange label. And of course, both source and target, vertex of the edge have to be fulfilled this vertex predicate. So in this example, the result would look like this. We have only those vertices that fulfill the predicates and the edges between them that fulfill the predicates. So in the output is again a logical graph, which again can be input for another operator. So this is the data flow on the graph level. That's what a gradoop gives you. And what it also gives you is pattern matching. So we have again a logical graph as input. Then you define a pattern. For example, we are looking for sub-graphs that start from a green vertex with an orange edge to an orange vertex. And then you have just the graph here. You say match. You define a cypher pattern. It's just green, orange, orange. And then the result is a graph collection in that case, which contains all the sub-graphs that match this pattern. And the sub-graphs are of course sub-graphs of this input graph. So this is from a high level point of view how a gradoop works. And now we will focus on really pattern matching. And then we will look at how we do this. Because it's just an example operator, we have multiple operators, but this is a very interesting one. So I will give you an overview, and then Max will give you more details about the implementation. So the steps are really similar to every query processing system. So you have to query. So this is the query from the beginning. We pass the query at first. So we've written our own antler grammar. We could switch to OpenCypher. The problem here is that OpenCypher doesn't support the graph model, but it's just an implementation detail. So then we translate this into object representation of this query. We translate the predicates to a conjunctive normal form, because it's easier to evaluate them in that case. Then we use statistics about, for example, how many red labels are in the search graph, how many blue edges are in the search graph. There are more statistics. And then we use this. We build a cost-based greedy optimizer that takes the query, this representation, the statistics, and produces a query plan. And then using this query plan, every node represents a physical operator which maps again to fling transformations. And the output, if you execute this, is then the collection of found patterns in the graph. And now Max will tell you a bit more about what's happening after this part. So we now have generated this plan, and then we execute this. OK. Yes. So as Martin said, after we have parsed and planned the query, we end up with something like this. This is the query plan for this particular query. And just to give you a short overview of what this actually means, so this is actually a tree. Maybe it doesn't look like this in this kind of visualization. But to execute it, we basically start with the leave node. So for example, we want to find all relationships that match this sub-pattern. So we want to find all relationships that have the type HATES. And so what we basically do is we filter and project the collection of all relationships we have in our graph. And after we've done this for all atomic patterns, we start kind of building up larger patterns, and we do this by joining. So for example, to get this sub-pattern, we join all nodes with the label or with every label org, together with all relationships with type HATES. And so we continuously build up larger patterns by joining together smaller patterns. In the end, this is, for example, this is the last operator. In this plan, we do a filter step where we filter out all non-valid results according to a given predicate, which is in this case making sure that org one and org two aren't the same and plan one and plan two aren't the same. So we throw away all embeddings that are all results that don't hold for this pattern. A slightly special operator is the expanded embedding operator. This is a traversal operator. So in this operator, you don't only expand one hop, but you can expand multiple hops using one operator. This is the star thing you see or might not see over here. So in this case, we specify you want to step at least or do at least one hop, but at most 10. Yes. And just to walk you through some or how this works internally and how this maps to blink. So, oh yeah, every of these operators produces results, intermediate results or the end result, of course. And so what we did is we implemented two main, we have two main data structures. For once, this is the embedding. The embedding stores one particular intermediate result or the end result. So it has two types of, three types of entries. One is the list of ID entries. This stores the mapping of actual nodes or vertices in our graph to nodes or edges in the query graph. Then we have a list of properties. So every node, as you know, in the extended property graph model, every node and edge and even graphs, but that's not relevant here, have properties like names or like Frodo has a name or every Hobbit has a name. And we store this in a separate list. And then we have paths. This is just an implementation detail to make sure that if we do this variable length expand, that we always end up with embeddings of the same size or with the same amount of entries. But it's not too important for now. To say these embeddings, they are kind of dumb. They don't know, they only store things, but they don't know what they actually store. And to make sure we have a mapping, we actually store this knowledge as well. We have also embedding metadata. The reason we split these things is that since we are in a distributed environment, we eventually have to send data over the network. This happens usually when we do a join. And that's quite expensive compared to just accessing RAM or something like this. And so we have to make sure that our data structures we actually send over the network are as small as possible. And since this metadata information is the same for every intermediate result for every step, we can just have one metadata object for every note in the query plan. And this basically stores the mapping of which were variable in our query graph maps to which entry in the embedding. And the same goes for properties. So to have some examples how this actually works under the hood. So we have three layers here. This is like our example layer. This is the actually data we have, our orgs. Then we have a more or less relational view. So kind of what we do internally is we map things to relations. And then you have the same thing as the flink view or how our physical operators in the end map to flink operators. So this is an example for a leaf note. So we want to find all notes that are a hobbit and have the name for the baggings. So we scan through all our notes in the graph, filter out these notes which have the according label and the according name. And what we end up is with a couple of or maybe just one note and all its properties. And this can in a relational view, this would be a table with the ID and the properties of every hobbit named for the baggings. And as a follow-up step, we do a project step because we want to again save some data size. We throw away all properties we don't need anymore because why would we carry them if we never access them again? In this case, we are not really interested in any property of photos or read forward everything. In flink, this maps to a flat map function. Flat map allows us to combine these two steps into one. This is some internal thing of flink, why that's better. We could also do a map or a filter and a follow-up map step. But this would cost us more, it would need more time. So, flat map is basically you have input, you have as input, you have one vertex or in this case a vertex and you can output one or zero or even more but that's not the case for us. You can output a reliable amount of results basically. So, this is a leaf node operator that projects and filter embeddings over it says. This is always the start operator in an acrary plan. Then you have the maybe most important operator is the join. So, this is to join the sub-patterns to larger patterns. So, for example, you have this pattern, you have the orc which is as related to a clan and you have an orc which hates another orc and you basically want to join this together to become a larger pattern. In relational view, this is the same as for relational databases. So, you have two tables basically. In our case, it's two intermediate results and you join them together. We can only just join them because we have some constraints. This is due to how graph databases work. You have to basically check that you don't assign the same node or the same relationship to different nodes in your query graph. So, you don't want to do this orc and this orc to be the same or you don't want this relationship and this relationship. In this case, they would never be the same but in other cases, they could be the same. You don't want to traverse the same relationship twice, at least depending on your settings. But, it's out of scope for this talk maybe. So, we take those two subgraphs and join them together and get one larger graph. You can see that in the example, we don't store this node twice because why would we? It would be a waste of space. And again, this maps to this time a flat join, it's not a flat map because as I just said, some intermediate results wouldn't be valid under the assumption of isomorphism or this relationship or node uniqueness. So, maybe the most interesting operator is the ReviableLength expand. This is basically, in our example, this would be, we have an orc and we want to find Frodo Beggins in one to ten steps. And for this, what we do is we use an iterative expand. Under the hood, this is basically just an iterative join with some, you can do some optimizations which hopefully help you making it faster. So, what we do is we have the left side which is only the orc and then we have a set of nodes relationships and their nodes. And then we kind of join them iteratively and yield every possible result, of course, if it's longer than one hop and shorter than ten hops. So, it's not that we only yield the largest result, but we yield all intermediate results as well. This is what Martin mentioned before. Here we use the Botic iteration. This is kind of a cool thing for streaming processing systems. Spark, for example, doesn't have an iterative operator, Flink does. And so, we use this iteration operator to iterate over the graph and produce our results. And as a last step, we do filtering. So, we had this predicate that orc one and orc two are not allowed to be the same. So, obviously, we have to filter this result and that's quite straightforward. And this maps to the filter operator which is already present in Flink. So, that's quite straightforward. That's basically the inter-operators we have implemented so far, which allows us to run most Cypher queries right now. What we don't have at the moment is a return statement. This is kind of work in progress, but we will eventually have it. And there is some caveats on how to map the projected results to a graph again. That's why we don't have it right now. And what we also don't have is aggregation. But you can do aggregation by just adding, I mean, GraduP itself has operators for aggregation. So, what you could do is you do the pattern matching with the aggregation and do the aggregation as a follow-up. So, but most of the rest of the Cypher standard we support right now. So, now that we have planned our query, this is what we end up with for Flink. So, this is the visualization of what Flink generates. So, we have our query plan and then we translate it to a Flink query plan. And this is for this particular query. This would be the Flink query plan. You don't see much here. That's completely okay. We just wanted to show you that it's quite a lot of steps you have to take in order to produce the result. And this is just a more detailed view. So, you have basically your data source, then you do a set of filter and project so you get all the vertices and edges you want to work with and then you join them over and over. And in the end, you end up with one dataset which contains your results. And then you can, for example, write them into HDFS or wherever you want to store them. So, what we want to do, what we would like to do is that we want to integrate the Cypher technology comparatively toolkit. This is a set of cucumber tests which allow you, if you implement them, they tell you if you have implemented OpenCypher correctly, hopefully. We haven't really done any, like, broader checks now if what we produce is correct. So, this would be perfect. We want to do benchmarking. We have some preliminary evaluations, but we should go into more detail for this. And for this, we want to use the LDBC social network benchmark. This is a benchmark specifically designed for graph databases or graph systems. And we can do a couple of optimizations. So, right now, we use a greedy planner. We could use a dynamic programming planner, hopefully yielding better tree plans and faster execution times. We want to improve our cost model. So, in order to be able to produce a plan, you have to estimate how costly each step in the plan would be. Right now, we have a very basic, very simple model based on probabilities, but this doesn't scale at least, not really if you have this amount of joints. And, yeah, you can do reuse of intermediate steps. Basically, for every org, we have two orgs in our pattern. Right now, for every org, we would filter the list of all nodes in our graph, but actually, that's the same. I mean, since we only filter for the label org, we yield the same result on both sides, so we could reuse it, basically, which we don't do yet. And another problem in the distributed environment is partitioning. So, if you have vertices which are with a high out-degree or in-degree, so in a social network, for example, people with many, many friends, they increase the computing time on the particular node which is in charge for handling this node. They increase the computing time unevenly to every other node in the cluster, for example, and so we have Datasq, which means that some workers take way more time to produce a result than other workers. And it's a hard thing to partition the nodes over the cluster evenly so that you end up with an even computing time. And, yeah, of course, we want to support the rest of the Cypher features and maybe even introduce more Cypher features like regular path queries or support for this extended property graph model. And here's where you can come into the big picture because we'd like to, if you are interested in this and if you think that's cool, then it's open source so you can work on this with us together and, yeah, just contribute if you like to and if you think that's a cool thing. Yes. Thank you. Thank you.
 Hello everyone, my name is Marcel.  I'm a PhD student at the University of Lysus.  This is Max, he's a master student who's working with me together  and also at the University of Lysus in the future.  And for me and my crew, we are working on a group.  I will explain what it is, it's based on questions.  And we will show you how we implement side power inside the system.  So for the talk, we thought this was a motivation.  Then we will show you what we do for those who don't know what are the changes  and those who don't know what kind of this.  Then we talk about how side power is realized and implemented in the system  and therefore shown in the picture.  So let's start with some motivation.  So at first, when we talk about graphs, the typical example are social networks  and also the example that we choose.  In that case, it's the social network of words.  So each vertex here we can change the word  and the relations between them is the HH, so a H other.  And smart analysts would ask something like,  who are the closest enemies of each order?  So that's the typical question in different fields of the world.  And if you want to express such a question in the system,  you could do it for example with Neo4j, side power  and then you have it very like this.  So we have a mesh part, there is an org,  we create another org and there are one or two of them.  So one or two HHs.  Okay, let's have a look at this.  How this one would be looking like in plain jelly,  which is the graph-presetting frame on top of the patch of the game.  So this is the same.  It could be nothing but it's expected that it will happen.  Because a lot of quotes express the same thing.  It's even very soon in.  So what's happening here is the messaging between those two.  Here we have something like the H-fabers,  like we've heard about it before,  and here the two means that we get the right twice.  So there are a lot of quotes that express the same thing  and just express it with the size of the screen.  And another problem that we have here  is typically these graphs are not homogeneous like this one.  So we have different kinds of both,  so there's different kinds of action.  For example, we have plans and we have the hobbits, for example,  and we have not only the H-relationships,  we have the plan leader, the plan leader,  and the node relationship between the hobbits, for example.  And now the enders would ask something like  which plan leaders hate each other  and one of them knows Kono over one to ten options.  So that's also a really good question.  Let's look at this how we would say this in the cycle.  So this would be the ferry.  This would be the ferry.  So we have the plan.  We have the leader of this plan,  this leader against number of which is also a plan leader,  and the other plan leader knows Kono in one to ten options.  So we have some predicates here.  Okay, how would you express this on a craft processing framework?  You can try it, but no.  So it's kind of our motivation,  so we want to make craft queries  or craft queries especially  usable on those kind of frameworks.  So this could be a data flow framework  like a project link, for example, such as Spark.  So to be honest, of course,  each of those systems has a specific use case  or they are at some point.  So we have here two dimensions,  more complex than this, but for this one it's enough.  So we have the ease of use, which is like things of how easy  is to express the problem in a system,  and we have data volume.  So it's about scalability, using more machines to compute.  So on the one hand of the spectrum,  we have craft database like for a company of which age.  So you have this declarative languages,  easy to express the problem,  visualization, a lot of tooling, community, and so on.  And on the other hand of this spectrum,  you have craft processing systems.  The main one is Apache G-Raf, which is really dedicated  to compute craft algorithm on this particular cluster,  so a shared-computer cluster, but only for this, nothing else.  Then we have a bit more extraction here.  Those are the craft data flow systems.  So like for example, it's also a craft processing system,  but on top of a data flow system like Apache Spark  or Apache Flink,  CraftX for example on Spark, Jelly on Flink.  And now you see there's a free area in the right upper corner,  and this is where we position our research project,  which is called CRADUP,  and where we want to combine the expressiveness of craft databases,  not as much as craft databases, but a little bit less,  but also the scalability.  So that's kind of the goal of the project.  And I will give you a short overview about the project  before we dive into the Cypher implementation.  So we built on the ShartNothing cluster like I already said.  So we use Hadoop and HDFS.  It's our data source, data sync, so that's where we store our crafts.  We use Apache Flink as a distributed operator execution layer.  So this is where we implement actually our craft operators.  And CRADUP itself is just a layer on top of Apache Flink,  which uses the batch API of Apache Flink or the stream API.  I will talk about this maybe later.  And CRADUP, what it provides to you is an abstraction of property graphs  on top of this data flow system.  So this is the data model that we implement.  I will talk about this on the next slide.  And we provide a set of graph data flow operators.  So one of them is, for example, pattern matching,  which we will talk about in detail today.  But for those who don't know what Flink is, on the website, they quote,  it's an open source streaming data flow engine,  and it provides to you data distribution, communication, and fault tolerance.  So all the hard parts of parallel and distributed programming are handled by the system.  It does this really nice, especially for the streaming use case,  but it also has a batch component, which we are currently building on.  So what are the basic abstractions that you have in Apache Flink?  So one is a data set.  A data set is just a distributed collection of data objects,  like Pochos, so normal Java objects, or tablets, things like that.  And then you have transformations.  And transformations are just operations on data sets to produce another data set,  like, for example, MAP or reduce.  So for example, we have a data set here.  A transformation produces a new data set.  And a data flow program is just a composition of those transformations.  And this is the way how you program in these kind of systems.  So you have to translate your problem into a data flow program  using those abstractions that the data flow framework provides to you.  So what Flink does for you is the parallelization.  So it scales out the computation and the data to multiple machines,  computes it, and gives you a result.  So what are those abstractions?  I guess it's a bit small here.  It's just categorized in, so this is Flink data set transformations,  unary, so which means one data set is input and binary two data sets.  And you have the common suspects like MAP and reduce,  and these SQL-like things like filter, group by, blah, blah, blah.  And we have the binary stuff like join, for example, for a union, co-group.  And there are special operators for iterations.  So like data iteration or bulk iteration,  which allow you to express, yeah, iterative algorithms,  graph algorithms, and machine learning algorithms are iterative  to express on this framework.  We also use it.  So Crabboop on the other hand, like I already said,  is an abstraction on top of Apache Flink,  which provides you not the data set API.  You have it underneath, but it provides you already a property graph,  abstraction on top of Flink, plus several operators.  So the basic data model that we have is a property graph.  So we have vertices and directed edges.  And we have this concept of so-called logical graphs.  So we have not only one graph, but you can have multiple graphs  inside your database, which can also overlap.  And the concept here is that these graphs are also first-class citizens  like vertices and edges, and they can be input and output of operators,  of transformations.  So this is why they have also identifiers.  So they are uniquely identified in the system.  You have type labels.  Maybe you can't read it, but it's like this is an org.  This is a clan.  This is an area they live in.  It's more doors, so it has attributes.  And here we have the hobbits.  So we have attributes everywhere and labels.  And you can just load this into the system.  The abstraction is already there.  You don't have to care about how this is mapped to the underlying data set API.  So this is what Gradu provides you from the data model point of view.  And it's called the extended property graph model because of these logical  graphs that you have additionally to the regular property graph model  like in Neo4j, for example.  So transformations that we provide are not based on data sets,  but on logical graphs as input.  So a single graph or a collection of graphs as input.  We also have unary and binary operators.  So for example, what we have here is pattern matching,  which we will focus on today.  And other things like grouping.  It's an OLAP graph, OLAP operator, subgraph, for example.  Binary operations, for example, to combine two graphs to a new graph  or to compute the overlap between two graphs or the exclusion.  Also, graph collection operators.  They are kind of relational.  So you have a collection of a large collection of graphs.  And you want to, for example, filter them so by predicate like selection  in relational algebra or also pattern matching on each of these graphs.  And there are also these theoretical operators like union of collections  or intersection difference and so on.  So I want to give you just two examples how these graph data flows look like  using the subgraph operator.  So the input here is a logical graph.  You have different kinds of labels.  So the colors are the labels on these vertices and edges.  And what you do as a programmer, you just say,  okay, I have my logical graph.  I load it from HDFS.  It's just a method here, but there's a bit more happening.  And you say something, okay, mycraft.subcraft.  And then you give two predicate.  So it's two lambdas, which are just filtering functions for vertices and edges.  So what we say here is a vertex is in the output graph.  It has a green label and an edge is in the output graph  if it has an orange label.  And of course, both source and target, vertex of the edge have to be  fulfilled this vertex predicate.  So in this example, the result would look like this.  We have only those vertices that fulfill the predicates  and the edges between them that fulfill the predicates.  So in the output is again a logical graph,  which again can be input for another operator.  So this is the data flow on the graph level.  That's what a gradoop gives you.  And what it also gives you is pattern matching.  So we have again a logical graph as input.  Then you define a pattern.  For example, we are looking for sub-graphs that start from a green vertex  with an orange edge to an orange vertex.  And then you have just the graph here.  You say match.  You define a cypher pattern.  It's just green, orange, orange.  And then the result is a graph collection in that case,  which contains all the sub-graphs that match this pattern.  And the sub-graphs are of course sub-graphs of this input graph.  So this is from a high level point of view how a gradoop works.  And now we will focus on really pattern matching.  And then we will look at how we do this.  Because it's just an example operator,  we have multiple operators, but this is a very interesting one.  So I will give you an overview,  and then Max will give you more details about the implementation.  So the steps are really similar to every query processing system.  So you have to query.  So this is the query from the beginning.  We pass the query at first.  So we've written our own antler grammar.  We could switch to OpenCypher.  The problem here is that OpenCypher doesn't support  the graph model, but it's just an implementation detail.  So then we translate this into object representation of this query.  We translate the predicates to a conjunctive normal form,  because it's easier to evaluate them in that case.  Then we use statistics about, for example,  how many red labels are in the search graph,  how many blue edges are in the search graph.  There are more statistics.  And then we use this.  We build a cost-based greedy optimizer that takes the query,  this representation, the statistics, and produces a query plan.  And then using this query plan,  every node represents a physical operator  which maps again to fling transformations.  And the output, if you execute this,  is then the collection of found patterns in the graph.  And now Max will tell you a bit more about what's happening  after this part.  So we now have generated this plan,  and then we execute this.  OK.  Yes.  So as Martin said, after we have parsed and planned the query,  we end up with something like this.  This is the query plan for this particular query.  And just to give you a short overview of what this actually means,  so this is actually a tree.  Maybe it doesn't look like this in this kind of visualization.  But to execute it, we basically start with the leave node.  So for example, we want to find all relationships  that match this sub-pattern.  So we want to find all relationships that have the type HATES.  And so what we basically do is we filter and project  the collection of all relationships we have in our graph.  And after we've done this for all atomic patterns,  we start kind of building up larger patterns,  and we do this by joining.  So for example, to get this sub-pattern,  we join all nodes with the label or with every label org,  together with all relationships with type HATES.  And so we continuously build up larger patterns  by joining together smaller patterns.  In the end, this is, for example, this is the last operator.  In this plan, we do a filter step where we filter out  all non-valid results according to a given predicate,  which is in this case making sure that org one and org two  aren't the same and plan one and plan two aren't the same.  So we throw away all embeddings that are all results  that don't hold for this pattern.  A slightly special operator is the expanded embedding operator.  This is a traversal operator.  So in this operator, you don't only expand one hop,  but you can expand multiple hops using one operator.  This is the star thing you see or might not see over here.  So in this case, we specify you want to step at least  or do at least one hop, but at most 10.  Yes.  And just to walk you through some or how this works internally  and how this maps to blink.  So, oh yeah, every of these operators  produces results, intermediate results or the end result,  of course.  And so what we did is we implemented two main,  we have two main data structures.  For once, this is the embedding.  The embedding stores one particular intermediate result  or the end result.  So it has two types of, three types of entries.  One is the list of ID entries.  This stores the mapping of actual nodes or vertices  in our graph to nodes or edges in the query graph.  Then we have a list of properties.  So every node, as you know, in the extended property graph model,  every node and edge and even graphs, but that's not relevant here,  have properties like names or like Frodo has a name  or every Hobbit has a name.  And we store this in a separate list.  And then we have paths.  This is just an implementation detail to make sure that if we do  this variable length expand, that we always end up with  embeddings of the same size or with the same amount of entries.  But it's not too important for now.  To say these embeddings, they are kind of dumb.  They don't know, they only store things,  but they don't know what they actually store.  And to make sure we have a mapping,  we actually store this knowledge as well.  We have also embedding metadata.  The reason we split these things is that since we are in a  distributed environment, we eventually have to send data  over the network.  This happens usually when we do a join.  And that's quite expensive compared to just accessing RAM  or something like this.  And so we have to make sure that our data structures we actually  send over the network are as small as possible.  And since this metadata information is the same for every  intermediate result for every step,  we can just have one metadata object for every note in the  query plan.  And this basically stores the mapping of which were  variable in our query graph maps to which entry in the  embedding.  And the same goes for properties.  So to have some examples how this actually works under the  hood.  So we have three layers here.  This is like our example layer.  This is the actually data we have, our orgs.  Then we have a more or less relational view.  So kind of what we do internally is we map things to  relations.  And then you have the same thing as the flink view or how our  physical operators in the end map to flink operators.  So this is an example for a leaf note.  So we want to find all notes that are a hobbit and have the  name for the baggings.  So we scan through all our notes in the graph, filter out these  notes which have the according label and the according name.  And what we end up is with a couple of or maybe just one note  and all its properties.  And this can in a relational view, this would be a table with  the ID and the properties of every hobbit named for the baggings.  And as a follow-up step, we do a project step because we want to  again save some data size.  We throw away all properties we don't need anymore because why  would we carry them if we never access them again?  In this case, we are not really interested in any property of  photos or read forward everything.  In flink, this maps to a flat map function.  Flat map allows us to combine these two steps into one.  This is some internal thing of flink, why that's better.  We could also do a map or a filter and a follow-up map step.  But this would cost us more, it would need more time.  So, flat map is basically you have input, you have as input,  you have one vertex or in this case a vertex and you can output  one or zero or even more but that's not the case for us.  You can output a reliable amount of results basically.  So, this is a leaf node operator that projects and filter  embeddings over it says.  This is always the start operator in an acrary plan.  Then you have the maybe most important operator is the join.  So, this is to join the sub-patterns to larger patterns.  So, for example, you have this pattern, you have the orc which is  as related to a clan and you have an orc which hates another orc  and you basically want to join this together to become a larger pattern.  In relational view, this is the same as for relational databases.  So, you have two tables basically.  In our case, it's two intermediate results and you join them together.  We can only just join them because we have some constraints.  This is due to how graph databases work.  You have to basically check that you don't assign the same node  or the same relationship to different nodes in your query graph.  So, you don't want to do this orc and this orc to be the same  or you don't want this relationship and this relationship.  In this case, they would never be the same but in other cases,  they could be the same.  You don't want to traverse the same relationship twice,  at least depending on your settings.  But, it's out of scope for this talk maybe.  So, we take those two subgraphs and join them together  and get one larger graph.  You can see that in the example, we don't store this node twice  because why would we?  It would be a waste of space.  And again, this maps to this time a flat join, it's not a flat map  because as I just said, some intermediate results wouldn't be  valid under the assumption of isomorphism or this relationship  or node uniqueness.  So, maybe the most interesting operator is the  ReviableLength expand.  This is basically, in our example, this would be, we have an orc  and we want to find Frodo Beggins in one to ten steps.  And for this, what we do is we use an iterative expand.  Under the hood, this is basically just an iterative join  with some, you can do some optimizations which hopefully  help you making it faster.  So, what we do is we have the left side which is only the orc  and then we have a set of nodes relationships and their nodes.  And then we kind of join them iteratively and yield every  possible result, of course, if it's longer than one hop  and shorter than ten hops.  So, it's not that we only yield the largest result, but we yield  all intermediate results as well.  This is what Martin mentioned before.  Here we use the Botic iteration.  This is kind of a cool thing for streaming processing systems.  Spark, for example, doesn't have an iterative operator,  Flink does.  And so, we use this iteration operator to iterate over the graph  and produce our results.  And as a last step, we do filtering.  So, we had this predicate that orc one and orc two are not  allowed to be the same.  So, obviously, we have to filter this result and that's quite  straightforward.  And this maps to the filter operator which is already present  in Flink.  So, that's quite straightforward.  That's basically the inter-operators we have implemented  so far, which allows us to run most Cypher queries right now.  What we don't have at the moment is a return statement.  This is kind of work in progress, but we will eventually have it.  And there is some caveats on how to map the projected results  to a graph again.  That's why we don't have it right now.  And what we also don't have is aggregation.  But you can do aggregation by just adding, I mean,  GraduP itself has operators for aggregation.  So, what you could do is you do the pattern matching  with the aggregation and do the aggregation as a follow-up.  So, but most of the rest of the Cypher standard we support right  now.  So, now that we have planned our query, this is what we end up  with for Flink.  So, this is the visualization of what Flink generates.  So, we have our query plan and then we translate it to a Flink  query plan.  And this is for this particular query.  This would be the Flink query plan.  You don't see much here.  That's completely okay.  We just wanted to show you that it's quite a lot of steps you  have to take in order to produce the result.  And this is just a more detailed view.  So, you have basically your data source, then you do a set of filter  and project so you get all the vertices and edges you want to  work with and then you join them over and over.  And in the end, you end up with one dataset which contains  your results.  And then you can, for example, write them into HDFS or wherever  you want to store them.  So, what we want to do, what we would like to do is that we want  to integrate the Cypher technology comparatively toolkit.  This is a set of cucumber tests which allow you, if you implement  them, they tell you if you have implemented OpenCypher  correctly, hopefully.  We haven't really done any, like, broader checks now if what we  produce is correct.  So, this would be perfect.  We want to do benchmarking.  We have some preliminary evaluations, but we should go into  more detail for this.  And for this, we want to use the LDBC social network benchmark.  This is a benchmark specifically designed for graph databases  or graph systems.  And we can do a couple of optimizations.  So, right now, we use a greedy planner.  We could use a dynamic programming planner, hopefully yielding  better tree plans and faster execution times.  We want to improve our cost model.  So, in order to be able to produce a plan, you have to  estimate how costly each step in the plan would be.  Right now, we have a very basic, very simple model based on  probabilities, but this doesn't scale at least, not really if  you have this amount of joints.  And, yeah, you can do reuse of intermediate steps.  Basically, for every org, we have two orgs in our pattern.  Right now, for every org, we would filter the list of all  nodes in our graph, but actually, that's the same.  I mean, since we only filter for the label org, we yield the  same result on both sides, so we could reuse it, basically,  which we don't do yet.  And another problem in the distributed environment is  partitioning.  So, if you have vertices which are with a high out-degree or  in-degree, so in a social network, for example, people with  many, many friends, they increase the computing time on  the particular node which is in charge for handling this node.  They increase the computing time unevenly to every other node  in the cluster, for example, and so we have Datasq, which  means that some workers take way more time to produce a  result than other workers.  And it's a hard thing to partition the nodes over the  cluster evenly so that you end up with an even computing  time.  And, yeah, of course, we want to support the rest of the  Cypher features and maybe even introduce more Cypher features  like regular path queries or support for this extended  property graph model.  And here's where you can come into the big picture because  we'd like to, if you are interested in this and if you  think that's cool, then it's open source so you can work on  this with us together and, yeah, just contribute if you  like to and if you think that's a cool thing.  Yes.  Thank you.  Thank you.
_Cypher__ON__ApacheFlink__USING__Gradoop_0_131.png _Cypher__ON__ApacheFlink__USING__Gradoop_132_983.png _Cypher__ON__ApacheFlink__USING__Gradoop_984_1187.png _Cypher__ON__ApacheFlink__USING__Gradoop_1188_1475.png _Cypher__ON__ApacheFlink__USING__Gradoop_1476_1703.png _Cypher__ON__ApacheFlink__USING__Gradoop_1704_2063.png _Cypher__ON__ApacheFlink__USING__Gradoop_2064_2147.png _Cypher__ON__ApacheFlink__USING__Gradoop_2148_2219.png _Cypher__ON__ApacheFlink__USING__Gradoop_2220_2375.png _Cypher__ON__ApacheFlink__USING__Gradoop_2376_2423.png _Cypher__ON__ApacheFlink__USING__Gradoop_2424_2471.png _Cypher__ON__ApacheFlink__USING__Gradoop_2472_2867.png _Cypher__ON__ApacheFlink__USING__Gradoop_2868_3011.png _Cypher__ON__ApacheFlink__USING__Gradoop_3012_3311.png _Cypher__ON__ApacheFlink__USING__Gradoop_3312_3515.png _Cypher__ON__ApacheFlink__USING__Gradoop_3516_4175.png _Cypher__ON__ApacheFlink__USING__Gradoop_4176_4223.png _Cypher__ON__ApacheFlink__USING__Gradoop_4224_4607.png _Cypher__ON__ApacheFlink__USING__Gradoop_4608_4727.png _Cypher__ON__ApacheFlink__USING__Gradoop_4728_5699.png _Cypher__ON__ApacheFlink__USING__Gradoop_5700_6743.png _Cypher__ON__ApacheFlink__USING__Gradoop_6744_6875.png _Cypher__ON__ApacheFlink__USING__Gradoop_6876_7715.png _Cypher__ON__ApacheFlink__USING__Gradoop_7716_7955.png _Cypher__ON__ApacheFlink__USING__Gradoop_7956_9611.png _Cypher__ON__ApacheFlink__USING__Gradoop_9612_17279.png _Cypher__ON__ApacheFlink__USING__Gradoop_17280_17327.png _Cypher__ON__ApacheFlink__USING__Gradoop_17328_18083.png _Cypher__ON__ApacheFlink__USING__Gradoop_18084_22355.png _Cypher__ON__ApacheFlink__USING__Gradoop_22356_25583.png _Cypher__ON__ApacheFlink__USING__Gradoop_25584_29231.png _Cypher__ON__ApacheFlink__USING__Gradoop_29232_32003.png _Cypher__ON__ApacheFlink__USING__Gradoop_32004_34367.png _Cypher__ON__ApacheFlink__USING__Gradoop_34368_36431.png _Cypher__ON__ApacheFlink__USING__Gradoop_36432_42863.png _Cypher__ON__ApacheFlink__USING__Gradoop_42864_43031.png _Cypher__ON__ApacheFlink__USING__Gradoop_43032_43079.png _Cypher__ON__ApacheFlink__USING__Gradoop_43080_43127.png _Cypher__ON__ApacheFlink__USING__Gradoop_43128_43145.png
Hello everyone, my name is Marcel. I'm a PhD student at the University of Lysus. This is Max, he's a master student who's working with me together and also at the University of Lysus in the future. And for me and my crew, we are working on a group. I will explain what it is, it's based on questions. And we will show you how we implement side power inside the system. So for the talk, we thought this was a motivation. Then we will show you what we do for those who don't know what are the changes and those who don't know what kind of this. Then we talk about how side power is realized and implemented in the system and therefore shown in the picture. So let's start with some motivation. So at first, when we talk about graphs, the typical example are social networks and also the example that we choose. In that case, it's the social network of words. So each vertex here we can change the word and the relations between them is the HH, so a H other. And smart analysts would ask something like, who are the closest enemies of each order? So that's the typical question in different fields of the world. And if you want to express such a question in the system, you could do it for example with Neo4j, side power and then you have it very like this. So we have a mesh part, there is an org, we create another org and there are one or two of them. So one or two HHs. Okay, let's have a look at this. How this one would be looking like in plain jelly, which is the graph-presetting frame on top of the patch of the game. So this is the same. It could be nothing but it's expected that it will happen. Because a lot of quotes express the same thing. It's even very soon in. So what's happening here is the messaging between those two. Here we have something like the H-fabers, like we've heard about it before, and here the two means that we get the right twice. So there are a lot of quotes that express the same thing and just express it with the size of the screen. And another problem that we have here is typically these graphs are not homogeneous like this one. So we have different kinds of both, so there's different kinds of action. For example, we have plans and we have the hobbits, for example, and we have not only the H-relationships, we have the plan leader, the plan leader, and the node relationship between the hobbits, for example. And now the enders would ask something like which plan leaders hate each other and one of them knows Kono over one to ten options. So that's also a really good question. Let's look at this how we would say this in the cycle. So this would be the ferry. This would be the ferry. So we have the plan. We have the leader of this plan, this leader against number of which is also a plan leader, and the other plan leader knows Kono in one to ten options. So we have some predicates here. Okay, how would you express this on a craft processing framework? You can try it, but no. So it's kind of our motivation, so we want to make craft queries or craft queries especially usable on those kind of frameworks. So this could be a data flow framework like a project link, for example, such as Spark. So to be honest, of course, each of those systems has a specific use case or they are at some point. So we have here two dimensions, more complex than this, but for this one it's enough. So we have the ease of use, which is like things of how easy is to express the problem in a system, and we have data volume. So it's about scalability, using more machines to compute. So on the one hand of the spectrum, we have craft database like for a company of which age. So you have this declarative languages, easy to express the problem, visualization, a lot of tooling, community, and so on. And on the other hand of this spectrum, you have craft processing systems. The main one is Apache G-Raf, which is really dedicated to compute craft algorithm on this particular cluster, so a shared-computer cluster, but only for this, nothing else. Then we have a bit more extraction here. Those are the craft data flow systems. So like for example, it's also a craft processing system, but on top of a data flow system like Apache Spark or Apache Flink, CraftX for example on Spark, Jelly on Flink. And now you see there's a free area in the right upper corner, and this is where we position our research project, which is called CRADUP, and where we want to combine the expressiveness of craft databases, not as much as craft databases, but a little bit less, but also the scalability. So that's kind of the goal of the project. And I will give you a short overview about the project before we dive into the Cypher implementation. So we built on the ShartNothing cluster like I already said. So we use Hadoop and HDFS. It's our data source, data sync, so that's where we store our crafts. We use Apache Flink as a distributed operator execution layer. So this is where we implement actually our craft operators. And CRADUP itself is just a layer on top of Apache Flink, which uses the batch API of Apache Flink or the stream API. I will talk about this maybe later. And CRADUP, what it provides to you is an abstraction of property graphs on top of this data flow system. So this is the data model that we implement. I will talk about this on the next slide. And we provide a set of graph data flow operators. So one of them is, for example, pattern matching, which we will talk about in detail today. But for those who don't know what Flink is, on the website, they quote, it's an open source streaming data flow engine, and it provides to you data distribution, communication, and fault tolerance. So all the hard parts of parallel and distributed programming are handled by the system. It does this really nice, especially for the streaming use case, but it also has a batch component, which we are currently building on. So what are the basic abstractions that you have in Apache Flink? So one is a data set. A data set is just a distributed collection of data objects, like Pochos, so normal Java objects, or tablets, things like that. And then you have transformations. And transformations are just operations on data sets to produce another data set, like, for example, MAP or reduce. So for example, we have a data set here. A transformation produces a new data set. And a data flow program is just a composition of those transformations. And this is the way how you program in these kind of systems. So you have to translate your problem into a data flow program using those abstractions that the data flow framework provides to you. So what Flink does for you is the parallelization. So it scales out the computation and the data to multiple machines, computes it, and gives you a result. So what are those abstractions? I guess it's a bit small here. It's just categorized in, so this is Flink data set transformations, unary, so which means one data set is input and binary two data sets. And you have the common suspects like MAP and reduce, and these SQL-like things like filter, group by, blah, blah, blah. And we have the binary stuff like join, for example, for a union, co-group. And there are special operators for iterations. So like data iteration or bulk iteration, which allow you to express, yeah, iterative algorithms, graph algorithms, and machine learning algorithms are iterative to express on this framework. We also use it. So Crabboop on the other hand, like I already said, is an abstraction on top of Apache Flink, which provides you not the data set API. You have it underneath, but it provides you already a property graph, abstraction on top of Flink, plus several operators. So the basic data model that we have is a property graph. So we have vertices and directed edges. And we have this concept of so-called logical graphs. So we have not only one graph, but you can have multiple graphs inside your database, which can also overlap. And the concept here is that these graphs are also first-class citizens like vertices and edges, and they can be input and output of operators, of transformations. So this is why they have also identifiers. So they are uniquely identified in the system. You have type labels. Maybe you can't read it, but it's like this is an org. This is a clan. This is an area they live in. It's more doors, so it has attributes. And here we have the hobbits. So we have attributes everywhere and labels. And you can just load this into the system. The abstraction is already there. You don't have to care about how this is mapped to the underlying data set API. So this is what Gradu provides you from the data model point of view. And it's called the extended property graph model because of these logical graphs that you have additionally to the regular property graph model like in Neo4j, for example. So transformations that we provide are not based on data sets, but on logical graphs as input. So a single graph or a collection of graphs as input. We also have unary and binary operators. So for example, what we have here is pattern matching, which we will focus on today. And other things like grouping. It's an OLAP graph, OLAP operator, subgraph, for example. Binary operations, for example, to combine two graphs to a new graph or to compute the overlap between two graphs or the exclusion. Also, graph collection operators. They are kind of relational. So you have a collection of a large collection of graphs. And you want to, for example, filter them so by predicate like selection in relational algebra or also pattern matching on each of these graphs. And there are also these theoretical operators like union of collections or intersection difference and so on. So I want to give you just two examples how these graph data flows look like using the subgraph operator. So the input here is a logical graph. You have different kinds of labels. So the colors are the labels on these vertices and edges. And what you do as a programmer, you just say, okay, I have my logical graph. I load it from HDFS. It's just a method here, but there's a bit more happening. And you say something, okay, mycraft.subcraft. And then you give two predicate. So it's two lambdas, which are just filtering functions for vertices and edges. So what we say here is a vertex is in the output graph. It has a green label and an edge is in the output graph if it has an orange label. And of course, both source and target, vertex of the edge have to be fulfilled this vertex predicate. So in this example, the result would look like this. We have only those vertices that fulfill the predicates and the edges between them that fulfill the predicates. So in the output is again a logical graph, which again can be input for another operator. So this is the data flow on the graph level. That's what a gradoop gives you. And what it also gives you is pattern matching. So we have again a logical graph as input. Then you define a pattern. For example, we are looking for sub-graphs that start from a green vertex with an orange edge to an orange vertex. And then you have just the graph here. You say match. You define a cypher pattern. It's just green, orange, orange. And then the result is a graph collection in that case, which contains all the sub-graphs that match this pattern. And the sub-graphs are of course sub-graphs of this input graph. So this is from a high level point of view how a gradoop works. And now we will focus on really pattern matching. And then we will look at how we do this. Because it's just an example operator, we have multiple operators, but this is a very interesting one. So I will give you an overview, and then Max will give you more details about the implementation. So the steps are really similar to every query processing system. So you have to query. So this is the query from the beginning. We pass the query at first. So we've written our own antler grammar. We could switch to OpenCypher. The problem here is that OpenCypher doesn't support the graph model, but it's just an implementation detail. So then we translate this into object representation of this query. We translate the predicates to a conjunctive normal form, because it's easier to evaluate them in that case. Then we use statistics about, for example, how many red labels are in the search graph, how many blue edges are in the search graph. There are more statistics. And then we use this. We build a cost-based greedy optimizer that takes the query, this representation, the statistics, and produces a query plan. And then using this query plan, every node represents a physical operator which maps again to fling transformations. And the output, if you execute this, is then the collection of found patterns in the graph. And now Max will tell you a bit more about what's happening after this part. So we now have generated this plan, and then we execute this. OK. Yes. So as Martin said, after we have parsed and planned the query, we end up with something like this. This is the query plan for this particular query. And just to give you a short overview of what this actually means, so this is actually a tree. Maybe it doesn't look like this in this kind of visualization. But to execute it, we basically start with the leave node. So for example, we want to find all relationships that match this sub-pattern. So we want to find all relationships that have the type HATES. And so what we basically do is we filter and project the collection of all relationships we have in our graph. And after we've done this for all atomic patterns, we start kind of building up larger patterns, and we do this by joining. So for example, to get this sub-pattern, we join all nodes with the label or with every label org, together with all relationships with type HATES. And so we continuously build up larger patterns by joining together smaller patterns. In the end, this is, for example, this is the last operator. In this plan, we do a filter step where we filter out all non-valid results according to a given predicate, which is in this case making sure that org one and org two aren't the same and plan one and plan two aren't the same. So we throw away all embeddings that are all results that don't hold for this pattern. A slightly special operator is the expanded embedding operator. This is a traversal operator. So in this operator, you don't only expand one hop, but you can expand multiple hops using one operator. This is the star thing you see or might not see over here. So in this case, we specify you want to step at least or do at least one hop, but at most 10. Yes. And just to walk you through some or how this works internally and how this maps to blink. So, oh yeah, every of these operators produces results, intermediate results or the end result, of course. And so what we did is we implemented two main, we have two main data structures. For once, this is the embedding. The embedding stores one particular intermediate result or the end result. So it has two types of, three types of entries. One is the list of ID entries. This stores the mapping of actual nodes or vertices in our graph to nodes or edges in the query graph. Then we have a list of properties. So every node, as you know, in the extended property graph model, every node and edge and even graphs, but that's not relevant here, have properties like names or like Frodo has a name or every Hobbit has a name. And we store this in a separate list. And then we have paths. This is just an implementation detail to make sure that if we do this variable length expand, that we always end up with embeddings of the same size or with the same amount of entries. But it's not too important for now. To say these embeddings, they are kind of dumb. They don't know, they only store things, but they don't know what they actually store. And to make sure we have a mapping, we actually store this knowledge as well. We have also embedding metadata. The reason we split these things is that since we are in a distributed environment, we eventually have to send data over the network. This happens usually when we do a join. And that's quite expensive compared to just accessing RAM or something like this. And so we have to make sure that our data structures we actually send over the network are as small as possible. And since this metadata information is the same for every intermediate result for every step, we can just have one metadata object for every note in the query plan. And this basically stores the mapping of which were variable in our query graph maps to which entry in the embedding. And the same goes for properties. So to have some examples how this actually works under the hood. So we have three layers here. This is like our example layer. This is the actually data we have, our orgs. Then we have a more or less relational view. So kind of what we do internally is we map things to relations. And then you have the same thing as the flink view or how our physical operators in the end map to flink operators. So this is an example for a leaf note. So we want to find all notes that are a hobbit and have the name for the baggings. So we scan through all our notes in the graph, filter out these notes which have the according label and the according name. And what we end up is with a couple of or maybe just one note and all its properties. And this can in a relational view, this would be a table with the ID and the properties of every hobbit named for the baggings. And as a follow-up step, we do a project step because we want to again save some data size. We throw away all properties we don't need anymore because why would we carry them if we never access them again? In this case, we are not really interested in any property of photos or read forward everything. In flink, this maps to a flat map function. Flat map allows us to combine these two steps into one. This is some internal thing of flink, why that's better. We could also do a map or a filter and a follow-up map step. But this would cost us more, it would need more time. So, flat map is basically you have input, you have as input, you have one vertex or in this case a vertex and you can output one or zero or even more but that's not the case for us. You can output a reliable amount of results basically. So, this is a leaf node operator that projects and filter embeddings over it says. This is always the start operator in an acrary plan. Then you have the maybe most important operator is the join. So, this is to join the sub-patterns to larger patterns. So, for example, you have this pattern, you have the orc which is as related to a clan and you have an orc which hates another orc and you basically want to join this together to become a larger pattern. In relational view, this is the same as for relational databases. So, you have two tables basically. In our case, it's two intermediate results and you join them together. We can only just join them because we have some constraints. This is due to how graph databases work. You have to basically check that you don't assign the same node or the same relationship to different nodes in your query graph. So, you don't want to do this orc and this orc to be the same or you don't want this relationship and this relationship. In this case, they would never be the same but in other cases, they could be the same. You don't want to traverse the same relationship twice, at least depending on your settings. But, it's out of scope for this talk maybe. So, we take those two subgraphs and join them together and get one larger graph. You can see that in the example, we don't store this node twice because why would we? It would be a waste of space. And again, this maps to this time a flat join, it's not a flat map because as I just said, some intermediate results wouldn't be valid under the assumption of isomorphism or this relationship or node uniqueness. So, maybe the most interesting operator is the ReviableLength expand. This is basically, in our example, this would be, we have an orc and we want to find Frodo Beggins in one to ten steps. And for this, what we do is we use an iterative expand. Under the hood, this is basically just an iterative join with some, you can do some optimizations which hopefully help you making it faster. So, what we do is we have the left side which is only the orc and then we have a set of nodes relationships and their nodes. And then we kind of join them iteratively and yield every possible result, of course, if it's longer than one hop and shorter than ten hops. So, it's not that we only yield the largest result, but we yield all intermediate results as well. This is what Martin mentioned before. Here we use the Botic iteration. This is kind of a cool thing for streaming processing systems. Spark, for example, doesn't have an iterative operator, Flink does. And so, we use this iteration operator to iterate over the graph and produce our results. And as a last step, we do filtering. So, we had this predicate that orc one and orc two are not allowed to be the same. So, obviously, we have to filter this result and that's quite straightforward. And this maps to the filter operator which is already present in Flink. So, that's quite straightforward. That's basically the inter-operators we have implemented so far, which allows us to run most Cypher queries right now. What we don't have at the moment is a return statement. This is kind of work in progress, but we will eventually have it. And there is some caveats on how to map the projected results to a graph again. That's why we don't have it right now. And what we also don't have is aggregation. But you can do aggregation by just adding, I mean, GraduP itself has operators for aggregation. So, what you could do is you do the pattern matching with the aggregation and do the aggregation as a follow-up. So, but most of the rest of the Cypher standard we support right now. So, now that we have planned our query, this is what we end up with for Flink. So, this is the visualization of what Flink generates. So, we have our query plan and then we translate it to a Flink query plan. And this is for this particular query. This would be the Flink query plan. You don't see much here. That's completely okay. We just wanted to show you that it's quite a lot of steps you have to take in order to produce the result. And this is just a more detailed view. So, you have basically your data source, then you do a set of filter and project so you get all the vertices and edges you want to work with and then you join them over and over. And in the end, you end up with one dataset which contains your results. And then you can, for example, write them into HDFS or wherever you want to store them. So, what we want to do, what we would like to do is that we want to integrate the Cypher technology comparatively toolkit. This is a set of cucumber tests which allow you, if you implement them, they tell you if you have implemented OpenCypher correctly, hopefully. We haven't really done any, like, broader checks now if what we produce is correct. So, this would be perfect. We want to do benchmarking. We have some preliminary evaluations, but we should go into more detail for this. And for this, we want to use the LDBC social network benchmark. This is a benchmark specifically designed for graph databases or graph systems. And we can do a couple of optimizations. So, right now, we use a greedy planner. We could use a dynamic programming planner, hopefully yielding better tree plans and faster execution times. We want to improve our cost model. So, in order to be able to produce a plan, you have to estimate how costly each step in the plan would be. Right now, we have a very basic, very simple model based on probabilities, but this doesn't scale at least, not really if you have this amount of joints. And, yeah, you can do reuse of intermediate steps. Basically, for every org, we have two orgs in our pattern. Right now, for every org, we would filter the list of all nodes in our graph, but actually, that's the same. I mean, since we only filter for the label org, we yield the same result on both sides, so we could reuse it, basically, which we don't do yet. And another problem in the distributed environment is partitioning. So, if you have vertices which are with a high out-degree or in-degree, so in a social network, for example, people with many, many friends, they increase the computing time on the particular node which is in charge for handling this node. They increase the computing time unevenly to every other node in the cluster, for example, and so we have Datasq, which means that some workers take way more time to produce a result than other workers. And it's a hard thing to partition the nodes over the cluster evenly so that you end up with an even computing time. And, yeah, of course, we want to support the rest of the Cypher features and maybe even introduce more Cypher features like regular path queries or support for this extended property graph model. And here's where you can come into the big picture because we'd like to, if you are interested in this and if you think that's cool, then it's open source so you can work on this with us together and, yeah, just contribute if you like to and if you think that's a cool thing. Yes. Thank you. Thank you.
together and also at the University of Lysus in we have the hobbits, for example, and we have not relationship between the hobbits, for example. example, such as Spark. So to be honest, of of a data flow system like Apache Spark or Apache will give you a short overview about the project our crafts. We use Apache Flink as a distributed abstraction of property graphs on top of this data flow system. So this is the data model that we implement. I will talk about this on the next pattern matching, which we will talk about in open source streaming data flow engine, and it is an abstraction on top of Apache Flink, which underneath, but it provides you already a we have is a property graph. So we have vertices model like in Neo4j, for example. So interesting one. So I will give you an overview, patterns. In the end, this is, for example, this in your query graph. So, you don't want to do uniqueness. So, maybe the most interesting cool thing for streaming processing systems. follow-up. So, but most of the rest of the Cypher