Tame Your Data with Akka Streams

scale.bythebay.io: Jacek Kunicki, Tame Your Data with Akka Streams

Recording: scale.bythebay.io: Jacek Kunicki, Tame Your Data with Akka Streams

you hi welcome everyone I hope you enjoy your lunch - my name is Jia tech I work at software mule we're a software consultancy based in Poland there are two things that are unique about us at least in Poland we are we are a fully remote company and we have a flat structure I think it may not be like that that special here but in Poland I think we are the one of the pioneers so today I'm going to speak about reactive about stream processing so let's start with a quick recap what stream processing is and actually there's going to be a lot of code here and we have limited size of one of the screens so if you have problems with reading the code you may just try to look at the screen on your right-hand side okay so stream processing well basically not like typical stream processing architecture you need you need some kind of source of your data and it's usually called a producer you have a consumer that it's which is the place where your data lands after its processed and you have a number of processing stages so actually you have data like flowing from left to right from the producer to the processing stages up to the consumer and the data yet is it flows in in this direction so what's actually special about reactive because well we're in a reactive truck after all so the special thing about reactive is how to handle a situation where you have a fast producer and a slow consumer because when is the other way around I mean the producer is kind of slow and the consumer is fast there are really no problems because well it there will never be a situation where the consumer is flooded with data but if the producer is fast and and the consumer is slow like that the consumer has to somehow deal with the situation and how how it deals with it it really depends on the implementation for example if you if you look at the implementation of networks talks like on the hardware level it may be the case that for example the consumer is dropping the like the excessive data but in the reactive approach we use something that is called back pressure and back pressure is basically data flowing well it's not exactly flowing in the opposite direction but it's it's a way for the consumer to signal demand for data to the producer so in in react in the reactive specification we have we have a mechanism which allows the consumer to ask for for a given certain amount of data as the producer for a certain amount of data and this way it's never going to receive more than it's actually able to process at a given moment so this is like general terminology for stream processing and since during this talk we're going to deal with acha streams which is one of the implementation of reactive the reactive trees manifesto let's see what the names look like in a customer's world so the producer is called a source the consumer is called the sync so this is the place where our data lends the intermediate stages are called flows and all those together form something that we call a graph and actually the graph appears also in some other places of the API but we can we will see it later when when I when I show it in the code so there is actually a lot of off stream processing tools available this light comes from my colleagues talk about introduction to Big Data and you have a number of stream processing technologies here ARCA streams is even not among them so let's let's see what was the what are the characteristics of ARCA streams actually how it is different so first of all our customs is is not distributed so it's it's usually used to on a single JVM not not in a cluster of nodes for example and actually this may not be entirely true because when when designing connect us or well programming Kanaka streams application you have two parts like in an ARCA streams application of the first part is the recipe for processing your data and the second part is the runtime and they are completely separated so the runtime part is called a materializer and this is the part that takes the recipe you have written for your processing your data and turns it into an into actual something that can be run and there is an API for the matter eliezer there is a default one provided by ARCA streams is based on actor systems but actually you are free to implement your own materializer and there was an attempt to implement one based on a particular pump which is one of one of the stream processing libraries and it's actually a distributed one so this is October 6th but this was announced like a year ago and actually I recently talked to the ARCA streams guys and they haven't heard anything more about the a particular pump implementation so the guys from a particular pump tried to create their own implementation of a materializer so a runtime for ARCA streams but no one knows whether it actually succeeded or not so in the default implementation in the default actor materializer it's not distributed because it runs on a single JVM when writing akka streams application you design like reasonable building blocks so the the smoke smoke smoke parts of your stream processing pipeline are small recipes for processing your data and they are of course reusable so you can you can define how to process something and then reuse it somewhere else and it's important that when writing those recipes all of them are lazy act until you actually run them so you need to be explicit about saying that I'm done with my recipes now I want to materialize them so to create the runtime and only then the graph is run so all the definitions that we'll be writing here are going to be lazy until explicitly run okay so let's let's see it in the use case that we are going to deal with today well choosing a good example is always the tricky part this one is actually taken from real life scenario that it's just a bit simplified but we used it in a Mac to add product so our input data are going to be some comma separated values we're going to have a simple lines consisting of an ID and the value forever ID we are going to have two lines and it can be the case that either one or both of the lines for a given ID contain an invalid value so one that cannot be parsed our files are going to be deceived and we're going to be extracting them on the fly then grouping by the ID computing the average value for every given also gracefully handling the in value vials values that may appear them and eventually we'll be storing those those data into Cassandra of course it can be like anything at the end of the pipeline it can be Cassandra and can be any other database but in in this example that was taken from her like from real life weird Cassandra so I'm also using it here okay so let's go to the live coding part and once again if well I'm using the presentation mode but if it's if the font is too small here please I know you either need to come closer or look at the screen at your right okay so we're going to start with like a very basic building block not yet related to akka streams at all since we're going to be parsing clients we need to write some logic for parsing our lines so let's write a method called parse line actually the first argument will be for for logging purposes it's going to the bitter path to the file from which the lines come for line comes from and the second one is going to be the line and it's going to return a future because well also in the in the reactive approach your you need to try to be as non-blocking as possible so of course eventually you'll need to block somewhere but all the all the intermediate implementations should be should be non-blocking so we're going to use futures wherever possible and we're going to return a future of something called reading and reading is is our domain model so ever reading has an ID and we have we have a reading that can that is valid so it also has a double value but we also want to gracefully handle the case where the line has an unparallel you and then we have a another case class for it dinah valid reading okay so we're creating a new future here now is there this is some like not not too complicated code for just parsing the line with the same semicolon separated values and returning a valid reading but this is just a happy path you may remember that we also need to gracefully deal with the invalid values so let's just surround it with to it a try catch we'll be cutting anything here and doing two things first we're going to log it that's why we want the file path to be able to detect in which file the in volatile pine was and we're using the invalid reading so the other part of our domain model to return the information that we got a reading which has an ID but it doesn't have a value value so this is the first part as I said it has nothing to do with our streams yet but we're going to start with our customs right now so the next thing we are going to implement is a way to actually deal with our stream of bytes because of course we won't be reading the entire files into memory we'll be streaming them instead and we'll have about a stream of bytes arriving and then we want to like split those stream of bytes into another stream of bytes but the the output stream of bytes consists of lines so so a piece pieces of the file that consists an actual line so we need to write some kind of a delimiter so a component that is going to like split the stream of bytes by a newline character to emit sequences of bytes that represent a line so this is our line delimiter and its type is going to be a flow so a flow as you remember from the introductory slide is that it is the processing intermediate processing state in Arcis trims it gets three type parameters first off first is the input type so the type of data that gets into the flow second is the output type so the things that get out and are passed to the next element in the pipeline and there is also the third parameter the mod and this stands for materialized value and actually what what is a materialized value so the in and out types are so to say internal to the processing pipeline so those are the types of data that flows like within the pipeline but those values those types are not visible to the outside world but sometimes you may want your your flow or the like or your calm content of the pipeline to emit something to the outside world so it can be so it can be accessed and this is the materialized value so this is something you emit to the outside world most of the time will not be using it so we'll only be focusing on the types of values that are flowing inside the pipeline but there is there will be one place where we are actually going to use the materialized value so we will then see what what is it for so our input type is going to be a byte string our output type is also going to be a byte string and byte string is basically an arc a streams wrapper for a sequence of bytes and for the materialized value we have a special type that is called not used you might wonder why why aren't we using I know any for example the reason is that the akka strings developers relief are doing their best to to give a good API both for scala and for Java because many users of this library are writing Java and in Java we have the void type for representing nothing in Scala we have yeah we have any for example and the reason for using not used here is that we want something like a common abstraction that can be used both in Java and in Scala so no matter whether you are using a customs in Java or Scala you're just going to use the not used value and actually it had it also like carries some meaning so it tells you what like that not only it's an empty value but it you explicitly say that you like don't care about it and you don't use it so for writing our line delimiter we have a built-in component that's called framing and we need to supply three things first of all is that is the delimiter so in our case it's going to be a newline character then we specify the maximum length of the line our lives are pretty short so one habit 28 is a reasonable default here and there is a third flock called a load truncation and we're going to set it to true which basically means that we also accept lines then don't end with a newline character if we set it to false and our and then our encountered a line that doesn't end with a new line character it would fail this may be for example a line at the very end of the file which doesn't have a trailing newline so we want to gracefully handle it and that's why we sail it's allowed truncation to true so we have now written our first first flow so first processing stage of our of our stream processing pipeline which basically takes the stream of bytes and turns it into another stream of bytes but now consisting of actual lines from our files so the next step is to actually parse to parse an entire file this is once again going to be a flow since we'll be mostly dealing with flows here its input type is going to be a file in the plain Java IO file so nothing special here the output type is going to be a reading so so our domain model and the materialized value we are once again going to ignore it so it's going to be not used now when writing our own flow we just start with a flow object which has an apply method that takes a type at the input type so we started with a flow of file and now for every file that comes to our pipeline we are going to convert it into a source so in terms of our customs a source of byte strings so so basically a source of the bytes that are coming from the files and not having multiple sources for multiple files we want to somehow come cut them and emit amid the values downstream already converted to a reading but the important thing here that is is that for every file we're going to have a separate source and we want to combine the sources somehow and in our case we're going to concatenate them for this we're going to use the flat map concat method with basically for every file it acts it we can return a source and the values emitted from the source would then be concatenated and emitted downstream so what we got here is the instance of a file that will we go into process and first of all while our files were gzip so we need to obtain a kind of an input dream for for it of the files so we take a new gzip input stream using file input stream inside and it's going to be our input stream and now we have a built-in helper for converting like the java io streams into into actual sources in terms of akka stream streams so we have the stream converters object it has a method that takes an input stream or actually a function that returns an input stream and now we are going to like connect connect the source to our to the other building blocks we have so first of all we want to use our line delimiter to split the stream of bytes into a stream of lines so we are going to use the via method to connect our source to another flow so we just lose the line delimiter here and now it may be it may happen that our our files contains a our files contain header lines for example and we just want to drop them so we have a configuration parameter for for a number of lines that we want to skip from it file and we are using the drop method to just like skip a number of elements and now what we what we got here is is is a line represented by by a by string and what we actually want to put to use when calling our course line method that we we wrote at the very beginning we want the line represented as a string so we need to call them up on the flow and actually the byte string does have a utf-8 string method that just converts it to a normal string that we can use and now the final step here is to a synchronously execute our parse line method with a sum sum given parallelism level to just parse the lines into our domain model for doing a synchronous computations we have a map async method takes two parameters first is the parallelism level and we have a parameter here for non-oil operations so the those that just happen in memory and now we need to provide a function that that takes a string and returns a future because it's going to be an awesome computation and actually our parse line method has this exact contract we just need to like partially run it with with the part of the file which will be used for logging and then the second parameter is that it takes is actually a string so we'll just be passing our our our lines here and converting them into readings so once again about this flop map Concord stage here we are using here for each file we are creating a source of byte strings so all sorts of sequence of bytes and then we are converting them into inter readings into our domain model and what flood napkin cut does it takes values from from those sources concatenates them flattens and emits them downstream so that's why the output type of this stage is going to be to be a reading so this is actually the where where the readings would be emitted from one by one so now having this part we can we can do the average computation so basically group the readings by two and try to compute the average so let's implement it now and this stage is going to be once again a flow it takes a reading like any reading and the output type is going to be a valid reading because for even if one or both lines have an invalid value we would still like to omit some some dummy value so that it can be stored to Cassandra and we'll be using minus 1 for a dummy dummy average when we are not just just not able to compute it the materialized value is still not used here and once again we're starting with the flow object and its applied method we just say flow reading now we have the grouped method and we're grouping our readings into groups of two and then once again we are going to a synchronously compute the average for each pair of readings but this time you're not going to use map async but my pacing unordered now the difference between those two methods is that map async keeps the order of the arriving elements so we have elements arriving in a given order we have icing computations executed for every element and the order of the elements that are emitted downstream resembles the order in which the elements arrived map async unordered on the other hand and means the values downstream as soon as the computation finishes so it's it may like break break the order in the previous case we wanted to keep the order because we needed the readings with the same ID to be next to each other but here we are we actually don't care about the order the order so we can omit the averages downstream in any order so that's why we're using the unordered variant here which is of course quicker in most of cases but it breaks the breaks the order so you need to like be able to choose between those we're once again using the parallelism level for non i/o operations and what we get here is is a sequence of readings which is also always going to consist of two readings we need to be icing because this is the map async method so we need to return a future here and the first thing we do is to extract well not know what I'll be writing right now is not perhaps the most efficient way to compute an average but is for the sake of simplicity it would be like as as fraught as possible of course it's a well it's not about performance in in this very piece of code so please forgive me and like just just accept the implementation as it's going to be so we're going to take our readings then just extract the valid readings from the collection yep and it's going to be mm oh I I need a pattern matching here sorry in the case here as well yeah much better now so this is valid readings and now we compute the average so as I said it's going to be pretty simple but not necessarily performant so we just check if the valued readings are not empty if not then we we map those to vote their values we take the sum and just divided by the size when we wanted to to optimize it we could like for example handle the case where there is only a single valid reading but as I said it's not our main purpose here and otherwise so if none of the readings are valid we're just using a dummy value of minus 1 and what we return is a valid reading and now we need an ID but since from our dominant model all the older readings be it valid or invalid contain an ID we are safe to just take the first value from the from our pair and use its ID and use the average here so once again we we define the float that accepts reading as its input it groups the readings in groups of two and then a synchronously executes some some logic that is going to it compute average if possible out of those two and this is go it is doing it in an unordered fashion which means it doesn't care about the order it emits the results downstream as soon as they are ready so it doesn't it doesn't keep the order of the elements that arrived now the next step is going to is to combine the the steps that we defined before into a single step for processing a single file so it's going to be called process single file it's going to be a float takes a file and emits a valid reading without a materialized once again and this is just to show how you can combine those those different building blocks into into bigger ones that can be reused later for example so once again we start with a flow of file and then we just use the via Combinator to connect the subsequent stages so the first one is is the parse file and the second one is the compute average one now it's important to note that the compiler guards us from from connecting inputs and outputs with incompatible types for example if I switch the order of lines here you can see that it doesn't compile because the the types like the in and out types of flows don't match so this is a verified already at compile time so you are safe about the types you are you are passing from from one flow to the other so this is this is the entire pipeline for processing the file so so we have the num like a single flow so the middle part of the pipeline now we would need a source but we're going to implement the source later and we did a sync so the the ultimate element of our pipeline so let's define the sink now so the story readings is going to be not a flow this time but a sink and sink has two type parameters first is the input type so that arrives and we also have the materialized value so the input type is going to evaluate reading here and actually this is the place where we are going to use the materialized value because when we run the entire pipeline we want to be able to wait until it finishes so we will use a materialized value called future of done so future is future and done is another type which unifies the Scala and Java API it's it's similar to not used but once again the developers focused on one meaning being carried by the types so it's it's we are not using not used here but we're using done instead to signalize that we are actually interested in in the fact that the processing is completed and actually when defining our sink we'll start with a flow once again this time this is a flow of valid trading we once again used them up I think an ordered from Combinator this time with a different level of parallelism because it's going to be an i/o operation like storing to a database so we have a configuration parameter for this one as well and we have some kind of a reading repository that has a safe method that takes a valid reading and returns a future of something this is actually a results that from the Cassandra driver but we're we're just going to ignore it so it's important that it's asking so it returns the future and now what we want to do we want to connect it to some predefined sink which doesn't care about what what gets inside doesn't process the elements in any way but just wait until the processing is complete and signals that so we are we are going to use the two mod method which is to so come to two is used for connecting a flow to a sink and the mud part stands for the fact that we are actually interested in the materialized value and we want to extract it so we're going to use that predefined sink called ignore and sing ignore is basically a thing that takes anything that comes wait until all the weights onto all the elements arrive and then complete the future that is returning so it's returning a future of done because this is its materialized value and with this future it signals the completion of the pipeline but this is not enough so we need to supply one more parameter here and let's get back to the slide for a second to see what what we need here actually so in the middle you have the two Matt Combinator and now to the left you have the flow that we are using which it has its input and output types and it has its type of the materialized value and then on the right you have you have the sink which would like the input type matches and it has its own materialized value and now while using the two Matt Combinator you need to be explicit about choosing which materialized value do you do you want to pick like whether you want m1 or m2 and actually we're interested of the material in in the materialized value of the sink which is to the right as you can see and that's why we are using the keep right parameter to indicate that we want the value from the right in other cases we could use keep left to take a value we could use keep both to take both values and we could also use keep none for just just ignoring the materialized value so once again we have two materialized values the VM 1 to D to the left DM 2 to the right and we want to choose one of them in our case we're interested in the materialized value from the sink which is M 2 that's why we're going to use keep right here ok so that's that like the the ultimate part of our pipeline so now having all of those building blocks defined we are ready to like build and execute the entire pipeline so we're now writing a client line method that returns the future of done which is the the materialized value of the sink we are using it's going to signal the fact that the processing is complete and well we start with somehow obtaining the list of files to process so we're going to be files now let's do some logging so that we see that anything is happening at all let's log just record the start time also for logging and now we have we have our flow defined we have our sink but we need a source so that's the moment well we'll be defining the source of our data so the source in its simplest very own just takes an iterable so we can pass files here now we use the via Combinator to connect it to the process single file flow and then we are using the run with method which does two things well it first of all it's it excited it you can provide a sink that you are going to use for running and actually run it we could also use two that the two methods to attach the sink and then run but since this is a very common scenario there is a single method for that called run with and it just takes a sink now this returns the future so we can we can use the and then Combinator on the return future to just do some logging afterwards this is not necessary so like the the main part that which is enough to run it are those three lines but we also want to do some logging so let's just add add a bit of logging here okay so well looks like we are almost there so let's see what the compiler has to say so let's run SBT here and try to compile that and actually we have some problems here two of them are about a miss in execution context for the futures so the thread pool data is going to be used for executing the futures and the third one is about the materializer so as you remember from the beginning we have now only defined a recipe for processing our data but we need an actual runtime that is going to be used to run our computations so as I said previously the on default and the only one so far materializer is the one based on an actor system innaka so let's assume we're going to have an implicit actor system available here and actually you can you can see here that the wrong with method expects an additional parameter and implicit one and this is the materializer so that's why the compiler is complaining that we don't didn't provide a runtime for our graph and we also need a needed thread pool the execution context and now I'm going to do something very dirty that you shouldn't normally do but for the demo purposes I hope you find it acceptable so I'll say import system dot dispatcher which is an execution context but this is the is the thread pool that belongs to the actor system so normally we shouldn't be using this thread pool for executing our i/o operations because there is a danger of starving doctor system and starving the actor code the communication between actors so normally what you should do is have a separate execution context for the i/o operations and use this one but since here I just want it to compile for the demo process so the execution context that comes from the actor system will be enough to satisfy the compiler but don't do it in production so now we like we have the we have the execution context but we are still missing the materializer so the runtime so let's just create one so we'll just say implicit valve materializer and we have an actor materializer apply method which actually takes the actor system as an implicit parameter so now we are all good so we have we have defined runtime for our pipeline and the compiler is happy so actually there's one more thing to this code because what we are doing now is that we are processing the files like one by one and what we actually would like to do is have the files processed like in parallel so basically it would need something like a load balancer that would take the files that are coming we would have a worker in our case a worker is the process single file flow and we want to execute like a number of instances of the of the process single file flow to just process the files in parallel and then like combine all the results together and emit them downstream as as as valid readings so let's try to implement a simple load balancer and a big generic one that we could use here so we have an object balancer and we have a method called create that takes two type parameters for the input and output and it has two two parameters first the number of workers so the number of parallel process instances that we want to have it's well Anand and we also need the workers of the actual flow and is going to be a flow with the generic types in and out and actually any materialized value in our implementation we're going to ignore the materialized value and just like combine the values together to omit them downstream now to implement it we're going to use something called graph DSL which is a bit low level tool to implement your own your own processing stages or flaws for example it provides us with a create method and in this method gives us access to an instance of something called a builder which is basically a thing that we can use to manage the structure of the graph and now we are going to do two things first off first of all we need we will be using two building stages here one for balancing the work and and the other one called merge to combine all the all the results together and they meet them downstream so we're using builder add to add a built in stage to our graph so the first one is going to be the the balanced stage which well it takes a type and it takes the number of outputs that is going to have about balance is is one of the so called fan-out stages which means that it has a single input and a number of outputs that we define so it just fans out the values that come in in our case is going to be the number of workers we'll call the state balance then we need another built-in stage called merge which hosts also has a generic type this time is going to be out and it now merge is called is a fan instead which means it has a number of inputs and a single output so it's like a bit a bit similar I mean we can we can connect those two together so this is our merge stage and what we are going to return is is a flow shape actually so it's a chef a flow shape means that it's a flow so it has a single input and a single output so we need to provide those so as the input of our stage we're going to use the input of the balance stage and as the output so we're going to use the output of the of the merge stage and actually now if you if you look at the return type you can see that it's not a flow but it's a graph that has a shape of a flow so this is this is what I mentioned at the very beginning that the graphs appear also in the API somewhere so basically every part of the processing pipeline is a graph with a given shape and in our case the shape is a flow shape which means it has a single input and a single output and the graph has a materialized value as well and this is actually not enough so what we want to do to make it work is to is to connect each of the each of the outputs of the balance stage to the total to a new instance of our worker and then connect the output of the worker to one of the inputs of the merge stage so we're going to we need to like iterate from 1 to number of workers and for each like for each worker that we want to use oh and I need to import some fancy DSL here it's grab DSL increase it's and now I'm able to say that for that like we want to take the balance stage use its output connected to the worker and connect the output of the worker to the merge stage and this is actually creating a new instance of the worker and connecting the inputs and outputs accordingly ok so now we should be able to run it well we just need to use it here so now instead of piping our stores we our process single file will say balancer create here we have a configuration parameter for concurrent files and we'll use the predefined like that the one written stage that we wrote before the process single file here the worker okay so we have fact we have Cassandra running in the docker container somewhere here we have an octa stream trading stable that is now empty so actually sorry we're missing a method here like the run actually is going to be fast so we just create the configuration create the actor system and and run our methods okay that's this one so now we see that there are some invalid lines actually because they are locked and we we see from which file does the in value value come and now we see that the processing is done so let's inspect our table and we can actually see the DAR indeed some values here okay so that's the code I wanted to show so of course there is there's a lot more about our customs you have already seen the graph DSL of course you can use it to create less rival stuff as well using some some more advanced features of it you can build your own flow so you can use an even lower level API then the graph DSL to implement your own flowed then you can deal with with state with the internal state overflow I've actually written a blog post about it there will be links at the very end so I encourage you to have a look and you can also use a library or a set of libraries called alpaca which comes from light band as well and this is this is basically kind of reactive integration so they are alpaca is a set of connectors for reactive streams to different files databases like for input and output so anything can imagine it's under heavy development if you if you'd like to like engage into open source this is I think this is a good time for writing your own alpaca connector for example because there is not many of them at the moment there is also monix monix is another implementation of the reactive manifesto so did the reactive streams i've also written a blog series about about the comparison between our customs and monix the links will be at the very end as well so he won't say help please do and one more small commercial we add software mill are issuing scala x which is a Scala newsletter so if you want to stay if you haven't already subscribed I encourage you to subscribe to get like once weekly newsletter with a lot of interesting stuff so that's all I had under the QR code you have the link that is here as well this is a landing page with all with the source code with all the links that I mentioned previously if you have any questions you can come now you can ask now you can stay here we have like 20 minutes of this talk I think so thank you for your attention [Applause] yeah well actually there is the question was how how could you basically group not by by a number of values but by by a field and with by a similar value of in this field right so for example to to say that you want to group until the ID changes right yeah so well there is no built-in stage for this but actually I've written one myself and I contributed it to it's cotton that there's a repository called a customs con trip and actually I needed this so I've written this one use and that the blog post that I mentioned about writing your own your own stage is actually about writing it it's called accumulate wire while unchanged so basically you provide a field for I think I don't have a on trip independence is here but but you just say accumulate while unchanged you provide the field so ID in this case and then it basically accumulates the elements so maintains internal state accumulates them until the value changes then emits the accumulated ones and starts accumulating once again waiting for for the next value so this is well as you see like not everything is built in but it's not that difficult to write your own implementation so you can you can have it this blog post as an example for this one thank you I'm here until the end of the conference so yep so I've written a small oxygen application and you talked at the beginning about the back pressure and initially that's that's one of the people talked about being you know feature reaction streams and this is an example but curious as to why it's actually in my mind the ability to fan how would you don't get in a full system like spark right and smart there's no you can't you can't there's just all the team when somebody tries to tell you like now you got to make these undetermined number of things there's just this pain so like I'm curious as to why you know back pressure is the one you mention and you know being pushed actually I'm not sure I'm getting the point of her question so yeah okay so sorry about that [Applause]