OK, thank you. So I'm happy to present Steve Ruhrmann from Ocean Apache Flink Project with our designs. And he's going to talk about a bunch of Flink streaming.

Yeah, thanks for the introduction. I'm Anto. I'm a Flink committer. And today I'm going to tell you a little bit about stateful stream processing with Apache Flink. So for those of you who haven't heard about Flink, it's an Apache top-level project since December 2014.

By now it has become quite active with about 150 contributors working on the project. In a nutshell, Apache Flink is a parallel streaming data flow runtime with which you can do stream as well as batch processing. But in this talk, I will concentrate

on the streaming part. The runtime is capable of giving you low latency as well as high throughput if you set the configuration right. It has a fault tolerance mechanism, which gives you exactly one semantics, which is nice if you require the correct results.

And it lets you do stateful operations, which we will see that is really useful later on in my talk. But before this could be talk about stream processing and why it has become so popular these days. One of the reasons is that many of the problems

we are facing now for which we use batch systems is actually a stream in nature. So it would make sense to process the events of the data when it arrives not batching it up to process it at a later time. By doing so, you basically have faster results.

You don't have to store all the data. And you have more predictable resource consumption in your cluster because they're constantly processing the incoming events. But since all theory is gray, let's look at an example. And this example is going to compromise for the rest of the talk.

I want to come with you on tweet impressions. So you have an input stream of events where each event signifies the certain tweet has been fed to a feed. So and now what you want to extract from this input stream

is how often was maybe the blue tweet or the green tweet how many impression did it get over a given amount of time. So what you usually do or how you would do it is you have the input stream.

You group according to the tweet. Then you find a kind of time frame for computation because the input stream can be theoretically infinite. And doing an infinite aggregation of computation will take you infinitely long.

So you need some kind of, let's say, I want to know the impressions over the last hour. OK. That's quite easy, I guess, for the example. So how would we implement that with a batch of things? OK. First of all, we have to define the data type

of our input stream. It can be any Java or Scala type. In this case, I've used the case class because there's so much 16. One field is the ID, which identifies the tweet. Then we have the timestamp when the event happened

and the count value, which will be used for summing up the impressions. We have to gather execution environments. In this case, it's a stream execution environment, which will tell Flink that we are executing a streaming job. Once we have that obtained, we

can use it to read from a source. Flink supports various sources like Kafka, WebMQ, file sources, and many others. But here, I've used my magic Twitter source to directly read from Twitter the impression events. This will give us a data stream of tweets.

OK, now as we've obtained the tweets, we can start with the actual computation. The first step was, if you remember it correctly, to split the stream into events for the individual tweets. And that was done, I think, by this key-by method,

which you give the field name of your type, or which it shall group. Next, we define our time frame. And here, I set it to a time window of 10 minutes.

And within this time window, I compute by calling sum the number of impressions I've received. OK, that's the job. This gives us a new data stream where the tweet contains now the counts for the impressions

the last 10 minutes. And in order to execute the program, you have to specify a sync. In our case, it's simply printing it out. Then we have to trigger the execution by calling execute. OK, that was quite easy, I guess.

But already at this simple example, you've seen an important concept of stream processing, which is giving, we have to, for aggregations, we have to define a time frame, or what we call a window. Link comes with a different notion to windows.

The simplest one is the tumbling window, which is basically a contiguous, non-overlapping window. So it splits up your input stream into segments of four elements, in this case. And on these four elements, you can, for example, do the aggregation of summing it up,

which will give you, in this case, 27, here, 22, and 8. A slightly more powerful window is the sliding window, which allows you to have overlapping windows. So by defining a sliding window of length 4,

so it contains, again, four elements, to the slight length of 2, which basically says that the first two elements of each window are overlapping with the last two elements of the seating window. You have a different window semantics, which lets you do different things.

Here, in our example, the window length depends on the size, on the number of elements. But it could also define it depending on the time. And for time, Link supports processing time, which means the time when an element arrives at an operator,

as well as event time, which means the time when an event was created. This is really powerful, because this means that we can handle out of order elements. So if you read, for example, from multiple sources, it's not sure that order elements will arrive in order.

And for that, you need event time support. And if the windows, which are like the tumbling and the sliding window, which are supported natively by Link, isn't enough for use case, you can also use quite a custom window with the evict and trigger model, I think. That's not a big deal.

OK, so far, so good. The problem with windows is that they keep the data around. So as long as the window is not finished and hasn't been computed, you have to store the data somewhere. In case the window's either really long

or the elements are really large in size, it might not be feasible to do so. And when it comes to counting, we don't really need the information of the events, right? So the only thing we need is to know that there was an element, and then we have to increment a counter along there, for example.

So to do it a little bit more memory efficient, we can create a stateful mapper, which simply increments the counter value every time it sees a new element. That's a different way to implement a solution to this problem of counting a question.

How would that look like in the API? Well, the only difference to the previous program is that instead of a sum method, we call the mapWithState method, which takes a function, which adds two parameters.

First parameter is the event data, so our tree. The second is the state, which is a long, which you can use a little function to take the current count value and where we implement it to reflect the new one. That is quite easy.

