The next board is on graph aggregation, graph grouping, especially on streaming graphs.
Graph grouping is a really interesting challenge because we've all seen individualizations,
hairballs, and complex graphs.
What graph grouping allows you is ready to pick these graphs, group them by certain attributes,
and you have kind of these better nodes that then can be selectively expanded.
But you can also then, on a group graph, which is mostly a monopod type graph, in many cases,
you can also run graph analytics, which is a really interesting problem.
So I'm really excited to have Christopher here because both working on streaming graphs
as well on temporary graphs with graph grouping is a really challenging and interesting aspect.
So I'm really looking forward to the talk, and so without further ado, I come to the graph network.
Yeah, thanks.
Yeah, so thanks for that introduction and also thanks for accepting my abstract for this talk.
Yeah, my name is Christopher Rost.
I'm a PhD student of the University of Leipzig, and I'm currently writing my thesis,
so I'm glad that I bring some free time to doing this talk.
Yeah, so about us or our team, so that's me, and I have also a master's student working on this project.
We called Graph Stream Zoomer, and this project is a result of two master's thesis of our university
from Lia Salman and Rana Nurideen, and I think it's also nice to show the result of a master's thesis here
at the FOSTEM.
At the top is our professor of the database department.
Yeah, just to say that.
Okay, what you should take away from this talk.
So you will see what the property Graph Stream is and why it's important to have the streaming idea
inside the Graph topic, and second why you should or should not group a Graph Stream,
and then you will learn what the Graph Stream Zoomer, so this specific project is,
and the main idea behind it, because we provided Java API to do that.
Okay, let's just start what is an event stream.
I think I don't can skip that maybe.
So we say that anything that happens as a specific time and that can be recorded is an event,
and if an event stream is now stream of these events, so a sequence that is ordered by time.
So and I think everyone knows why we need event processing, so we cannot store everything into a database or whatever to analyze it.
So I want to identify these meaningful events and respond to them as quickly as possible.
Okay, what is now a Graph Stream?
A Graph Stream is an event stream where each event is a Graph element or some Graph update.
Yeah, that could be edges, could be vertices, triples or whatever.
And a Graph Update could be a modification of this.
For example, the addition of an edge, the removal of an edge, the addition of a property or an edge or whatever.
So this is just an overview.
Okay, why should I use now a Graph Stream?
Because I can execute on this Graph Stream all algorithms and also all mathematical stuff from Graph Theory on this stream of Graph Data.
For example, calculate page strength concurrently with the evolving Graph of the Graph Stream.
Okay, I can update my analysis results with a low latency if I combine that in a stream processing engine.
And my goal is to monitor the changes or monitor the changes in the Graph or to create some notification or some reactivity.
For example, if something, some average goes over threshold, then I create a notification.
Okay, by now, the Graph Stream could be very heterogeneous.
That means it consists of many different types and it can also occur on a high frequency.
So it is advisable to summarize the Graph elements in a specific way.
And we can summarize Graph elements by three criteria.
For example, by time, that means Graph elements that belong together.
For example, the time window, we group them together by structure.
That means, for example, edges that share the same source or target vertex can be grouped together.
And by content, that means vertices and edges that share the same label or a specific value of a property.
And we introduced for our algorithm so-called grouping key functions.
That means it is a function that maps a vertex or an edge to a grouping key.
And that could be everything that is inside this vertex or edge.
It could be labeled, temporal information, some kind of property or whatever.
So you can map everything that is represented by a vertex or edge to a key function or to a key.
And on this key, we group the Graph.
So that means, at the end, the result is again a Graph stream, but the grouped representation of that.
Okay, now you can question, okay, why I need this?
So what are the applications of that?
You can think about it as a pre-processing step.
For example, before you calculate the page frames, you just group the vertices to the city attribute of users together or something like this.
You also use it as a pre-processing step.
Second application could be as a post-processing step.
For example, after you apply the Graph stream analysis, for example, a community detection,
you can now group on this cluster ID with our grouping algorithm to summarize the different communities together.
You can also use it to understand the Graph stream in more detail.
For example, okay, just to know which vertex or edge types exist in my Graph stream,
how frequent these different types arrive, or how vertices and edges with different characteristics are connected together.
So just to get deeper insights, or if you use our aggregation functions or aggregation,
for example, counting or calculating an average on that,
you can also get or reveal some hidden information that you would not see in the Graph stream itself.
Okay, so this is an introduction.
Now I explain the ideas behind this Graph stream zoomer application just by an example,
and then afterwards I summarize this.
For example, we're using bike rental data that can have two different Graph schemas.
I named them A and B.
So the first Graph schema A is that a bike rental is an edge between two station nodes.
You see it on the left side.
So a station has several properties like the name, the number of bikes, latitude, longitude, and so on.
And the trip edge has properties like the user ID, so who rented the bike, which bike was used by the bike ID,
and from and to, so when this trip happens, until when, and the duration, for example, in seconds.
So this is schema A on the left side.
On the right side we have a more heterogeneous schema.
So we have also stations and trips as vertices here.
And we have also bike nodes and user nodes with several properties.
So I just divided into this because I can explain the examples that follow a bit better compared to just using a simple schema like here on the left side.
Okay, so how a Graph stream of these schemas could look like this.
Yeah, of schema A I have just these trip edges between two stations and all information inside.
And from schema B I have here the trip nodes connected with all the other vertex types.
So this is just an exemplary view how a stream of this graph data could look like.
Okay, so now begin with a very simple or a simple example of our grouping algorithm.
So the input of the grouping is the graph stream, I think it's clear, and we need a grouping configuration.
And the grouping configuration consists of five attributes.
The first is the window because we are doing windowing on our graph stream.
So I can define a window size, which is here, for example, 10 minutes.
And then VG key are the vertex grouping keys, that means the key functions that leads to the grouping of vertices.
EG key are the edge grouping keys, that means the key functions that are needed to group the edges together.
And we also have a collection of vertex aggregate functions.
This is VA-GG and EA-GG are the group of edge aggregate functions.
So the four on the bottom are just an array of several key functions and aggregate functions.
Okay, and now having the input stream and applying that grouping, we get a result.
And just looking like this, because we define for the vertex grouping keys a function that maps every vertex to an integer value.
And that results in, it doesn't matter which vertex exists in our graph stream, we group everything together to one vertex.
And that's the white one with an empty label.
And the same for edges, that means every edge that exists in our graph stream, we group them together to a super edge,
we call them super vertex and super edge that is displayed here in gray.
And because of the count aggregate functions, we add a new property to the super vertex with the count of all vertices that are grouped together.
And a new property to the super edge with also the count value.
And this is now our result for every window that we defined on the graph stream.
For example, here the first window, second window and so on.
That means we are creating a stream of grouped graphs here.
Okay, that is for schema A and that's the most zoomed out view.
So I group everything together that exists in the graph stream.
For schema B and same grouping configuration, it looks the same because it doesn't matter which type label exists, so we group everything together.
So we have just different counts because we have a bit more vertices and edges and also for the second window it looks the same.
Okay, so this is my first example, so the most zoomed out way.
The second example are called a graph stream schema.
So that means now we are using as a vertex grouping key a function that maps a vertex to its label and a function that maps the edge to its label.
So what's the result here?
Now our node has a label station and the count because the count aggregate function stays the same and our edge has a label trip.
So it's more or less the same because our graph streamer has just one node type and one edge type and that's it also for the second window.
Now it gets a bit more interesting when we are using schema B.
The result here with the same grouping configuration as before is now this.
That means every vertex is grouped by their label and every edge is grouped by their label and now I have like a schema representation of my graph stream.
And again with all counts because we are just using count aggregation.
Okay, and that's for the second window and so on and so on, I just leave here the properties.
Okay, the next example we stay with the vertex grouping keys and edge grouping keys, but now I added several aggregate functions to vertices and edges.
For example I say okay for the vertices I want to calculate the average of all available bikes for these stations.
And for the edge aggregate functions I want to have the minimum, maximum and average duration that a trip between two vertices has.
And the result would be this.
So same grouped graph, but now I have three additional properties on the edges.
Minimum duration, it's in seconds here, maximum duration and average duration and the average bikes available on this station also as a new property.
Same for the second window.
And now my last example, it is, I call it, not the last example, there's one more.
So I added now here a second vertex grouping key function and that's an important thing of the graph stream group.
You can also implement your own grouping key function.
For example this one called getDistrict consumes the latitude and longitude property of the vertices and then calculates like a district identifier.
For example here of Priscilla, whatever, so in which district that belongs to.
And then the graph is grouped on this representing district identifier.
And we also say that for the edges we want to group the edges on the user type.
So that means for every edge, so in this data set we have a user type subscriber and something else we will see it shortly.
So that means for every edge I get new now two edges, one for this one user type, one for the other one.
And also here some aggregate functions added.
And the result is something like this.
So here exemplified for three stations.
And here the district ID one, two, and three and the average latitude and longitude for example for visualization,
proposes to place it on the map.
And for the trips between two stations or between two district representatives,
we calculated also the minimum, maximum and average duration and counted them.
And you see here the green edges are for the user type customer and the red edges are for user type subscriber.
So and the last example is then this one here if I say okay as vertex grouping key functions,
I say please extract me the identifier of that.
That means every vertex that exists in the graph stream is placed here and also for the edge identifier.
That means since we have unique identifiers, every vertex and edges are placed here.
And this is more or less like a snapshot of the current state of this graph stream for the specific window.
So therefore I call that zoomed in.
It's the most zoomed in configuration that you could use.
Okay, you could imagine implementing this is not that easy.
So the master students found a way using Apache Flink and its table API.
So everything works distributed since we are using just the API functions of Apache Flink.
But we also figured out several implementation challenges.
So first was to find a good graph representation.
Second one is since we are creating a workflow of this graph stream,
we have to ensure the chronological ordering of every step in this workflow.
As a third point is also you want to ensure the scalability.
Since if we scale out this algorithm, the scalability should be also high.
And also keep the state as minimum as possible and provide a low latency and high throughput.
So these were several challenges the master student solved quite well.
And at the end we created a grouping operator looking like this.
I don't want to get into detail.
This is just an architectural overview of every Flink steps we used.
What is quite interesting is that we created like an operator encapsulation of this.
That means the operator consumes a graph stream at input and has a graph stream as output.
That means you can combine several of these grouping operators.
Or if you define another graph stream algorithm that produces a graph stream as output,
you can just put them before.
So you can like chaining these grouping operators together.
And like I said, this consists of the mapping of the input data, the duplication of vertices,
grouping of vertices and edges, and then mapping it to an output graph stream.
How an API would look like?
It looks a bit messy, but I think it's quite fast clear what's happening here.
So first you have to define the execution environment of Flink.
Then we read the data from some streaming source, for example, a socket source,
or some Kafka stream, or whatever you want, whatever Flink supports in our case.
Then we map it to a graph stream object, which is the internal representation of our stream.
The interesting part here, you define the grouping operator.
So in the middle, that's the grouping config I showed you in the examples.
You can define it here by an API.
You set the window size.
You set the vertex and edge grouping keys.
You set the aggregate functions and so on.
And at the end, you just execute this operator on this graph stream,
and then you can define a thing or just print it to the console, or whatever you want.
So that's the operator call, how you define it in the API.
And current state, the students are about at 90% of the complete implementation of that.
We figured out some bugs at the SQL or at the table API of Flink that were not fixed yet,
so we had to define some workaround that cost us time.
But like I said, we found the workaround.
And the next steps are that we plan an evaluation.
So how is the latency and throughput of this complete system?
And we want to test it on real-world and synthetic graph streams.
And maybe then publish some results, so let's see.
And also, the user-defined key and aggregate functions are still under development.
Okay, then that's it. That's all folks.
Please check out our GitHub repository, or maybe you want to contribute, so we are open for this.
The two links here at the bottom are also the icons are two other projects.
The one is Gradube.
This is a big temporary graph processing engine also based on Apache Flink.
So there where I'm also a main contributor to that project, which was initially created by Marty Nugans,
who's now working at Neo4j.
And also, the temporary graph explorer is a user interface for that system,
where you can play around with the evolution of a graph, but in a historical data set.
Okay, so that's it. And please ask questions. Thanks.
Yeah, please.
On one slide, you said a problem was to decide on the, on slide 20, I think, as well.
Yeah.
The optimal graph representation in the streaming model.
Yeah.
What was the answer?
And so the question was, what, so we had this challenge to find the optimal graph representation,
and what was the answer?
The answer was a triple stream, but a rich triple stream, we called it,
since two property graph vertices are connected with an edge.
That means every vertex consists of the label and possibly a big set of key value pairs as properties,
and the same for the edges.
And this was our optimal, because you can then model everything with this model.
But the counterpart of this was in here that we have to do a vertex de-deplication.
For example, if you have a self-loop, so from one vertex to another one,
we have a duplicate of that vertex, so we have to de-deplicate it afterwards for this model.
So this was one counterpart.
Yeah, but we figured out that using every concept of the property graph model there as a triple is,
that was the best choice for the students.
So another, yeah?
Would you comment a bit more on the scalability, like what graph size should test this on?
Yeah, so the question was some words about the scalability.
The scalability is an open point of future work, so we don't have concrete results of that.
We tested it with some city bike data that we interpreted as a stream, so some historical data.
And we could process, I think it was about 600,000 edges in a few seconds.
But this is just some first results, and we have not tried it on big and high-frequent graph streams on a cluster.
Because we have huge flint clusters at our university, so we can benchmark the scalability of that in a later step.
Yeah, thanks.
Yeah?
These aggregate functions, are they part of, you know, like a Java API, and how do you define them?
Yeah, so the question was how we define the aggregate functions.
So we have a set of predefined aggregate functions, like the count, average, min, max,
and then you have an interface you can implement against, so there's an interface called aggregate function,
and then you have to implement, I think, two or three functions, and then you can define your own and use it then here on, yeah, there.
Where you give the classes of the account and average property, you can give your own class, and then it will be used.
Yeah?
Could you elaborate more on the real-life use cases or real-life applications?
So the question is if we elaborate more real-life use cases and real-life questions.
So applications.
Applications.
So since, yeah, so we are in, so I'm at the university, that means we are missing real-world data a lot,
and we need also some input from companies to provide us with real-world data that we can use.
So use cases could be, we also have to, we only have this bike sharing stuff or Twitter data and whatever,
and I think if you have something like this aggregated function, like here an average property,
you can use, because at the end it's a time series of changing values, for example, of the average property,
and of like defining a threshold and get the notification afterwards.
I think this is maybe one good application afterwards.
So to, for example, if you have network traffic, you see, okay, now the average, I don't know, packet size is increasing.
Now I get notified, for example, like this.
But this could be an application, yeah.
The idea was to use like a video stream for that, but then the question is,
how much graph that is, could that maybe not be done just in a regular stream processing way?
So I think this is just advisable to use that if you have some quite complex relationships between entities,
then you can use this system besides just an ordinary stream processing engine or complex event processing engine.
So I think the unique point of this is to have the graph aspects into the streaming world, yeah.
So any further questions?
Yeah.
So could the events are only additive, right?
So you can only add to the graph, but not delete from the graph by streaming it, right?
Yes, so at the moment everything is interpreted as an insult only,
but since Flink supports everything, like also updates and also deletions,
it is thinkable about some future work that we also can support this.
Because at the end, the result of us is also, since we are using windowing,
it's also an insult only stream at the end,
but if we maybe think about to remove the windowing aspect,
so they have something like a continuous aggregation or whatever,
then we need to support like a continuous addition on the end
to update already existing aggregating results.
Okay, any further questions?
Thank you again. Thanks.