sfspark.org: Alex Oleinikov, Practical insights on monadic error handling in Spark

sfspark.org: Alex Oleinikov, Practical insights on monadic error handling in Spark

Recording: sfspark.org: Alex Oleinikov, Practical insights on monadic error handling in Spark

so yes I'm an engineer my name is Alex I have been working in Yandex before to do I've been crunching data they're concentrated here and they stock will be really technical so please interrupt me ask questions because if I lose you it will be up until the end of the talk and they're going to talk about the magnetic way of error handling the way we do is in spark and what typical problems we have when we put this into practice in production so couple words about criterias or video we are a global company but we are based in Paris so we have HQ there we do bunch of stuff in at tech the pretty decent scale just to give you an example for instance we have our to be product which runs at 30 billion requests per day or we have offline processing infra which accounts something like 2200 Hadoop nodes so that is a big scale that's one of the duties of the attack per se because it gives you a chance to play with all that intra and all that data you have but at the same time companies try to evolve so that's why we have all the response short project and one of those long shot projects is search search is a project which I'm currently working on and it was greatly explained before me so like the event that was quickly explained before me but search is slightly different because the time constraints we have are even harder so that's why when it comes to search when it comes an auction which Google hauls every time you have to you fire aquarium Google that auction actually has to run in way more stricter time than 200 milliseconds so all that 200 milliseconds thing is actually crunched into way tighter constraints and to do that we would actually imply series of constraints to you meaning that you cannot bid on real-time anymore but you can still influence the way the auction works like you can place it obviously it can change the ads or you can tell Google whether you know the user or the user is new or you can change the way the at the targeting works or depending on the device or on the location and stuff like that so that it gives us a number of leverages which we can use when we do optimizations that all like we do a lot of times it's tough to do that but there is one problem that if we like we will take a lot of data system data from different sources and that data may be inconsistent it made the partial abusing so that we get the most full picture of of the world but if you want to concentrate on those sexy topics like machine learning or whatever you want we have to have a pipeline which is robust enough and having a robust pipeline it was a big challenge in the first like when it just started the project because we were quite a small team in the beginning and everything was in connect everything was not consistent at every point of time so we had this constant investigations why we don't have that part of the data in the output or why the pipeline behaves behaves not as we expected so we came up with the idea but before we start let's formalize the problem so let's take at the very naive example let's say we want to process some URLs don't know images from those URLs and drop them and upload to some other service looks pretty easily done but it's an implementation because if we try to add error handling and want to do truly so we have full control on the pipeline we start to run into program problems like for instance we start with a for loop then we try to download the URL then we drive it drop it only if it works we try to upload the URL and then we have to go in the backward direction we have to handle all years if we fail to load here the data so what do we do we have to do bunch of stuff then if we have another failure we have to do project other stuff and then we have to do bunch of other stuff it will be fail to download the URL and then of course we have to get rid of all these entities which failed and it turns out to be like pretty complex and though a thing which is that is dead so we call it pyramid of Doom and for novice reason and start to read the logic is kind of spread very thinly across the code so there should be a way to deal with that because those 26 lives roughly represent what we see here and six life which is ridiculous so we try to think about the problem in more details and think why exactly we have these problems all the time and essentially when we can move the workflow what we are doing is functional composition because we take functions we plug them into each other and expect that in the end you get the result out of the last of them so that's what we do and those are the signatures of our functions and well you could look closely into objects of those say the double function it takes the URL and it returns image but it's not exactly the case because in fact it likes to us it lies because the actual result of this function is either we can either fail or we can actually get the result and it turns out that we try to compose those very things which are not because the internet each other or those of each other did not match the output so we have to handle those inconsistencies every single time and we thought about it and actually there is like a parasite recipe from a functional programming world which is called binding so what we do is if we have a function which might take the input and it may produce either the result or Leggero which is like the lower track we can actually pretty easily come up with a solution and modify the function so that it actually takes the input as an S which is our data or the failure so that it has exactly the same as those and exactly the same elephant and the logical it is pretty easy you know just everything in focus there we fire the original function every input is not there we do nothing we just wraps it so that's called binding there is the similar concept is called listing so what it means is what you want to take function which does not return the errors you just proxy B Earth through so once you do that you have two functions which have two in twenty two outputs which match so that's what we can do and I wanted to give give credit to the idea image consolation he came up with a nice analogy for this and he actually did it like a good dog who started all this thing it was a London couple years ago and he called this railway oriented programming while railways because like essentially can think about it as a few the two parallel tracks of freeway and every entity has to pass through each of those like through either of those tracks and every time you have a bunch of processing at the entity mages switch the track and once it got to do lower track it can never get back so we only have two tracks one of them can develop you delivers you the data one of them of the receiving areas so let's try to implement the previous workflow but having what what I just said it looks roughly like this so you have these the bind function which you have like two doubles and we applied to the URL that we have punch function to my bind function to modify the crop function and apply to your own that we have vine functionally which multiplies upload function we get a function which takes identical Falco's and we apply to the to the conversion of image not too nice in Scala right so what we can do do the idea Matt equate so that it would be expected for scale programmer we can actually define capital helpers and those are basically flat map and map top functions which we know from Scala so flat map is exactly essentially a tall version of the function of the art we should give us as an argument applied to itself and math is a liquid version of a function which we give to which you give to the act sorry which will give to the entity as a parameter at like to itself so now we have two primitives which are notes a twist programmer and we can actually do pretty nice foreclose because that's how it looks with the ever heavily turned out now once we have all that we might want to think about how to apply that it's far because its participation is different you can be hardly distributed but we might have the same ideas so if we want to do it in spark we have to have a bunch of situations like for instance most of our workflows there they are gathering the information from all those different sources they process the data in intermediate stages and then output some kind of merged entity which we use for interesting stuff like machine learning or here is big for mathematical models or stuff like that so that's pretty much how the workflow boards and as you see between each of those dots which represents the stages of the pipeline we have two pipes so it's logical step to define tentative which will be which will enclose the pipe for data and pipe for errors easy and we can argue with error class and the idea is exactly the same so you have [Music] an entity represented by these two tracks in the beginning the third the air track is empty then you apply the function which is download function note now actually take the our delay of a and returns with our video a and don't you know this is that are the day of the download function does not have to know all about the errors which came before it it just produces the errors which are one for the context of downloading probably should not do its point and that's just for example then we can use implicit because they don't like those and I say guys for instance if you want to make the function you apply not even know only their hand mechanism we have we can afford it by making an implicit conversion so that the other function will have better signature for our clients who can let other teams write the code or we have you can let other people who just load know about the framework pending all these errors regular working subsystems or you can even go and implement a function which does not know about our duties and tested separately tested it nice and easy isolated ways you don't have to you know fire all that pipeline stuff and for every single unit test so that it will its index looks and the way we implement joints which are the essential part of what you do is also pretty interesting because we exploit the idea that once you open up the black lab function it turns out that well actually somatic semantically it means that we just get rid of all the years which were before us and we only care about what will happen now so if we have to order these with errors and we slap map over one of them we just got rid of one over here so the first one compared we do the same thing once again so we get rid of years of the second one as well and we know that those will be handled behind the scenes so we are free to remember whatever join function or whatever function which takes multiple parties into as input without even knowing what to really without that function knowing that there is a framework or is heavily the errors and them Indians love 11 so having that tools we can actually build a pipeline which will work like this and the person can do either stoplight you know here light the years put them in some kind of logs and process them later and of course we can persist the results but nature support is that because we managed to move out all the error-handling from the business logic it means that we can have multiple consumers for those errors and those consumers that may be pluggable or those confused consumers may be may be optional so it gives us a lot of flexibility plus all this code is totally separated from the business logic and we can develop it in whatever way it was it makes different team it may be different people may be different framework whatever you want so that's the guy the end the advances we've got now when it comes to real world it turns out though you know in distributed environments nothing can be done easily and one of the problems for instance we have is what we call a diamond problem imagine the station when we have three yard leases and input we have a bunch of events which once process say the events from our partners and you have a bunch of products to which are related to those events for and we have aggregation workflows for each of those sources which is configured by Caltech and say we have a routine which parses become together instance that something fails within that coffee person routine of course a foreign error and we use that stream of config integrating to our DS and once we want to join the result it turns out involved actually when we build a training set the error stream of the Bill of them in the training set actually contains two sets of errors for parsing counting and well that's not kind of wooden one so with the minimal modifications to our workflow we can actually get rid of this because currently the workflow of workflows the web sorry sacred errors things looks like this and actually what we can do we can just join the error strings either late early so that we can do it in the end the workflow and before that happens whenever we use the same errors twice pages got neglected by the set so that's the first problem where is one second problem is that is double computation which comes as granted with sparks laziness so for instance if you only told the biggest like people only the data stream from sparking what he realized the data it turns out that after you finish that all the caches who missed it or in all the pipeline will be scalar will be stale for instance if you have a data track and the errors tracked all the labs ought to be will be computed and in the enve this doesn't be materialized and if you wait some time after that and only in couple of minutes you pull the second track so much inside of the errors it will turn out that you know all these intermediate dots in workflow will be of actually lost so the the shuffle stage will be reused this kitty kat aruba will be stale and well turns out that we essentially have to recompute the whole workflow twice just to get the errors and it's not nice the easy way to get around this that basically separates all your jobs in advance so in the workflow you have an executor which runs all your materialization or prophesy or whatever you want in a threat though so it serves two reasons first as this second it just utilizes your cluster more homogeneously one more problem with session at the approach is that you cannot afford they have none resource pages we spark highly this encourages that but technically technically you can but if you want to use an era can relate like this you cannot because you can totally imagine the station when and you have in computations on say the entity did not fail and you have a in the end in there in the results dream and then something fails one of the aggregators has died and then you have to recalculate it's tough to get the errors because you cannot have a guarantee that that will not happen and what your ability errors because the stage is flicking it can actually get the negative result in one of the stages and you get in there so in the end you will end up in beam is in a situation where you have and I mean the result and the years so there's inconsistent and the only way to get around this if you really absolutely need to do it is to make snapshots which is of course which is of course killing all your performance so you cannot really afford to have non-trivial if you go this way one more thing is that once you have your pipeline essentially split into two different paths for each of the all the entities it turns out that those pipelines they have different execution times and if you have different fractions of your data dispatched into different parts of the pipelines it will mean that the execution times will be unpredictable there is no real way of get to get around this you can the differences between the pipelines it's all bit of magic but this is the same you have notice that before it starts in doing any kind of things we do so to wrap up the good parts are that we have black bulgur handlers we have the disc logic code which is super clean you don't have to care about any kind of error handling you have a very clear track record for each of the inputs you have in the pipeline so if something is missing or something if something behaves oddly you just go and check what is there is the error corresponding to each of the tiny little things you have in the input and you do like literally definitely so that is that of course we have the calculation overhead we have unpredictable execution time so if you're in strict environments you have to take that into account and of course we have to explain that to people it takes like roughly they every time to explain what the binding means why this is the way we go what is a way to use the framework stuff like that so that's about it questions yes involved [Music] and so the question was about the advantages about both using our lease instead of dataframes so there isn't as the API because indeed they get France or they descent maybe faster but if you want like if you have a big pipeline in which you want to test which you want to which is developed by different people you want that to be consistent and freedom of being checked by compiler is what we are buying into when we go for our keys [Music] very gently la season captain and in ages that party bye-bye so the question was about whether we can do counters how those analytics what are the possibilities basically of this implementation well counters at the lunch table because counters is a whole different topic because in part you cannot have real counters right you the state is recomputed you have counters for both of the computations if you just increase it the counters so to address the counters we actually have a slightly more complex implementation because I always tell totally a part of the story is sort of to our energies in between this industry we have communities in one stream so we have data errors and we have also have meter data so that every step has optional way to record metadata or something like oh I did projections in five in machine learning ID like 10k and two of them have failed and then we can actually pass process the new data to get real counters because they are not influenced by regulations of any kind for the analytics those are usually summaries of errors or into dicta or done right and spark or you can actually get the calendars expert than 2/3 power existence and do whatever you want with those numbers or again you can export the result or the workflow but for that we don't have to have the fancy or anything really we have three so we have analytics DB for theta we have counters in graphite and wheels have summaries and stuff like that right in spark [Music] for what both of you to question the question was what kind of error they like what exactly we define error whether this runtime error with whether it is business error so we do not try to address from time errors because we are like doing best we live in a pretty stable environment for the errors we handle yet those are mostly business kind of errors for instance if your sources are inconsistent and you expect you know matching part of the same entity from different sources and they're missing so we cannot you know continue your workflow for the parts which are pretty good even for the parts which are present we fail them and we produce the Earth's estab or things like I don't know like if the data just does not make sense or you fail to parse the input things like that [Music] really we don't want that crap you guys we have bunch of different jobs the smaller ones are then like today they made 18 votes are counted in tens of gigabytes the bigger ones are about one terabyte [Music] the person holding back of me and take a point what I [Music] the question was how to manage shuttling given the skelter gas given the multiplet we have how to measure shuffling well we try to get the data in a format which is readable in small chunks in HDFS so that's the number one way to get get bigger shuffles in nice and neat way if you cannot get that we just read the data from mr. Pat and as soon as possible we reshuffle right in spark so that you know lays out in joints in stated way I will so one more little thing as Italian TV today the thing is I get about it I did not want to include into the talk one of the reasons why we have this music data thing is that it is actually nice going to have logs delivered for you given the infrastructure we have you don't have to have a separate infer for a lot processing log delivery and all of them are timed which is also a nice feature because you can have your log processing stuff done in typesafe way so that's like a nice feature but not not to say like not the main goal of this framework so that's it and if you guys have any questions thank you [Applause] [Music]