However, what happens if the machine in which our operator once crashes? Then our data is lost, right? And henceforth, we are counting incorrectly, which is not what I've said before that I think gives you exactly what processing guarantees.

So there has to be some mean to recover from a failure like a machine outage, for example. What we basically have to do or what the system has to do, is store your state somewhere. And what this basically means is that the system

has to be capable of drawing a distributed snapshot from a parallel data flow, including the operator states. Since Flink uses a continuous stream model, which means that all the operators of the data flow are online at the same time, and the events

are continuously streamed from one operator to the next operator once they're ready, it's not so trivial. So you somehow have to synchronize the individual operators and to tell them when to take a snapshot so that the complete snapshot,

so every operator, gives you a consistent. And how this happens with Flink is by introducing our markers in the stream pipelines between the operators.

Here it's important to note that these barriers cannot overtake and cannot be overtaken by the elements, which are streamed before and behind the markers. This means that whenever an operator, let's say, hears our stateful method, sees this marker,

it knows that it has now seen all elements up to this point. And it can now take a snapshot of its state. And if all other operators behave similarly, the overall snapshot will be consistent. It can be recovered. It can be used for recovery.

In terms of processing guarantees and the streaming world, which are at most ones, at least ones, and exactly ones, with this HPCONOS barrier snapshotting algorithm, how we call it, we can guarantee

all of these three guarantees. It is the first guarantee, the at most ones, actually not so useful in real-world problems. This basically says that you might see an event in an operator, but if it crashed,

then you might not see it as well. So no guarantees at all. It's easy to realize. A slightly harder guarantee, the at least ones, basically says that you will see every element once. But in case of a failure, it might happen that you see or that you process some

of the elements multiple times. And the exactly ones means basically that in your final result, every element will be accounted for exactly once. And that is the hardest one, actually, but also the most important one.

However, as a user, just now, usually the at least ones guarantee is, for most use cases, sufficient. And you would always use or choose to be the weakest guarantee to minimize the event. OK, what does this checkpointing now

mean for our stateful mapper? Well, if we take a look at it, it looks like before. The only difference is that by enabling the checkpointing mechanism, the system inserts these barriers into your input stream. What the operator does is whenever it sees such a barrier,

it will take the value it has currently stored in this countervalue and write it to some state back end. In our case, it is HFS here. But this state back end is configurable. And currently, we support five systems,

like doing HFS, memory, state back end, and we are currently working on a works DB state back end. So for your user function or for the implementation of that, it's completely transparent. So you don't have to bother about what's happening behind the scenes. OK, let's take a look at the performance

and what the impact of this checkpointing mechanism has on the overall performance. When we talk about performance in the streaming sector, we always talk about latency and throughput.

These are the two important measures. Due to the fact that Bink has this continuous streaming model, where the operator's always online, and once an element has been processed, it can be sent to the downstream operators. It can achieve really low latency.

But to be precise, in order to also achieve the throughput rate, the elements are not directly sent to the downstream operators, but they are buffered for. And they are sent to the downstream operators once the buffer's full or once the buffer timeout has been captured. And with this buffer timeout, we basically

have an upper value for upper bound for the latency. In order to show you that, we wanted an experiment where we had the stream we ran from Kafka. We grouped on some key and counted

the elements in each group. And we ran this experiment on a 30-node GCE cluster with, I think, four cores and 15 gigabytes of RAM around the specs. And we measured the latency, which is the reading line here, versus the aggregated throughput.

And we did that for varying the size of the buffer timeout, so starting from 0 to 100 milliseconds. As you can see, with an increasing buffer timeout, the latency also increases because the elements stay

long on the buffer. And it's more likely that the elements are sent to the downstream operators once the buffer's full the longer the buffer timeout is. However, the upside of this is that also the throughput increases. So as a user, you can use this buffer timeout to tune your system, either to having lower latency

or higher throughput. Wanted to know, what's the effect of the checkpointing mechanism on the overall throughput? What we did is we fixed for the same job where we have a grouping, which means a network shuffle.

We fixed the buffer timeout to 50 milliseconds and computed once the aggregated throughput without checkpointing, and once with exactly once, so with the checkpointing enabled. And as you can see, without checkpointing,

we could achieve, I think it was like 80 or 7 million events per second. That's what we can process. And with the checkpointing of exactly once processing guarantees, we have 82 million events. So by sacrificing approximately 4% of the throughput,

you achieve exactly once processing guarantees, which is a really great oversight. OK, so much for the performance. Now let's take a look at the future and what's going to happen within the next month.

So now I've talked counting. Counting means having a long day, which is not really big. However, it happens that the state of operators can grow quickly really large. For example, if you have an NLP model in your operators, then it can easily be 700 megabytes.

And writing this model to some disk to store it can take some time. And with the current implementation, the problem is that writing the data out stores the operator and also the processing

of incoming events, which is for stream processing, well, kind of prohibitive. So what we are currently working on is an async call as snapshot so that the processing of events can continue while storing the count state to some persistent storage.

