sfspark.org: Stefan Ewen, Introducing Apache Flink

sfspark.org: Stefan Ewen, Introducing Apache Flink

Recording: sfspark.org: Stefan Ewen, Introducing Apache Flink

thanks everybody for uh for being here it's uh it's nice um that uh that you invited us and um and have us give a talk here um have to correct Alexi and one thing um he said I'm the CDO of Apache Flink that's not how Apache Works actually it doesn't have any Coos or cdos right they're just like committers in the project and uh people in the in the project management committee okay um yeah so um yeah I'm here actually to um to give you an introduction into into Apache Flink who of you has actually heard it before um heard about Flink before uh before tonight that's a few hands at least that's nice okay cool um the talk is going to be a bit of a of an overview what what the fling project is about and then I I'd like to focus a little bit on the on the aspects on the on the streaming site um yeah let's just get started okay so just to give you um a little bit of background on on on Flink and the recent and the recent history so it is within Apache a fairly young project I would say it's um it's it's a little more than a year ago that it um that it joined uh the Apache incubator it was still called Stratosphere it's a project that came actually out of uh out of original University Research by the time um we uh we we put it uh in in Apache um we were not working for uh for the University anymore it was um it was a few months old open source project on GitHub uh by then and it was um it was a fairly small stack back then but um with a with a strong um inner inner component um a good runtime which is a good foundation and within the um over the last year we've actually um done quite a bit of work so the community has uh has actually um and the community has grown quite a bit um beyond the uh core team and um managed to add add quite a few components to this deack so I'm actually zoom into that um what is what is Flink today so when I say today I'm referring to to release 0.9 which is actually voted upon I think today so if if uh yeah if everybody likes it it's going to be out by end of the week so so the um the most important thing about Flink um let's say the the one thing you you should remember if you go home and don't want to listen to anything else I'm going to say is uh it's actually a it's a stream processor um the core is a is a streaming data flow runtime that's streaming end to end which presents itself as well yes a stream processor and also as a batch processor um doing doing batch on top of on top of streaming um it it can be deployed um in a in a couple of ways on in different environments and it it has a sample to set the libraries on the on top of the two abstractions um we often refer to them as the data set abstraction and the data stream abstraction um data set being being a finite set of data distributed set and a data stream being a continuous infinite stream may also be a a finite stream but um in in general an infinite stream um since we since we've been polishing the the data set abstraction a little longer initially people were doing more bch than streaming so we uh we initially polished that a bit more it has it has a few more libraries than the um than the data stream API but um this one's actually what uh what we're focusing on a lot these days so to give you just a a very very brief overview I'm not going to talk about the libraries today in in any detail just uh just tell you a little bit what's there on top of the data set so for for the batch use cases there's um compatibility raer for Hadoop um reduce functions um a table API which is um for relational style queries it's it's you it's something you can can think of it as as link from Microsoft where you write SQL but not as a as a string but as a a sequence of uh like language embedded function invocations uh jelly jelly is a is a library for for graph processing it's it's built on top of uh flings iteration and incremental iteration feature um ml yeah you can probably guess that this means machine learning it's also a library built on top of the iteration incremental iteration feature and um and then the the gray components are not part of the cor link project but uh that are librar is that that either either we or or other people have built on top of link so there's an integration with Zeppelin it's alterate paty project for Notebook um web based notebooks um cascading a lot of you know cascading I guess um this uh this work in progress to to connect this to have cascading jobs run on Flink a Google cloud data flow um the batch applications from Google flaw data flow run um they can be mapped to the data set API and uh it's work in progress to actually have the streaming the streaming cloud data flow programs map to the data stream API yeah um Samoa is is also fairly interesting project I think it's a project that came out of um of some people from Yahoo originally and it's a library of um of machine learning algorithms that run on data streams so continuous algorithms rather than algorithms on on static data sets so yeah um given that you have these abstraction and you have the the stream processor in the middle what what does it look like if you if you actually execute something with Flink so um the data set in data stream API look like fluent apis in in Java and Scala these will um yeah you'll I think the the syntax will look familiar the what happens um when when you start executing such a program in in Flink it's it's not going to go through it's not going through the program lazily and and really interpreting the data sets as in a way that for example spark interprets rdds but what what it does it goes through it and and builds um it builds a a data flow Dack and in the process of building that it sends it through two components um called a tpe type extraction and the optimizer so it goes through it it does a lot of uh reflection if you use scalar we use also a lot of macros and generate um a lot of implicit values to to St to strongly type um everything that goes through this so it's it's building a basically a schema in the background it's preena serializers and so on the nice thing about this is when you write a fling program you you basically don't have to worry unless you're using very fancy types about um about how you how you serialize them right it's it's using its um its own generated serializers in the background they do fall back to cryo in some cases but even then they they look at what the types are that go through and um and pre-register them automatically so unless you're using someware types that are you low dynamically that are not visible anywhere um this is typically this typically something you don't have to worry about um for for the uh programs against the data set API it also sends them through an Optimizer which is very similar to a database Optimizer only that it doesn't work on on SQL queries but on yeah on the on the relaxed operations that are in in the Java and scolar API um as of the as of the latest version this actually doing some interesting stuff so it's running code analysis to the functions to to determine some of the semantics like how to how to input fields get coent output fields and so on because that's very interesting for the optimizer to discover potential okay once once this phase is done what what you have is a is is a data flow deck um this is a horrible picture I'm sorry for that um you can you can think about this really as a as a as a broad graph of um of vertices that are parameterized to do certain operations and then have um have data flow channels logical channels that get bundled into physical Channels with which they they execute data um are they exchange data sorry and this this gets um transferred to the master the master um checks this data structure and deploys individual operations tells the operators how to do handshakes on the data streams and so on um what what state each operators in and yeah maintains the metadata for recovery so um what is um let me quickly talk about a few use cases that we had in mind when we when we um when we thought about how how to build how to build Flink how to build the the core architecture so the the four big use cases that we were that we were initially thinking about is um yeah low latency stream um streaming data analysis um but also yeah bch batch pipelines um machine learning and and graph analysis and we were we were looking what what would be a common denominator for all these for all these um programs of all these applications to to ex execute them efficiently and natively in a system without without not actually working around in terms of um in terms of trying to yeah to to make um for example yeah I'm going to show you actually with a with a few counter examples what what we mean by by non-native non-native support so um the first one what people have been doing a lot and to to a large extent still doing today is um execute iteration actually just as a serious of of of steps a series of jobs so um if you if you run something in map reduce I think through logic extent that's actually also true in in in in spark um what what you do is you have the loop not really in the not really in in in what the what the master sees but they're in the in the driver right so what you're doing is you're submitting a lot of jobs and Hadoop this was a big pain because the intermediate data was on on on distributed file systems and Spark that works much better because the intermediate data is in memory and it's referred to in memory but it's still actually a sequence of jobs that you throw to the that you throw to the cluster so the the system isn't really aware that there's iteration it can cach data but it usually caches data as collections it doesn't cach the like the the optimal data structures in which they usually do the processing um I explained before that we have this Optimizer step so if you actually tell the system in a way that it understands deeply that this is a loop it can actually do a lot of of automatic optimization figuring out what is loop and variant caching that um it can in in the sense that for example SQL optimizers do push down of certain things it can also try and push them out of the loop so the these are the sort of things that that are interesting if if you actually deeply embed iterations in a system um another thing is is stream processing so if you if you think of stream processing a a continuous data flow is um is something that you that you would like to process as a as a continuous data flow it seems a more natural abstraction you can you can go go the way and say we we chunk this we Disco This we do mini badges this is a this is an approach that gives you a lot of um that gives you a fast path to to quite yeah to quite a quite a fast approach to stream processing but um It ultimately hits hits the limit because what what you're doing is when you when you chunk this into streams each computation actually looks at one at one part of the of the stream and everything that is is looking at more than than one of these discretized mini batches is something that has to be aggregated across things that each of the mini batches computes and you you start working around a lot it you you can make tooling that that makes this appear more seamless but it's it's ultimately it it's ultimately not quite as as natural and and efficient usually as if you you know actually process the stream as a stream so having all those um having those two examples in mind and um thinking about the the common denominator of of those use cases um the the five Corner points that we's gone that okay is there one on yeah okay thank you so the um the the five design Corner points for Flink that we that we came up with um are actually yeah first of all I said this in the beginning really do everything on streams right the core the core engine should be a stream processor completely pipelined um highly parameterizable so the stream exchanges can be can be back pressured or buffered um a model that is that that is pushing data but coordinating uh in in a pool fashion while this is nice you can do everything on top of of streaming you can do bad on top of streaming um actually without a lot of complication if you want to do it really efficiently you ultimately want to add some special code paths for bch these are not really different a different code path in the execution engine in in the sense that there are like two execution engines in Flink it's more um specialized scheduling modes for batch or or specialized uh recovery logic that that can make the additional assumption that you know the stream is not infinite so we can actually replay it entirely from the beginning because yeah it is it is a finite a finite data set that we're looking at um I hinted at this before we we've um added an abstraction that actually embeds it durations in the data flow so the data flow is not a cyclic um not not entirely there's a there's a form a restricted form of feedback you can do the feedback can be um coordinated across the parallel instances in a in a fashion that it it bulks synchronously feeds back data it can also be asynchronous in the in the streaming API the fourth one may sound a little surprising why do you want to deal with mutable state in in paralyzable computations but it is actually when if if you support it you don't need to expose it to the user but at least if the runtime thinks about it in in some terms you can you can gain a lot of efficiency and flexibility and the fifth one is is basically an artifact of Link being implemented in in Java and if you do extremely data heavy processing in the jvm you you have to worry about how how to deal with the AVM memory and I think most projects have um have either done this or are are heavily moving in this direction I think spark is done with this with has started this with tungsten now I think drill is extremely heavily engineered in that direction and yeah Flink also from the beginning has had this as a as a core design Point okay so um out of the out of the four use cases for Flink I'd like to focus a little bit on on on stream processing here so where where do uh where to put Flink in the in the architecture of um of a streaming infrastructure so if this is like a a simplified blueprint of what um a lot of of people are doing in a in in a streaming infrastructure and their in their um yeah and their company you often have these three Co grain Parts the first one is the is the creation of the of the streams by by various libraries coming from um taking server logs and and and um yeah shipping them or subscribing to database transaction logs the the part in the middle um the broker this is where the Kafka project is is very prominent is buffering these streams is is dealing with making them available to to multiple consumers um long-term storage and the third part um the third part is yeah doing actually something with the streams figuring out what is what is happening looking for event patterns um just refining them enriching them feeding them back on the broker making a making a a refined stream available for for Downstream consumption and this is the part where um we link plays plays well so what um what is what is a a special about about Flink as a as a stream processor what is different than um for example storm or spark streaming so I I would say that that Flink has a neat neat combination of of uh of features and guarantees that it that it gives that you that you don't find in this combination in in other stream systems which which alog together make it actually applicable nicely applicable to to a wide variety of of good use cases so the first one is um the the core runtime of link as a streaming runtime is an is an entirely pipelined runtime so we actually have a mode to um to do also batch exchanges in terms of fully buffered streams but it is um if you use it in streaming it's it it's parameterized to to be completely pipelined and to end pip P line shuffles um the the way that these pipelines run is that they um they also naturally flow control themselves so the the um if you have a task here that is very easy and you have Downstream tasks that are very heavy your um the downam tasks naturally U Back pressure the the Upstream tasks and and apply flow control so that the yeah they don't run out of sync this is something that you um yeah that that gives you naturally the ingredients for for low latency processing the the second part that we um that we added to Flink is a is a very flexible handling of of operator state so if we're doing something on on on streaming on streaming data that is that is more than just you know taking the data stream looking at every record individually maybe parsing replacing some values filtering something and then we're we're it into a search index or so or we're inserting it into a database if we're doing anything that that looks at multiple records at a time if we if you want to detect patterns if you want to aggregate over windows then the then the system is going to is going to maintain some some sort of State um the the most gener generic and flexible variant of of maintaining state in computation is actually maintaining userdefined State one thing that is very interest that is that is um important to notice I think let me go go back to one get let me go back one slide if you have a a processor that is not based on on Mini B Badges and is um is running in a in a pipeline fashion the IND individual operators are actually very long LIF they they they're scheduled they're started and unless a failure happens they actually stay around right they're not scheduled and torn down at the interval of of every mini batch so um it is it is fairly easy to to implement implement stateful computation just by by maintaining some some objects in your user function people have been doing this in in in storm a lot now the the interesting thing that we that we added to Flink is that this this user defined state it hooks actually in the systems um in the system snap snapshot Al um snapshop algorithm and the the checkpointing method that we added and and the system gives you actually exactly once guarantees with in in which way data is is reflected in these user defined data structures that that you create there so the the the guarantees extend not just to any system managed States but as soon as you implement an interface in your in your user defined function they also extend to any state you you keep around in in your user function that makes that makes implementing um a lot of a lot of complex logic um very easy and still and still gives you very very good guarantees as that actually the state that that you do that you get when you when you operate on on Windows that you you collect um you collect a certain amount of of Records be that a slice of time of the stream um a count of Records be that a session window where you say let me accumulate everything until I don't receive something for this partic particular group for for a number of minutes is is in in some sense just a special case of this of this state um that the that the system transparently knows about so if you if you solve actually this this very generic form of State very well then then you get this you get this very easily okay um uh fall tolerance for these um for these stateful computations how how how do we actually do that or what what do we do there uh in particular so there's I think there are currently two two forms of guarantees that that systems that different systems gives um the first one is uh what people like to call at least on processing um it is for example what what storm gives you as as um as a guarantee and that means in case in case something fails it guarantees you it replays everything that that you have lost but because it doesn't know exactly how much you have lost it it may replay a little more so you may have duplicates reflected in in in your in your state that you you maintain right um in contrast to that exactly once um is the guarantee that says Okay I I have a good notion of um of what of what has been lost and I can replay exactly what you have what I have lost and in that way I can guarantee that everything is reflected exactly once in my in my state the last one's obviously much more interesting than the than the first one the first one is is is easier to do it it just means you have to have a notion of of being able to replay data and if yeah if you're not sure how much replay replay a little more right um the last one is is is harder to do the last one is what um whether whether I would say three uh course Grant um approaches to do that the first one is um Disco streams or Min bches we should actually add Trident here as well that just says spark streaming but I think Trident is a should actually be mentioned there as well um which means that you you treat streaming as a series of atomic computations a mini batch once it completed um yeah everything has been reflected exactly once of the mini B fails if one operator fails I mean the whole operator is lost you just bring it up again you start with a fresh State and and you apply everything naturally only once so that is um that is a very fast fast way to get this for tolerance but it um it comes with the with the drawbacks of yeah we having to cut the stream into mini batches and and something that is that is actually very interesting about this that is um maybe not obvious on on on first side this this mini batch interval is is naturally also the the length of recovery if you want to actually tune this in order to trade off recovery time and checkpointing overhead you actually start changing this amount cont of the program because it it changes the window sizes right um what what Google cloud data flow uh tries to do to get around this is um is not discretize the streams but um do do more fine grain um operations as as basically Atomic transactions this works very well if you have Google's infrastructure and have spanner and big table and so on and you can just you know pump this um if you if you don't have this this is this is a bit trickier I would say to realize um especially if yeah yeah especially if you don't have the the Google infrastructure so the the third variant that um that that we came up with for for Flink is um is a variant of the Chandy Lampard algorithm for uh for drawing distributed snapshots of the state and and let me let me tell you um really quickly what what this looks like so um let's assume this is our streaming job right the green bubbles here are are um operators in Flink they um may be stateless they may be State full let's say the um the yellow tubes here are our kafa partitions um let's make the assum Assumption of um of of Link cooperating with Kafka here okay so um four parallel Kafka partitions and and a set of of Link Link operators that consume from here so at any point in time when the the when the system decides to to draw a checkpoint and and this may be actually fairly frequent maybe every second every half second or so um the job manager will send messages to the to the sources to the adapters in Kafka and what they do is they they start injecting a logical barrier into the stream this is this acts like a marker in in the Stream where what part of the input is actually reflected in the checkpoint and what part is is not reflected in the checkpoint at at the point of inserting these um these barriers in in the Stream will actually they will actually write their their Kafka offsets or whatever offsets this is in the Stream back to the back to the job manager and and forward these barriers into the Stream So these barriers start flowing through the stream and they um they they push records before them so they they trigger buffers they trigger window computations and everything and and make sure that when for example when they reach an operator and they trigger checkpointing there they make sure that before the checkpoint is triggered all the records before the barrier are actually inserted into the operator state and at the time of the barrier flowing through the operator a copy of the operator status is drawn um this state is then is then uh sent to wherever you want to actually back it up it like in the in the simplest most case it's it's just sent to the to the master or it can be sent to distributed file system or some other replicated um memory store Um this can happen synchronously or asynchronously depending on on what the um what this back end supports and then the the barriers flow through the The Operators here basically sync on on barrier arrival what um what to reflect in the state and what and what not and and forward it and once once all barriers have reached all things things then you you have actually drawn you've actually drawn a snapshot so um this mechanism has a few very nice benefits um I think the foremost one is actually that the that the snapshotting doesn't interrupt the the data flow so it's it's a bit like doing this doing this These Fine intervals like in in mini badges but they're they're actually flowing they're overlaying so you can have multiple of these things in progress at the same time it's very easy if you have a longer topology that there are five checkpoint barriers or five barriers from different checkpoints flowing after another they're just triggering in individually certain uh versions of the job to be to be committed and then at some point a checkpoint is acknowledged and and uh regarded as committed and there are five others already in the background and process so you can have a you can have a very fast interval that that doesn't it doesn't block any of the any of the Stream and the the other nice thing is because it doesn't um it doesn't bring up tearon operators it doesn't really um have to align with the intervals of the windows um you can you can tune sizes of Windows and um aggregation granularities and checkpointing interval completely independent of each other so trading this these things off uh yeah independent um and and with this comes actually um the abilities to support extremely rich window semantics so the window and Flink is actually something that that is that is extremely nice that um a lot of people have have start to like to like a lot um you can do you can do more than than time Windows you can do count Windows Delta Windows session windows and so on um we we're labeling the window still as as better because we're trying to um we're trying to to get a an orthogonal set of of of Windows where the semantics um complement each other right now it's it's a bit like we tried everything that we can do actually we're trying to like yeah do um yeah bringing it to a more orthogonal set so this may actually change change in the future but it's it's an extremely powerful uh window mechanism and yeah and the such as um it sort of gives you what what I what I called initially a bit of a sweet spot combination of different features you have a very high flexibility when it comes to windowing when it comes to State you don't have um the problem that that checkpointing and windowing and aggregation G aties are sort of interdependent you have uh you get the low latency and and checkpointing doesn't block doesn't block anything so that is um yeah so much uh for for fling checkpointing um this is the probably the only slide where I can where I show a little bit of the apis it's um it's word count again sorry for that but uh at least it's it's two variants of it the the batch word count and the and the streaming word count and the I think the point that this slide here here tries to make is um if you look at the two apis they're they're extremely similar the only real difference is that the this one here called calls the yeah the basic uh the basic data data abstraction and data stream while the other one calls it a data set and on data streams if you want to do anything that is uh that is grouping data you you basically have to Define windows in which in which these groups um are computed and the badg API has you can think of it implicitly one Global window that includes everything so yeah that's the that's the main difference um um just a few words on the on the road map um we're working heavily on streaming as I said we're trying to um to also yeah re rework the window in in their in their um simonics a little bit and the the next the next big goal is to graduate this data stream API from the beta status and um the other thing that we're working heavily on is actually um providing more more backends that you can put the state of uh of your computation in and actually exposing this it's it's more like an internal API right now but also also exposing this to the to the programmer and the way that they um Can for example store the entire contents of Windows or the contents of their userdefined state not inside Flink operations but in outside operations and that checkpointing sort of acts as as group commits to the outside uh outside world and and says okay whenever I'm drawing a drawing a snapshot I'm I'm coordinating with the outside world what is supposed to be regarded as as consistently reflected and what what might be rolled back as part of a recovery and yeah basically do do interaction with yeah with versioning with the outside world there there's actually um work on on getting the the table API which I explained is being a bit like Microsoft link um for the data set also on on top of data streams the first version of that exists but it it doesn't uh support anything that requires window right now so it's it's a bit limited um we're trying to uh to yeah at this window at this window operations to the to this data uh to the table API as well and the interesting thing is because this is this is very close to SQL just that it's not it's not a string in which you represent the query but like a DSL program this is a this is like a good underpinning to actually add add something like a um a SQL par on top of it right the the more longer term goals that we're that we're working on is um flinx Master is currently not highly available which is yeah which is uh something that is highly desirable so we we started working on that um in the yeah in the space of like long running queries um Dynamic scale in and scale out is is pretty important for uh yeah for reacting to to um yeah load Peaks and um the way we we want to go about this is actually making making scale in and scale out um basically a controlled a controlled failure that comes with a with a checkpoint and a recovery with with different parallelism and there's actually work in progress to um to adopt the the state interface as such that this is um that they actually understand the partitioning of the state and can redistribute State upon scale in and scale out and yeah then the the libraries given that we're that we're worrying a lot about data streaming right now it's it's a it's a natural next step for both the machine learning library and the and the graph library to actually add at at streaming variants of the of the algorithms they currently only contain batch variants of the algorithms okay um actually want to see some code or should I go on with slides code is good oh it's good okay then then keep your fingers crossed that this actually uh it's not a what's it called in English the the demo effect or so that whenever you try to demo something it it horribly fails okay let's see if I can do this okay so what river is this one say again what river is this it's a lake in Argentina um okay um let me actually briefly explain with the help of um of this slide here where's the cursor here um what what this demo is um is is showing and um I tried to pick a yeah simple but I hope sort of interesting use case and that is um detect detecting event patterns using using US state machines so um what this scenario is um is is motivated by is is a is a simple yeah thought experiment or use case to do let's say intrusion detection in networks or just validating that certain yeah that certain actions happen in in in certain order so assume we have a set of of machines here that generates Events maybe firewalls or authentication services that generate for certain uh users or IP addresses events when they when they come there and events are supposed to occur in certain orders for example some there shouldn't be an interaction with some application server before some authentication has happened or so right so uh the events should follow a certain pattern and if any at any point in time you you detect that that this pattern is violated then there's there's something goofy right and and this should generate an alert so what what we have here is uh is as a source a set of CFA partitions which which get the events from the different sources um the the yellow bubbles or the Flink operations um a connector to Kafka which um yeah which dezes the events it Roots them based on let's say originating user or IP address it's highly simplified so uh bear with me here and um these are the interesting operations they actually run the run a state machine and um of they they start at a certain State when they when they first um receive an event they they transition to to the next States and they they make sure that the the next event actually does V transition if not then they generate an alert and and the SS here in in our case they just print the alerts to the command line you can think of they sending messages to whatever dashboard you want to have uh these alls displayed in okay so the the topology all in all is actually um I hope sufficiently simple for for a demo it's a it's three operations Source partition State machine and sync okay um I've I've implemented this in can you actually see this probably not no then let me switch the let me switch the uh the screen to to miror okay okay wow this is a tough resolution um okay so what what uh what we have here is the let's say the the skeleton of a the simplest skeleton of a of a Flink job right we have the we have our our enclosing scalar object we have the main method we start by creating a stream execution environment um we use this call here get execution environment that uh that gives you the the context uh defined environment so if I run it from the IDE it will give me an environment that that spawns a a local yeah cluster in the background to allow me to to debug the the job and because I'm running this in a in a VM that's not very powerful I'm setting it to a parallelism of four and um I'm I'm creating a a a data stream here as um adding a source to the environment which is an event gen the event generator source so the the event generator Source let me quickly go into that one doesn't actually look extremely extremely complicated these are the the interesting Parts it um it just instantiates this generator and then while while running collects generator next this is an an interval in in in which the IP addresses that that the events are attached to should be generated so it's um it's it's a fairly fairly simple example all bit in um if we have time left later I can switch to connecting this to Kafka this is also fairly straightforward but uh let's try let's try this one first okay um now that we have that we have this uh the sample stream we can we can do a few things with it so let's say the first thing that we do is actually print it so you have an have a an idea of what um what's going on here so this defines the program in the end we have to um tell the system to actually run this program because yeah it doesn't it doesn't immediately trigger it as soon as you um as you run the as you define the program so let me hope that this resolution actually can show these things and that I didn't mess up my program so it actually compiles okay so yeah that's that's what it uh what it looks like it's just taking this this this stream and and printing it um yeah it prefixes it with a let's say with a parallel instance that is doing this so it's it's events Source PR address and and some form of interaction just lab with the May CD you can think about it you know firewall authentication Service whatever okay um now if we um if we want to do something um something interesting with that the first thing we should should make sure is that that all events that correspond to the the same Source um actually go to the same Target so we um we do uh we just let's say hash partition it and um what do we hash partition it on so the the types that are flowing through the system are these case Clauses here um case Clause event it has a source IP address it's I was lazy I just used an integer it's IP V4 it's pretty outdated but it's just for the sake of example and um and an event type and yeah that's that's basically it so what we want to root it on is um is the The Source address so that's what we're going to tell the system um okay so all right so what happens here is um you can you can just give it the name of the IP address the system will uh it will analyze the the program before you sub submitted and it will actually generate a way of serializing this case class basically as a as a two topple um because it has two Fields it yeah as as a scolar product type and it will it will um it will compile this down to to a reference to a certain field in the tle to a certain offset so this is actually not using at at runtime reflection and so on to look this up okay so that now that we've rooted the events um we want to to run them through the state machine I've implemented the state machine as a as a flat map function so the flat map function is is actually just just here it's very simple um I have my I have some State okay I I chose not to use a window because I want to have this very um I want to have very fine Grand control of of the data structure here so I'm just defining the state as as a hash table and whenever I'm receiving an event here's the collector for alerts where where I want to um I may want to push my alert to the first thing I I do is I I take the the current state that is associated with the event sequence on the source address if I'm not getting anything I assume it's the initial St up I haven't seen this before and then I I tell the um the state machine okay transition with that event if it is an invalid transition collect an alert otherwise if you're transitioned and it's not a terminal State then then remember that does that make sense so far okay so um so we we just um we just flat map by this year um what did I call it State machine okay I think I told my event generator to generate an illegal State transition once every 10 million events so if we run this now we should see some alerts and not the original events in the in the output so again it's it's bringing up a it's bringing up the um can I can I maximize this probably not huh it's bringing up the uh it's bringing up the the system in the background and you see the alerts so it's generating quite a few alerts here even though um it's doing it's doing an illegal transition once every every 10,000 elements so let's actually have a quick look what the what the throughput rate is for this um this is not just um running is that is not just running a passing a few elements through some list what's actually happening is this is bringing up in the background a full a full fling stack you know in including nety Network layer to rout it um serialization stack it's it's putting it into buffers it's flushing these buffers to the channels it has latency watch dos that say if events don't arrive in in the necessary frequency to guarante the latency I'm I'm early flushing the buffers and so on so this is this is all happening here so it's um it it gives you a good a good approximation of what what the throughput is you can expect if you have if you have let's say a a decent Network right if the um yeah if there's any if there's any bottleneck in in flings mechanism to um to to process these records this would show in this in this setting so I think in this events generator Source I had I added some code to actually give us the um give us the throughput that we that we get for this simple job here that's including yeah including visualization hash partitioning and running the state machine so yeah once this is actually remed up what we what we see here is that that the throughput is roughly let me stop it at some value um we're seeing this at all at all four parallel partitions that we see and we see actually each parallel partition is processing something like 1.4 million per per second or so that is um yeah including Source generation partitioning shipping and so on that's per core VM so I think that's that's actually it's actually good number if you put this on on serious machines you can expect this to to go up okay so this is running without any any guarantees right now so what what actually how does it actually impact it if we if we start if we start um yeah if we start um doing the doing some some State backup and so on so one thing that we that we need to tell the system is um how to actually how to actually uh Salvage the state from this function in order to um to be able to restore it so um we just we just add um add the interface here with checkpointed I'm not can't type here really sorry so and what we're checkpointing is a yeah it's basically this hash table this is what we want the system to uh to say for us um in the in the in the simplest most version it's it's just applying Java calization when it when it chips this to to the persistent store and yeah of course now I have to implement a few methods they're they're very simple in my case because I don't want to do um a terrible lot of complicated stuff I really just want to tell the system take this exactly as it is and back up it um so the this method here snapshot State um is is invoked by the system whenever it draws a snapshot it yeah it gives you the idea of the of the checkpoint and and and a time stamp um these are not actually named nicely these parameters I don't seem to have the source code connected but uh yeah for for our case here they're actually not very important they're important if you want to in in some way interact with the outset world if you want to restore State um we're we're getting the state that we we should restore and and all we do is basically when we restore it our initial States should be empty so we just um add them add them back to our initial States so that that should be it that tells the system okay here this one should participate in the in the chy Lamport snapshotting and let's now um let's not tell the system to to draw a a checkpoint every um yeah every every let's start with a th000 milliseconds okay let's see what happens okay so it's uh yeah you're seeing the the output from the checkpoint coordinate yeah you can't read this okay sorry okay yeah here um I don't know how to do this in intellig that I probably have to I have to stop it for a second just to be able to scroll yeah so what uh what you basically see is you see uh triggering checkpoint and completed checkpoint completed checkpoints when the barriers travel through so it doesn't doesn't really take a long time for these checkpoints like 30 30 milliseconds in the background um they're also not backing up a lot of State I think the number of State machines that this generator keeps concurrently running boils down to I think to yeah maybe a megabyte or so per per task so it's it's not a terrible lot here that that has to be snapshotted but I mean you can you can see the idea of of the the flow not being interrupted but being able to continue at the same time and the street uh the stream to keep keep running the TCP connections which have rammed up just keep keep pushing and and everything else happens asynchronously in the background um if you if you look at these numbers again they're actually pretty much like they used to be before right there there hasn't really been much of a drop by by doing a checkpoint every every second so let's see if we actually if we can get this down a little bit and um can we torture the system a bit and say okay it's snapshot every 400 milliseconds um I not tried this before so um let's see what happens I'm also on a development Branch this is super this is extra exciting because any non-released experimental stuff here um yeah yeah so what what you actually see here let me again again stop this you you'll probably start to see overlapping checkpoints are not even yet no so if if we if we have more complex topologies you'll start overlapping checkpoints like triggering checkpoints 16 17 18 19 and then 16 gets acknowledged and later and so on so you if you build more complex topologies you can can start seeing this Behavior okay um do you want to see how to connect this to Kafka or should we should we go on with the talk what what's your preference Kafka okay so one person said Kafka so if nobody else objects then and do that okay so um in order to to do this in in Kafka um what uh what we do is we basically just generate this stream in a in a different fashion so instead of uh generating this um sample stream here we generate a stream using the the Kafka source so I'm I'm running a Kafka mini cluster here on my my machine on against the temp directory that is yeah don't expect a gigantic througho from Kafka on this machine here it's just um it's just not not the right setup okay I'm just saying this because Kafka can actually do much better then then I can I can show this here okay yeah so it's um it's basically the the way you use Kafka you you configure some um you configure some properties um the important thing is that if we if we want to use the the flinks persistent kfka source which um aligns the the barriers in the kfka Stream together with the uh with the snapshots then we have to make sure that kfka doesn't autocommit because we commit at the end of a snapshot so that we basically we're reading and we we're keeping the offsets uncommitted until we reach the end of a snapshot and then we commit them back into into zookeeper so it's a bit like batch commit into a group commit into zookeeper so yeah um I'm giving it the IP address and and the group and the the demo topic and here's um here's just the the distalizer for for for the event data type which which tells you how to uh how to turn the how to turn the btes into this event type and the consumer config is kafka's wrap around this so so let's um oh sorry I think I confus the lines here okay so if we do this on the CFA stream here then we shouldn't see anything initially because we're not putting anything in Kafka or maybe there's just from from the last generator I run there may be some residual data yeah there's there's probably some residual data in here yeah that that caused the the first invalid State transitions we're also still checkpointing at 400 millisecond so let's not let's not do that because otherwise we're not going to see anything on the command line but the at these messages here um and we set this to something like a second or two seconds so it doesn't FL the command line the the overhead um of of checkpointing with Kafka is basically not not not different than than checkpointing locally because all it really does is it it treats the the offsets in Kafka as some user defined state that gets becomes part of the snapshot okay um good let me start this again and then um start another program that that puts some data into Kafka okay so we're seeing this some stuff that it picked up from Kafka that used to be there and didn't didn't align with any prior State and then the the the messages from the from the checkpoint coordinator okay I have another program here that generates data into Kafka okay so um by default it generates the the data with a basically with with a proper state State machine and it's not causing any any illegal State transition so as soon as I I hit this uh this here it will generate an illegal State transition and yeah okay we see already we see the alert over there so I have no idea how to put these windows next to each other in intellig otherwise I think you can see that the by the time you actually hit the button to generate the legal State transition I think the other side already reflects it so the the latency is actually pretty I think pretty sweet here um yep okay uh so so much for for what how much I can do with kfon this mini cluster so connecting to um connecting to Kafka here is um yeah is um it's basically this these four lines of code and this also since if we chose the persistent kka s here this this makes this automatically participate in the in the snapshotting as soon as we as we enable it checkpointing here so so I think it's a it's pretty lightweight integration okay um do I have a few more more minutes for some slides or are we over time already two minutes two minutes okay then um let me just get get a maybe two messages really quick across and that is the following so um I said initially that uh Flink is a stream processor that uh exposes itself as well as a um as as a batch processor and and the way it does it is it treats batch programs as a special kind of kind of streaming program so where a streaming program is a program over an infinite stream a b program is a program over a finite stream um this is actually not not a very far-fetched idea if you think about it right anything in for example in had doop when you consume a file you're not loading the file entirely in memory you're consuming a file stream right and you're reading it bit by bit and the mapper is producing it bit by bit and if you wish um doing doing bad streaming means just thinking this all the way through with a shuffle also is I mean it's not element by elements buffer by buffer but it's it's uh it's along the same lines the reducers can um if they if they are like running running hash reducers they can they can start doing this work already and and and so on in in streaming if you want to do any grouped operations you cannot wait for the entire stream to be consumed because it's infinite so you do windows and in bch you implicitly have a window that is the entire stream right so the window is full and and and triggers as soon as you reach the end of the stream and with streaming programs do pipeline data exchange always uh the batch programs do pipeline data exchange or or blocking data exchange actually most of the time you run them and Flink they do pipeline data exchange blocking is to resolve uh Deadlocks and to break the program into finer uh finer schedulable units so with that um let me actually skip over all these other slides um there there if you need other things in in Flink that you we'll have to go to the Flink meet up tomorrow if you want to hear more about them um like Clos Loop iterations uh feedback in the the data flow that that gives actually pretty good performance for for also for batch machine learning and also enables online machine learning um stateful iterations which make uh use of the fact that you can keep State around you don't recompute a model the entire time but you can can keep updating whatever changed and um and build build recomputation based on changes instead of like recomputing or at least forwarding all data every time and um like I said before on the on the batch site there's there are code paths that do static code analysis optimization of these programs um quite some interesting stuff so the um I think Alex said initially that the Flink Flink sort of popped out of of nowhere all of a sudden H if if you look at the at the Timeline here it's not quite true it has been it has been doing quite a bit of work it has not been as as heavily um as heavily advertised um it it basically started let's say a year ago when we went to aachi before that it was a yeah it was a a project that is um that was a a not too small community but a community was a bit close in itself so but now it's actually it's it's uh it's gaining quite a a few contributors and um yeah if you find this exciting um follow follow us on either either on Twitter subscribe to the news mailing list or um or read the Flink block once in a while it has actually some interesting um technical technical pieces and um if you actually want to learn even more then we're we're um yeah we're organizing the first uh conference on Flink in Berlin in October it's called Flink forward um I think the call for submissions is is actually still open right so um if you yeah if you're if you're trying it out are just uh are just interested in in in talking about other related things or or just about ideas um yeah send us a submission and if you want to learn about the other the other interesting um parts that I couldn't really talk about today then come tomorrow to to the the Flink Bay Area Meetup I I don't remember the exact address but yeah I mean you'll find it it's a Meetup Group yeah okay with that um I'm done thank you so guys we're really over time but we'll just take one question maybe then you can ask folks uh offline and come tomorrow to the mid using High L consumer for the low level consumer for cap integation okay say that again are you using the low level or for the high L consumer for the capg um okay uh we are using Robert you can correct me if I'm wrong I think we originally used the low-level consumer but it was a it was pretty hard because all the broker failures and M re-elections were up to us and and then we switched to the high level consumers but we over overrode the committing behavior of the high level consumer I think this is what we do so it's a it's modified high level consumer I think okay Robert Robert Nots that means I've I've said it correctly okay um do we do we have time for a second question just do it what does Flink mean in German oh uh Flink means Swift or agile or Nimble in in in German yeah actually it means it also in Swedish and we've since learned that it means brave and Dutch so it means a lot of good things in Lang different languages that we had no idea about all right thank you