Scale By The Bay 2020: Leo Benkel, Enhancing Spark's Power with ZIO
Recording: Scale By The Bay 2020: Leo Benkel, Enhancing Spark's Power with ZIO
[Music] today i am going to talk to you about zip archive my name is leo bunkel i work as a data engineer and you can follow me on linkedin github or patreon at leobonkey i am passionate about scala data functional programming anything engineering uh feel free to reach out if you want to talk about it you might have uh heard about me before because i've talked about the park io before for instance at uh scala sf or at skeletal and i am also the author of uh soteria and uh two of color as well as the learning scalar community if you want to take a screenshot maybe for the names let me switch to full screen there you go so soteria is a library that allow you to ensure that your libraries and projects are compliant with library versions as well as security vulnerabilities and code styling tour of scala is like a few exercises you can do in like two minutes a day to learn scala and the discord learning community is kind of like grouping all of that together um today we are going to talk about zipper ko but first i want to introduce to you scala and functional programming very briefly just so we are all on the same page as well as spark future how the i o improve all of that and finally zpac io so scala and functional programming skeleton functional programming the functional programming is kind of like this new way of writing code it's uh built on top of object oriented programming and the main thing that needs to be remembered just for this talk is that uh it's you that allow you to reason in terms of type transformation by using a field in mathematic called category theory your main goal when you write functional programming is to have pure function with no side effect push all the side effects to the edge of the program which allow you to write better unit tests and better uh handling of hkc's errors all those type of things and you also want to make sure that you leverage the power of using uh immutable values which allow you to have better straight controls and and don't have any surprise or weird bugs that can usually happen in object-oriented programming uh one way to think about it for me that very helped me uh write better card is if you were to take a function and replacing by the result of this function your application should behave the same way and for me that's very powerful statement because if you were to replace the function by its value by its result that you that mean you can write powerful reading test that means there is nothing secret happening within this this function that can't be expressed only by its results uh in the during this talk you're gonna talk you're gonna see some map which allow you to turn a type a into type b and that allows you to change the operation once you have the type b you can change it into the type c and then into type d etc and you might end up with code which looks like that with functions that turn a to b b to c c to d and it it glue all of that together it's very easy to read and also you can't compile if you flip around because you can't put b to c if your input is a and the compiler will yell at you and that means you can use the compiler as a friend which will help you leverage those features in scala we have some little syntax sugar which are called fork operation which allow you to chain all of that using a map and flat map under the hood and i'm just showing it to you here not to teach you about category theory but so you can understand the next uh the slides about zip io and the io the other piece of the puzzle is what is spark if you are scaled by the bear you probably know but at least everybody will know what's going on uh so spark is a distributed computing framework uh data set as method that we talked about like map and flat map which allow you to turn a data set of a into data center b into data setup c etc so the driver is like the kind of like the brain of the clusters that coordinate which executor does a what job and if you just use spark as is you is the driver will wait for the for all the executor to return the task and then the driver will send the next batch of operations and spark has a very weird approach on laziness some operations are lazy and some are synchronous for instance if you do maps they won't materialize until you you call something like a con for instance and it's very hard to as a beginner to think about how that works and when that happen spark is mostly used in what we call etl so that would be those pipelines where you load data transform them aggregate them and then save them somewhere else for instance you will take two table do a join and write it as a csv why not like a packet file or write it into s3 or all those functionality so usually your pipeline will look like something like that you will fetch data from database one and data from database two and you will do some transformations and then you will do some join and aggregation and then write maybe to one or several output maybe you need a table to postgres and maybe you need a s3 file to be able to continue your pipeline and you can notice that those operations are not interdependent and up until the join so those operation can be done in parallel but by default uh spark will not be able to be aware of that so one thing you can do let me talk about future first um so there was this kind of like revelation for me at the spokeswoman 2019 when anna talked about parallelizing with apache spark in unprecedent unexpected ways and basically in a presentation um and i will explain how zeus type of operation could be done in parallel one way to do it would be to write this operation into futures and which means that when you do that you can see in the spark ui that those operations are now in parallel and that's possible when uh your task will do some operation and using only part of the of the cluster if you have some executors that are idle then you will be able to push more more tasks to it i tried it myself on a little like project i had at work which used to take 40 minutes and i wrapped everything in future it was tedious but i've done it and it took 20 minutes only after so it's like a 50 increase i have to say that my this particular job had to fetch a lot of data from a lot of places postgres mysql s3 before doing those gigantic aggregation and so instead of waiting for each at time now those this data query all happened in parliament which is amazing so the cluster is never waiting for anything because usually you will query wait for the data to return then query the next one wait for the data to return right now because of writing everything in futures you can send everything and then it all come back it's it's really amazing but features uh can be tricky it does a lot of shortcoming for instance you can't cancel a future if you do an await result with a timeout of i don't know 10 minutes your program will return and throw an exception if it got interrupted but you can catch that and continue thinking that the task has been cancelled but it has not if you do some insert for instance they will continue in the background until the task is complete it will not stop at the timeout so that's a big problem another big problem is that if you were to implement a retry that's really tricky because if you you can't take a future and call retry on it because when the future starts it becomes this future of the result type that have started as soon as it's been instantiated there is no dot start on the future and so then you can't restart it because it either it just starts right away so you would have to write a very complex system that takes the function bust it inside of the future and does all the retro logic there in case of failure but the failure if it's a time out is not really a failure so you might end up doing a retries that actually consume more and more and more resources as the old threads are not killed so it's very tricky uh if you want to know more about futures i wrote an article on my tour of scala.com and uh it you can see for yourself uh the the shortcoming i am talking about so now why would we want to use the i o so the io is this library that wrap syncing asynchronous operation smoothly and abstracts that from the from the user you you will just see a series of operation in a four comprehension and they will all happen one after another and if you want one to be asynchronous you just call a function and this thing goes into another thread and you can like orchestrate everything on a macro level that means you can describe your problem completely in a lazy way before executing it so you will describe this like cha the graph of operation hopefully yours is not as complex as this but you see my point where you will describe a graph of task and so you could describe this graph of task and every so the i o is is composed in several parts you will have the graph of task and the service layers which you can feed to the graph and i see that kind of like as a vascular system where you will describe all your networks of operation and at the very end when you're ready to use it you can fuel it with any like gas you want and that allows you for instance in test instead of giving it a real database you can give it a fake database and your code do not change at all you can for instance call insert you can just call your main and fuel it with any of the services you want services being the real world interactions but you can have a file i o that fake fetching a csv you can have a fake effect aws effect bigquery and all of that will not impact anything in your code because your code do not have side effects all of them are pushed into layers um so that's what it looks like the io has three components the other are being the environment that's required to execute this task the middle one the error will be can this fail it's possible to have typed nothing in the middle meaning it can it cannot fail and the output of the type if it succeeds and that allows you to channel those per operation and remember the map earlier so you can turn on z i o of a into a zio of b etc etc and zio on top of all this kind of like foundation build a lot of cool feature on top of it so for instance uh everything is wrapped in cio you don't have you know part of your code that's synchronous and part of your code that is in future in in the i o everything is in the i o and if you want to turn it into an asynchronous call you just call dot fork it becomes a fiber that runs in parallel when you want to wait for it you just have to call dot join and the fiber will return with whatever it was doing uh in a synchronous manner and everything is done flawlessly you don't have any exceptions or or anything and you can cancel fibers and they will be cancelled they will not continue running in the background uh it also has easy retries which mean you can just call dot retry with a schedule system which can be for instance you want to retry three times with a exponential delay etc you can also have timeouts you can also be like i want this task to return within 10 minutes or you just fail you can also have race you can have like for instance you want to query postgres and mysql and you do both queries and they race each other and the ones that return first succeed and the other one is cancelled doing that with future is impossible since you can't cancel and because you leverage the environment here on the left that means you don't need to pass on you know all your dependencies all your services that you will usually do in a normal application you will always you know give the database into your function the first argument will always be your database and you will always have your spark station or an implicit you know and that makes the card like uh cumbersome and with zio you put all of that in the environment and when you need it it's there same for the logging server for everything it can be tricky to use spark and zio together because they are very different framework and there is a lot of boilerplate that needs to be done to assemble those layers to describe your environment and so we're gonna see with uh the parkaio how all of that is done for you i heard that the internet likes mem so let me grab some water while you while you have laughs the funny thing is that i made this slide back in march and they are still relevant you don't have to change it at all i mean it's sad but um so now uh the park i o so z park io is open source you can find it on uh github and it's part of the zio discord there's a dedicated channel there if you need help feel free to jump in and the logo have been made by my dad merci papa if you want uh any logos feel free to contact me i can put you in connections so what is um the park i o so the park is this like boiler plate to start the i o and spark together uh it's kind of like this uh these bowler plate things that you can just use and everything works i mean that's how i've tried to build it there is a few things that will have to to improve but it's it's a work in progress so there's a lot of things that sparked as well and that scala does well but i want to bring it to the next level i want the code to be easier to read i want easier to implement retries i want easier to implement timeout i basically want your spark project to work every single time i don't want you know s3 crash to to make your application fail or postgres database to timeout and then you have to redo like five hours of pipeline all over again with zipper kio and zio you you can do all of that right away which is amazing and you can parallelize dice to speed up performance and me and a few other people are trying to add more features to the package or for instance uh you know all the import spark implicit we want to get rid of that like being it able to be built for you or maybe have wrappers for all the spark that reads so it's easier and the spark that's right and like oh like we want to put like as much help function as possible there's a few mvp right now but there is a lot of work in progress happening right now i'm gonna send the slides but uh here is some links um so the two first one are examples that are in the repo so if you go to the repo you you'll find them the next one is a presentation i gave at scala safe so you can witness the difference in syntax one of the major changes moving from zio 3 1.0 and past 1.0 when the system of layers have changed which makes things a lot smoother it's a little bit harder to get your head around but the the card in z park io is much simpler now and there's a great article from my friend iupac here that have written this tutorial on how to migrate your classic spark application uh to zio with zip io now you can find it on on medium and all of those are like links you can get uh when i share the slide so how do you use zpac io so the goal is to uh your entire application isn't in a trade like this one for instance so you will have object main and extend application and that's it there's a lot more behind but that's the id and what do you do with that well you can for instance in a unit test call your your entire main as is so you can see um over here as a cli argument you know whatever argument maybe it's a past output or maybe it's debugging and you see at the bottom you can create a new version of the trade application which is amazing that's what i was talking about earlier about the vascular system you can fuel it with anything you want so you can put like some test gas into your system and with a fake database and it will work the same so that's the main app that application extends and you can see a few services here so the first one will be the command line the input configuration as of today it is depending on scallop which is an amazing cll library in my opinion but i understand the needs to decouple that and so that's what we are working on there's a few contributors that working on that as we speak the second one is the environment so that can be by default the environment will be composed of uh configuration spark station logging but you can add like your services i'm sure you're gonna need you know database and maybe a file i o or maybe bigquery services or aws services and so you will put all of that there and the output will probably be unit in most application but it's possible to say to set it to anything else if for instance you're not building like a full-on pipeline but you want to build a library that can be called by someone else so you can return you know a data set or anything you want and so how do you implement these zbarchio app trades so you you put it at the beginning as i said like you just extend it we only have like a 10 minute left so i'm gonna speed up a little bit so we're on time then you build the argument that is built from the argument so remember we saw earlier the dash dash whatever so you can put all of those there that would be you know dash dash pass output or some maybe you want dry run or you know one of those and so you just build the arguments then you want the spot builder we're gonna see the implementation later and then any other layer you might have so this one has nothing so you see it's like a succeed nothing but if you were to have your database and anything that would go there then your application and you see it's a full comprehension and you can do step by step like reading transforming writing and it's really easy to read and it's a very readable maintainable syntax so that will be your environment so for instance if you have like file io you remember that that would be the the top here the runtime environment so in this example and that's part of the example that is in the repo so you can go see it for yourself but for instance in this one we want a file i o and a database and that's a combination of those two and that build a new environment that you can use in your application so the configuration um that will be the command line argument service you see here it depends on scallop uh i love scalab but we are building a system so that you don't have to depend on scalp and you can use any other argument system you want and you see the command at the bottom is this kind of like shortcut that allows you to extract your own type from the trait and then you can access any of the argument anywhere so for instance you can do that in the middle of your card you can get oh i want the input id you know if you do like dash dash user id and you want to do this job for this user you can access that from anywhere you don't need to pass user id you like from function to function all the way and then you then you want to do a refactoring and it's a nightmare in this case you access all the arguments from anywhere it's all built in as i was saying the the zpac io community is currently working on removing scalab as a dependency and we're gonna have a separate library that implements the basic config traits that you can import in addition if you're using scalp and you can leave it off if you don't so what about spark so to build spark you just need to do that your builder it depends on your argument because you might want you know to define different things based on your cli maybe if it's a dry run or if it's like dash dash environment dev for instance maybe you want to have it as a different cluster type maybe you want local versus cluster and so that would be the default but you can as i say override in anything and like for instance in this example let's say you have this argument that you want to create i don't know maybe you want to specify the number of executors that still i don't know why you want to do that but it's possible and there is a if you go to the source code there there's a bunch of functions that you can override there's a lot of custom usability but also a lot of opinionated decision done by default that you can configure later if you really want to dive in but by default everything should work as is at least that's the goal of the library same thing you can fetch spark from anywhere you don't have to have use implicit or you don't have to have like this millions argument in your line the only argument your function should have is the input output and the tip all the tools that i use to build this output should be provided by the environment which makes it very easy to read very smooth and if you ever refactor anything the type change which propagates on its own so there is no gigantic reflectoring to do so in addition to all of that there is also a helper method that makes your life a bit easier for instance you can get a zds which is kind of like a transformation that tech spark can produce data set out of it so in this case you use spark to do some transformation and you end up with a data set at the end so you don't need you know to create spark module somewheres and import it somewhere else in it it's all built in for you this effort is to try to remove all the spark implicits you see here it's still there i lied earlier but we're working on it so you don't need those anymore and uh there is a plan so it's almost done on paper it's done and if you want to help please please jump in if you want to create a data set for instance during a test you can just pass a collection and it will automatically detect that it's a collection and that you want to turn it into a data set and you're done if you want to transform a data set so you may be so you see the first block is what we just saw so you create a data set and then you want to modify this data set using the i o so you would have to do this kind of like horrible thing at the bottom here where you have like a flat map of the data set that build the task that makes them up but you have a shortcut here with just the map you're done same for broadcast if you have a data set and you know you want to get the list of ids and broadcast it you can just do ziggy as the broadcast and you get your broadcast built in wrapped in in zio all done for you and now let's say you're in the middle of moving your application and you already done what i've talked earlier like everything is already wrapped in futures uh there is all these tools the future uh implicit you can call and that will allow you to transform all your futures into zio object so i've i'm using this library at work today and remember the application earlier from 20 minute to 40 minutes well the application has grown a lot so now it takes three hours but with the i o it drops one hour down like it's done to one to two hours now so it's a lot faster than futures there are less failures like you might have failures after you know two and a half hours and it's horrible you have to redo the whole thing but with all the retries you know you can do three retries for each of your read and write and that catch 99 of the errors because you have the timeout that means you don't have to rebuild the entire two and a half hours that means you're spending less money because you don't have to rebuild the entire cluster each time you also have better error logging i haven't talked about that but the zio air logging is incredible if one of the tasks fail it will give you the stack trace as expected but it will because it's a lazy graph it is able to tell you what it was running up to this point as well as what it would have run if it would have succeeded so it's amazing there's like so many side advantages all the one i talked about with the timeout with uh all of that and do you remember the the pale parallelization so i try to push it to the limit you know see how many like maybe tiny little file i can fetch at the same time and do transformation on them and there you go you can parallelize everything to the end it's like crazy you can read like billions of tiny files and also imagine zeus being serialized one at a time while 99 of your cluster just being idle that would be a nightmare we have one minute left so what's next here are all the issues so you can go and jump on any one thing i really want to make is the guitar eight so you can just do like sbt news epoch i when you're done more helper function and use any config library i think this is like the three main things and you'll see on the issue pages a little bit more but that's it so thank you very much this is a recording so if i am in the chat below ask your questions otherwise feel free to connect to discord or add me on linkedin and everything and we we can chat thank you very much have a good day [Music] you