Additionally, we've seen, as I said, some of the state can be really large so that it cannot be kept in the heap. So what we additionally work on is an out-of-course state so that once your state grows so big that it cannot be held

in the memory, but it will gracefully spill to disk. That can also be retrieved again. That's one big thing we're working on. Next, we've noticed that during the stream job, the load is not always constant.

So what you usually do to meet your SLAs is to provision for the maximum input you can expect. However, this means that most of the time, you will waste some resources because you don't receive this high input.

So what we want to do or what we are currently implementing is dynamic scaling in and scaling out, which means that whenever Flink detects that there's a slow operator, for example, let's say here, then it will spawn a new operator so that the load is more evenly distributed

across all operators and that the SLAs can be kept again. And the same works the other way around as well. Then whenever an operator in idle is too long, it will be the UP of the topology

will be decreased so that you don't waste any resources. However, that only works if the architecture is capable of requesting new resources, which means new machines, and releasing these machines from a resource manager.

And currently, we support YARN as such a resource manager. We have this feature that when the job manager sees, OK, I need a new task manager to deploy some more tasks, I will tell my resource manager, and the resource manager will give me that. And also, when the task manager is idle, it can be released.

And this will also be part of the new Mesos integration, which will also happen in the next month. OK, last but not least, there's also some other good stuff going on with the cloud of Keras. Currently, we have contributors working

on Stream SQL, which will basically bring SQL to the streaming world so that you can use your SQL knowledge also with streams. And we are implementing complex event processing. So that's really easy, not touching too much code

to detect complex event patterns in your streams. For example, if you want to detect suspicious behavior in your network, you might define some rules and could be used to detect tools, for example. And with that, the only thing which is left now

is to conclude my talk. I hope you've enjoyed it. If you got excited about it and want to stay tuned, you should check out the link at apache.org. They will also find the main list, so which one can subscribe.

And take a look at the code, which you find on GitHub or follow us on Twitter. And yeah, thank you very much for your attention. When snapshotting on the local disk,

how do you ensure all colors? With local disk, of course, they don't have photons. You would have to save it to HFS, for example, where this replicated. There was a benchmark made by Yahoo.

It was also made by you that showed very less performance of what you have shown here. So how can you explain the difference? Well, actually, the benchmark done by Yahoo measured the latency of, let's say, the redest throughput.

The results were not bad for Flink, actually. They were on par with Storm, which is known for having really good latencies, actually. Moreover, here's the benchmark. The measurement of latency is difficult in distributed settings, because the clocks might not be synchronized

between the machines. So even slight changes will disturb the results of the measurements. And the way the Yahoo guys measured the latency was basically that they had a job. Then there was basically a map operation,

or a group operation. And then in this operator, they look up to a redest store to do a join operation. And this result, they wrote every second

to some database to measure the latency between the time point when it was written and the end point of the window of one second, which basically means that if you assume you have a latency of zero, then the element would arrive in the first window,

but wouldn't be written to the store before one second has passed, because only after one second was written to this database. That way, the latency measurement was kind of flawed, in our opinion.

But the overall results were so that Flink and Sloan were on par with respect to latency. throughput was not measured there. If I have my query, can I change the query?

Do you run that? No. It's at the moment like think of having a static query where the data is piped through, and then file it around. Yeah, I mean, with the API, or if you specify it as a SQL query, you

construct a data flow graph, which complies with some internal structure, which is then used to execute it from the specification distribution. I want to move away from the rational database

to even so single database speeds. Can Flink help with data transactions? Flink allows to, within Flink, the guarantees

that there won't be data lost during this checkpointing mechanism. However, once you cross the border from Flink to a different system, they have to interact. I mean, the system has to respect these checkpointing notifications to make sure that no data is

hidden twice, for example. If you do that, then yes. But you have to implement it for the specific connector you're using there. What are the different ways you can connect with Flink?

Yeah, what do you want to connect with Flink?

I mean, to connect with Flink, as a client, you have used Akka internally. So you have like, you can send an Akka message to it. But usually, what you usually do to ingest data

is you have connectors. And there's a wide variety of actually all Hadoop compatible stuff, which is supported. I mean, all Hadoop performance can be used. But there's no other way. I mean, that's not true. You can use the REST interface to request some metrics.

But there's no other way to interact with the system yet, like other protocols. Not yet, for sure. Why shouldn't you use Spark streaming? Well, you shouldn't use it. But the thing is that they use different,

the concepts are different. Spark uses mini-batches to simulate streaming, which might be enough for some use cases. However, what you basically do is you don't have such flexible windowing semantics

as with a continuous stream model. One reason why you think that the continuous stream model is a period. And because of this mini-batching, you have really high latency. So if your use case, which you want to solve,

twice lower latencies, then it's always not possible to do it with Spark, for example. Take the rest of the questions online, please. What sort of equation on IoT, and also the same with stream, is you have a socket connector, for example, on a stream stream.

Can we send for a socket? Yes, yes. That you can do. And the difference, by the way, with stream and with stream from Spark, in this case, you will receive the message for sensing immediately as they go inside. This will be a big discussion of the stuff.