Scale By The Bay 2018: Jacek Kunicki, Practical Reactive Streams with Monix
Recording: Scale By The Bay 2018: Jacek Kunicki, Practical Reactive Streams with Monix
you okay so welcome everyone to this session on the reactive streams with monix my name is Jessica needs key I work at soft Romeo we're a consultancy based in Poland we write code as everyone here does probably but two there are two things that are specific about us we are fully remote and we are fully flood which is a bit unique especially there in Poland if you want to learn how certain organization works internally feel free to ask questions and today we're going to talk about processing reactive streams with monix so like every session here I'm going to start with the reactive streams 101 everyone in this track has their own introduction so I don't want to be worse so in a stream processing processing pipeline you basically have some source of your data you can call it a producer you have a destination for your data which you can call a consumer and you have a number of stages in the middle that do something with your data so they process your data and of course we have data flowing from left to right from the producer through the stages to the consumer now if you think about such a pipeline and if you think about what can go wrong here some problems might arise when the process when the consuming parties are slower than the producing ones because then you have some excess data and you have to deal it with with it somehow so there are of course several ways to do it the first one that you may think of is perhaps dropping the excess data and that's actually something that happens quite often for example in networking hardware so this is not something really uncommon another way to do it is just block so the slower party can block and not accept any data just block when processing the data anyone it's done it can process the next chunk what you can also do is to buffer so for example the slower consumer can allocate a buffer for the data that's coming from the producer but then the problem with bars is that they are limited so they they they have limited capacity and sooner or later if the producer is really fast you end up with an out of memory error so buffering only buffering is also not the way to go now the way that this problem is solved in reactive streams is using something called back pressure and back pressure is basically away for a slower consumer to tell the producer how much data it can accept so if you--if we have the data the blue arrows here flowing from the producer to the consumer you can imagine the back pressure as an OS data flowing in the opposite direction with the consumer basically requesting data to process from the producer and then what is important about back pressure Interactive streams that is that it's a synchronous so generally in reactive processing we want to be as a synchronous and as non-blocking as possible because if there are any resources we are not using at the moment we should pull them put them back into some kind of pool so that they can be reused by by other components and the same applies to back pressure so it's not only important that we have back pressure but it's crucial that the back pressure itself is a synchronous and non-blocking now every implementation of reactive streams because there are plenty of them has its own terminology for for those stages here so this is a general one monix the library we'll be covering today has its own and so the producer in monix is called an observable so this is the source for our data the consumer well is called the consumer so this is pretty straightforward and the stages in the middle are called transformers so that those are the blog's that are actually processing our data so this is for the introduction to reactive streams and now what is monix did anyone of you hear about monning's before okay more people that I haven't expected so monix is basically library it describes itself as a library for a synchronous processing and even driven processing it offers a couple of abstractions the first one is a task so task is basically a present representation of anah synchronous computations it's it's something like a Scala future but a better one because a future in in Scala is eager so when you create a new future it's executed immediately and you don't actually have any control of when it's executed but in in monix tasks on the contrary you have a full control because it's lazy so you define just define a description of a computation and a synchronous one that you want to perform but it's not executed until you exit we run the task another interesting thing that monix offers is something called Co evil and evil is basically something like like a lazy Val in Scala or like by name parameter in a scalar method but it's lifted to a type level so basically you have a type to represent a lazy computation but this time it's an immediate lazy computation and the benefits from from the fact that that it's represented by a type is for example that you can also include some some error handling so you can you can then pattern much on the result of a coup eval so this really behaves like a lazy Val or by name parameter but you have a type that represents it and you have some benefits of this being represented by an actual type and of course we have observables and that's what we will be looking at today observables are the are demonics abstraction for reactive programming so I want to show you a use case if anyone of you attended my talk last year which covered akka streams this is going to be the same example but with a different library so the example we're going to use is some input data in gzip CSV files and basically the the data looks like we have IDs and we have values so we have two value to two lines for each ID and then we want to process those values by by the given IDs so we want to take two values for each ID then average them if it's possible because it may not be possible since some of the values may be invalid as you can see in the example and then after averaging it we're going to store it into some database here is going to be Cassandra okay so let's then jump to the code and see how monix can can help us solve this problem so we start with us with an empty glass with some configuration parameters that we are going to use later and the first thing we're going to define is it's not yet a transformer so not not not a monix building block yet but it's going to be just a method for a utility method for parsing our our zahra lines from the CSV file so we're going to call it parse line is going to take a line that is a string and is going to is going to be a an asynchronous computation so in monix we represent an asynchronous computation by a task so it's a task of reading and reading is actually a part of our domain model that this is a silt ride that provides an ID and we have two to two implementations first one is a valid reading that has an ID and also has a value but we're also representing in our domain an invalid reading which only has an ID that's going to make computing the average easier and handling invalid values easier so we have a method that that returns the task of reading so to create a task you just use tasks apply this is very similar to how you use a future in Scala here's some boilerplate for parsing the line from a CSV file so it is split by the semicolon and that's that's pretty much it but this is the happy path so if everyone went fine we're returning a valid reading with an ID and the value but actually we want to handle a handle errors here as well because as you remember from the slide some of the values can be invalid so let's just use a try here we'll be cutting a throwable and then we want to do two things we want to log the line so that we know in our in the output of our app that something something went wrong and here we're going to return an invalid reading with chosen eye with just an ID so that's why we wanted to represent an invalid reading as well in our domain model so that we can return it here and have a have an observable of readings be the valid or invalid so so far this really doesn't have to do a lot with monix the only thing that is from monix here is is a task now we are going to actually use the observables and to implement the building blocks of our stream processing pipeline which are going to be dead transformers as you remember so the first transformer is going to parse the files is going to take a file and from from every file is going to emit emit readings you obtained using this parse line method so now we are defining the Transformers as values those are just descriptions of what we want to do they are not not doing anything until actually executed so the parse file is going to be a transformer and actually as you can see transformer in monix is nothing more than a function that converts an observable of type a to an observable of type B so this transformer is going to take a file the Java IO file and transform it to a reading so the input will be an observable of files and the output would be an observable of readings so we're starting with an observable of file here and actually what we want to do we want to we want to take the file read the lines from the file and then we're going to end up with an observable of line of of readings because we would parse them to readings so having multiple files we're going to have multiple observables and so what we want to do with the we want to flatten the observable to have a single observable of readings so starting with the observables files we will be mapping it to an observable of readings and then flattening it so when we think of mapping and flattening we probably think of flat map and now in the inside this flat map we're going to define the logic to process a single file so the first thing we need to do is create a reader of because we won't be reading the entire file into memory we want a reader that reads the gzipped file into an as the yeah in history big fashion so there is some boilerplate to create a reader using called all the Java wrappers but eventually we start from a file input stream and then unzip it and and with the buffer Twitter the reason we did it is that we have a factory method in the observable called from lines reader that takes a bufferedreader and now what we want to do here is to do some processing with the observable the first thing we may want to do is drop some lines from the from the from the beginning because for example we can have a line with with with headers with the name of names of the columns so we can drop an arbitrary number of elements we have a config parameter for that called lines to skip and then we want to execute the actual conversion from a strings from a line to a reading to our domain model with the parse line method we defined previously for this we're going to use the Bob task method which basically takes a description of an async computation that is going to be performed so we have a method that actually takes a string and returns the task of reading so that's what we want so we're using it here so once again what we did here is we took an observable of files and then for every file we we converted it into an observable of readings so we ended up with a number of observables of reading and then we wanted to flatten them into a single observable of reading that's why we use the flat map method here so the next step after parsing the file is actually computing the average so this is going to be another transformer sorry so this is a transformer from a reading because our previous transformer returned to reading to a valid reading because after computing the average we want to end up with something that can be stored to a database and in our database table we have we have two columns ID and the value so inside this average computation we are going to handle the invalid values somehow in order to always return a valid reading that can be stored into the database so here starting with an observable of reading what we want to do is tag is group the readings in groups of two and then for every pair of readings which we'll be sharing the same ID we want to compute the average so to group the readings into groups of two we're going to use buffer tumbling method which basically tag the observable and group sit and amid the group's downstream so the group size is going to be two and now we're going to use map icing because we want to do all the processing all the average computation i synchronously so map I think as you can see takes two parameters first is the parallelism level and the second one is the actual computation so the parallelism level is the number of computations that we want to be executed in parallel and we have a config parameter for parallel in memory operations called non-oil parallelism and here we are going to end up with a collection of readings of size two that we want to process and says since this is not a sink and since this is going to be an async computation we are doing it inside the task and not what we want to do is first well to compute the average will be extra we'll be trying to extract the valid readings and then if that if there are any valid readings will be just summing them and dividing by by two or by the size and if there is no valid reading in the pair we just return a dummy value of minus one for the average so let's start with its extracting the valid readings and for this we just do readings collect pattern much on valid readings and that's pretty much it sorry this is not necessary yeah then we're computing the average and to compute the average we basically check if the valid rings are no name or not empty if so we are taking the values of those and summing them and dividing by the size of valid readings otherwise we're just returning minus one as a dummy value for our average and the last thing we need to do here is to return a dominant presentation of the readings so it's going to be a valid reading and now we need an ID and since we're representing invalid readings in our domain model as well we are always going to have an ID so we are just safe to take first of the readings and take its ID and then take the average we computed to use as the value so once again what we did here in this transformer in this state is we took the observable of readings valid or invalid group them in groups of two and then for every for every group of size two we defined a nice in computation with the task that is going to try to compute an average and if there is nothing to come compute the average from is just going to us dummy value of minus one and now actually if you if you think about defining a building blocks for your processing pipeline and you may also think about a way to combine them so that you want to use some smaller building blocks combined into a bigger one that can be reused later and actually we can also do something like that here so let's write another transformer called process single file which is going to be a transformer from from a file to a valid reading and this would do nothing more than just combine the two previous ones that we implemented so it's just going to take them an observable of files then called transform to apply another transformer this is going to use the parse file and that we're going to use the other one the average computing computation also calling transform so this is a pretty simple way to it's actually just combining functions but you have you have the transform operator on the observable and you can take multiple building blocks that describe a transformer for your stream and combine them into a single one so now if you remember what the processing pipeline looks like we have a source of data we have a destination and we have something in the middle to process so what we have implemented so far is this something in the middle so now we need a source and the destination so we'll start from the end from the destination it's going to be called stored readings store readings and now this is not going to be a transformer anymore because this is the end of our pipeline so it's going to be the consumer it's going to take valid readings so the output of our compute average step and the return type would be unit which means that we basically don't care about the result we only care about the fact that it it has been executed and that it has finished successfully and for the consumer we have a useful useful factoring method that would enable us to do the entire processing in parallel so to store all the data into into Cassandra in parallel this is called for each parallel icing and it takes two parameters it takes the parallelism level and the description of the computation that we want to actually actually use so for the parallelism we have another configuration parameter called concurrent writes this is a different value than the parallelism used for in-memory computations and for storing the value into the database we have some reading repository that exposes a safe method and the only thing you need to know about it is that it it has a signature that it is useful for us because takes it takes a valid reading and returns a task of units so this is some icing computation that would consume our our value trading story and return a task of unit as a description of this computation so now having almost all the building blocks like that the processing blocks themselves and then the destination for our data we're going to assemble the entire pipeline so now we're just writing a method that will be will be executing in our application it's going to be called import from files and now we're going back from the monix word to the scala word so we'll be returning a future of unit here which means that we're doing something as synchronously and we don't care about the result but only about the completion so now to start our pipeline within the files that we want to process so we have a config parameter with with the directory where the files live we can we can just list the files and that would be our starting point we'll do some logging to see what's happening in the output of our application we're going to record the start time so that we know how fast that is and now we are ready to build the actual pipeline so we're starting with an observable and this means we have a collection of files that we want to start with we're using observable from iterable and just use the files here now we want to transform them using the transformer that combines our two processing stages so the process single file one and we want to attach our consumer using consume width and this is the store reading it should be store readings actually yeah now we can see that there is something in red here and actually it says that we have a task of unit here and we want a future of unit so the pipeline we built so far is it doesn't execute at the moment it doesn't do anything because we need to execute it so that it actually does something and only then we will receive the future that represents the the running computation so to run it and to convert it into a future we have a run I think there is some implicit missing here I that in a moment and actually we have some some useful callbacks as well on the on the pipeline so that we can observe whether it has completed successfully or with an error so when it completes successfully we can you do on finish for that we can for example compute the elapsed time and do some logging and when it's completed with an error oh sorry we also have an error handle callback which we can use to do some error handling like logging for example so now we need some boilerplate to actually run it this doesn't do anything more than just loading the configuration for our config parameters initializing the repository and basically running our import from files method the one here and this is what we just created and then it does some cleanup in the end so here you can also see that we have some some missing implicit so let's now get back to to SBT and try to compile it and you can see that we are missing two things the first one is a scheduler and the second one is an execution context here actually both of those boiled down to the same thing which is some kind of a thread pool that will be used to compute to perform our computations because in Scala you need to be explicit about the the thread pool ordered like that the resources that you are going to use for the async computations and it's also a good idea to think carefully about what threads are you going to use for for example not not to run the i/o operations and in memory operations on the same threads because the i/o operations can be slower and we don't want the slower IO operations to block the faster in memory operations so usually you separate thread pools for this kind of operations and use different ones and Scala actually here the compiler requires you to provide a specific thread pool which is a nice thing and for the sake of this example we're not going to separate the thread pools we're actually going to use a single one the default one because we just want it to compile and to run and actually the good thing for us is that the monic scheduler is also an execution context so we can just take the default monic scheduler and use it both as the as a scheduler in monix and as the execution context for this color future but this is something you really shouldn't be doing in production but like optimizing is not not the main topic of this talk so I'm just going to do something very very unsafe here and use the default default execution context or the default scheduler so we take the global one which is something like the global execution context in Scala but please also be aware of of what which execution context you are using so now it compiles just fine so let's let's see inspect our our Cassandra table so it's empty at the moment so we're going to run our application and see what happens so you can see some errors that they tell us that some of the lines are actually invalid you can see that the import was actually finished so we can check if there is something in our database and yeah that's it so looks like it worked but if we come back to our code for a second we can wonder because here while computing the average we're performing an icing computation so we are we are doing all the average computation as synchronously but here when parsing the lines we're actually going not going in parallel so sorry map I think it's not about well it's it's about a synchronous but it's also about doing things in parallel so here are we are computing the averages in parallel but we are not parsing the lines in parallel and perhaps we could also do it but unfortunately monix doesn't expose a built-in method for that and the difference between these two places is that here we don't actually care about the order because we are taking the we already have the groups of two readings we can we can process them in any and I meet them downstream in and order because this doesn't really matter but here on the contrary we want to process the lines one after another and we want to keep the order while emitting downstream because the lines are sharing the ID so if we mix them we want we won't have the same IDs next to each other and that's actually what the average computation step requires from us so here we we actually want would like to have something like map Ising ordered so I'm a PI sync method that executes things asynchronously and in parallel under the hood but keep tracks the order of things who even with they're processed and amidst them downstream in the same order this is of course going to be a bit slower than the unordered version because it needs to wait for the for some of the slower computations before the the other ones can be emitted downstream but it might still be quicker than just processing them one after another so since there is nothing built-in in monix let's try to implement our map Ising court-ordered ourselves so we're going to call it map async ordered it's going to take two parameters a and B because basically we're converting between two observables so what we are writing here is actually a transformer from A to B so from an observable of type a to an observable of type B and the signature is similar to map icing so the first parameter is the parallelism level which is an INT and the other is the description of the async computation we want to perform which takes a value of type a and returns the task of B so it's an async computation that converts between a and B and the return type of this is going to be a transformer between a and B so to do it start with an observable of a what we want to do first is convert is convert the values that are arriving to tasks so to the representations of the computation that we actually want to perform so we're going to call map F here which tags that takes the value of type a and returns a task of B so now we're we have an observable of tasks will be the next step is actually to group the computations in using the given parallelism level so for grouping you already know that we are using buffer term link but here the the parallelism level is that are mined by the parameter that we have here and now we end up with with with groups of observables of tasks of B and we want to like to execute them to execute the actual tasks and then emit a flattened allotment observable downstream so we want to end up with an observable of B so this this is a bit similar of with what we did when when processing processing files so we're basically using including flat map to combine several observables into one and here what we want to do it the task is actually execute all of them and wait wait until they are completed and for this we have a utility method in monix called task gather and they said this is a variant of of like executing the task and waiting until their code that they complete there actually and there's a number of variants that that you can do it because you can you can use tasks gather tasks sequence and I don't remember the third one but basically it is that is the matter of what what order you want to keep and actually tasks gather keeps the order of the results but it may not keep the order of side-effects so if if there were any side-effects in our task the order of the side-effects would not may not be maintained but the order of results would be maintained and the other one are either stricter or less strict so this the strictest one maintains both the order both of the side effects and the results and the less strict one doesn't doesn't care about the order at all so here we are somewhat somewhere in the middle and now we just want an observable of type B here so we're going to create an observable from the gathered task because gathered the type of gathered is it's actually a task of sequence of B so it's a single task that that represents all the that has all the results inside so we start with unobservable from task gathered and then we actually have we have an observable of sequence of B but we want to add and have an observable of B so what we need to do here is to flatten this observable and to flatten this you already know that we are using flat map and we need a function that takes a sequence of B and converts it into an observable of B and there is a helper method for that called observable from iterable so once again what we did here which we took the observable with the values of type A then converted them to tasks with the map map of F then we use buffer Tamblyn to group the task into bigger groups and then inside flat map we used to be tasks gathered to execute all the tasks and to collect convert a sequence of tasks to a task containing the sequence of results which is the gathered here and then we converted it to an observable of type B and now we just may want to add some syntax sugar that so that we have a method available to to run they come to the Rama pacing or dirt on the observable I'm running out of time so so I'm not going to write it by hand I have a snippet for that and this is basically the the pimple pimple my library metals so you have an implicit class which is basically a way of adding a method to an existing type so what we are doing here is adding adding a map izing ordered method to any observable of type a and inside the inside the method that we added we are actually using the method we implemented right before to do the conversion and here the method this is I'm at a new method on the observable that of type A that returns an observable of type B so that's what we actually want to use here yeah and now we just need to we have this is method like any other on the observable so we provide the parallelism level than on IO parallelism and I'm the call to our method the parse line we can try to run it to see what happens I clear the database first yeah so now there it is still three something second so we it's not clear whether it's faster or not and maybe that data set is too small but this is this is like a starting point to to think whether paralyzing is always the good way to go so you actually should always compare the numbers and see whether because paralyzing requires with additional resources it may require a complex switching and stuff like that so it's it's not always faster so you actually always need to think whether actually paralyzing is the way to go or you should you just go sequentially so this is for decode there of course some more things to monix and are some built-in consumers apart from the one that we used here monix supports multi casting so attaching multiple consumers to a single observable there is an extensive docks and actually I've been using monix to point something here but monix trace writer just around the corner it's like release candidate to I think so they will be releasing it shortly and actually it has some new stuff but not not really when it comes to observables like the reactive support doesn't didn't change much in 3.0 so that's why I didn't switch it and just before the talk if you want to see how it compares to a cast reims there is a blog post series I wrote there will be a link at the end as well so you can see on the same example how you can do it with akka streams and with monix at software we were issuing Scala times if you haven't heard about it this is a Scala newsletter with only like interesting articles about Scala no marketing no just good stuff so please subscribe if you haven't already and that's all I had under the QR code and under the link you have all the all the materials there are links to the source code links to Tom onyx and everything you can need if you have any questions you can ask now you can ask me later I'm here until the end of the conference so thank you for your attention you