scala.bythebay.io: Nimbus Goehausen, Logical Signatures for Spark
Recording: scala.bythebay.io: Nimbus Goehausen, Logical Signatures for Spark
you hi everybody I'm Nimbus thank you for coming to a talk on a very rather niche subject at 4pm on a Sunday go everyone here so my name is Nimbus i work at Bloomberg on spark platform infrastructure generally providing helping to provide and make life easier for data scientists and engineers who are trying to get various workloads done in spark and make their lives easier so what I'm going to be talking about our data pipelines generally and this is a very broad overview of a simple data pipeline where you take in some sources of data you run some process over it in order to get some output and you know kind of this is about a you know very sumed out you can imagine zooming into this process and there might be all sorts of other things but one thing to notice about these things is it's a dag or a directed acyclic graph and so within the within the process there might be like a sub dag they're going on so when you're developing this sort of thing if you are doing this in a notebook it's pretty straightforward you can kind of you can cache things in spark you can run some code you can see how it does and if it doesn't do what you want you can change it and rerun it and in the the notebook experience in spark is very good like you can you can cache things appropriately you can reuse them you don't have to start all away from the beginning each time but if you're if you're not there's a certain kind of like complexity of a project once you get beyond the kind of simple prototyping just to check if you have like a one-page kind of notebook thing and it works for you that's great but if you start wanting to like you know check your code into a repository have a whole project have compile time checks then you're in the situation where you have a library and if you if you run some code that depends on that library and you realize you need to make changes to that then you have to worry about this cycle here where when you make a code change in your library you have to redeploy your jar you have to restart your spark application whether it be like your notebook thing where you depend on that jar or some kind of submission to to a cluster and then you have to wait for it to run so if by contrast with like a notebook where you're making changes they're not having to reload jars you do have to start from the beginning when you are trying to re reload a new jar with some changes to your code so in each time you make this you make a change to your library and you want to see what it does you have to wait for the cycle but and this can you know obviously you know these you know these some can sometimes be slower to deploy a jar or to start a submission but generally most of the time comes in the execution of like the actual distributed processing if you if this is you know kind of very small you know just a couple minutes then it's not so bad but it can get large depending on what you're trying to do so when you go and run your code on your data pipeline you know you kind of cross your fingers hope it works and then sometimes something like this will happen you have some strange error or maybe you don't have some strange error but your output is coming not at what's you expected with you know no error but still not the answer you're hoping for but when something like this happens it's you know like in the regular course of programming most typical things I think a lot of people do you know something caught you know like monkey patching or just you know just hitting something until it works or some kind of percussive maintenance and it's not you know that's that's a reasonable thing to do it's just that's part of the debugging process is to fiddle with things rerun it and see if it changes anything there's sort of a great way around some of these things obviously it makes sense to test as much as you can ahead of time on small data sets and like you know see that your your your code is logically correct on a small amount of sample a small amount of input but there's this you know even even if you do your best practices you will get into situations where you're running into something that fails on one out of ten thousand things then those can be hard to do it's hard to catch ahead of time and you know if I think if you ask around data engineers data scientists where do they spend a lot of time kind of dealing with it and it's often dealing with problems that arise when you plug in the full data set and dealing with those mysterious issues so for that reason what people will go and do on their own is to split their through code up in two stages so what was just one simple process that goes from start to finish because if you're dealing with an issue kind of in the second half of that and it takes a while for it to run before you hit your issue people go and take their code split it up into stages so we're before we had just you know one single process now we have stage 1 and stage 2 so stage one it has its own main we read from some source and then we run some you know various processes over it and then we save some output and stage 2 is a separate main where we load the stage one path from the you know the previously saved output from stage one we run some process on it and then we save some output or compute some output so you know what do people do this they do this to save time because if you you're running into some kind of difficult issue in this area and this and say Stage one is you know it's at least it's simple and appears to be correct but it might be time consuming you might it might be something where you take a few datasets you join them together to produce some kind of denormalized output it might be working just fine but you don't want to run it well every time if you're trying to figure out something here so you'd rather just like run at once save your output and then you can fiddle with this as much as you like in order to to get your program working correctly but this kind of thing is we're creating a couple of new problems by trying to rearrange things in order to save ourselves time debugging some of the issues we create our well now before we're before we could just had it had a single entry point we could just trigger one thing and it would do the whole thing now we have two entry points so we would have to if you wanted a single entry point we'd have to have one another thing that maybe we have maybe we have three entry points now we have one that calls stage one we have one that calls stage two and then we could have another that calls stage one and then stage two so and maybe use that for when we want use use any of them depending on when you want it also if you have a if for some reason well you thought that stage one was correct at first but then you know it's at least running without error but then you find something in stage two that looks amiss you realize you need to change something in stage one and you need if you do that what you need to do is you have to run stage one and either override the output or save it to a new output and make sure stage two is now picking up from that new output and you know those certainly aren't like difficult things to do but it's just several simple things that you have to remember and that it's not going to happen automatically on these unless you make it so moving moving your code into kind of a multi stage setup when it really could have been easier to express as a single stage creates create some issues like that and also if you have say you have multiple people working on kind of an overlapping project or you know multiple people contributing to the same project but just kind of you know working on two parts of it rough running different sections at once you know often because some of these things are very large and you don't want everyone to save their own copy of these things people might you know share intermediate results so that you know maybe you have your initial thing writing to you know kind of a large amount and kind of you clean it up and save it in one place and then you have multiple people kind of working off of those if you realize those need to change for whatever reason if then it's thorny to deal with it can be thorny to do with that and then you also have to worry with if some people are ever over writing things and other people are kind of not you know that that can create some issues so what would be nice is if some of this stuff just we didn't have to worry about hand crafting this in a custom way every time if it just kind of did it itself that would be nice and so what would it look like if if these things did them if it did it by himself it might look something like this where we have these dotted lines are kind of paths that may or may not be executed in a given execution so now we're expressing this as just a single entry point just we just you know we run stage one on some source and then we get stage one out and then we've run stage two on stage one out in order to get some output so now we have a single entry point it does the whole thing but we also added a little dot checkpoint on stage one because we want we want to avoid redundant computation if we have already run stage one we would rather just use it and so ideally the first time we were on this we we just had start with a source we would run stage one we would save its output and we would also send its output to stage two which then runs and then does the output so this is like a typical full the whole thing plus another step where we save the output of stage one somewhere however the second time we run this since we already have since we already have stage 1 output saved we just want to use that feed it into stage 2 run that and into the output so stage one doesn't execute if we already have this this this around and then if we're for whatever reason stage one changed we had a new stage that's logically different we really want to disregard the old output because we've just changed this section we don't want to use what we saved before because now it's kind of invalidated so we want to save a new output and also say send the new output along to stage two so what are the things that we need in order to do this so one of the key things is we need since we're saving this somewhere and loading it later we need to save it to some path and you know so we need some way of generating a path automatically and that path we want it to stay the same if it's we are avoiding redundant computation we've already made that that computation and we just want to resume from it and it should be different if for whatever reason we you know there's a change in logic there is we definitely want to recompute compute something freshly and ignore what might be saved already so so if the path changes when that happens then it's good so now in spark so now this also apply how do we how do we define our dd's what do we how can we find some kind of identity for these things and this also applies to data frames and data sets so for a given our DD we have some set of dependencies on other our dps and the number of dependencies depends on what kind of our DD it is it could in fact be 0 in the case of like a text file or some kind of like basically a source rdd where you just specify a path and there's no prior rdd it's just you know it just comes into existence with no dependencies on r dds and there's some kind of logic we apply in order to produce the result and this is also how r dds have resilient in their name because because you can run this again if for some reason you lose part of the result you can just rerun and you still have the dependencies in the logic you can just rerun those so part of how spark works is it relies on this that you should as long as you have the dependencies and the logic it doesn't matter if you already have the result great but if you don't you can recompute it and it will be consistent so as a for like a concrete example say we have just a very you know just the natural numbers and our DD and we produce a new our DD by mapping a function over this one say you know double every number and then the result will be you know 2 4 6 8 10 and so on so if for some reason we forgot about this that's okay we can recompute it so really what determines the identity of this is the dependencies and the logic if we have these two things we kind of know what what this could be and if these things are the same then this will be the same so we can make we can encode this in a hash basically we can if a result is the logic applied to dependencies we can make we can say the signature of that result is the signature of the dependencies or the combined signatures of the dependencies plus a hash of the logic and then kind of combined and hashed together just so that and that the point is if we run this in a new execution and things come up the same we and they have the same signatures we know that we've already been here before we have kind of made this computation before and we can resume that and if we come up with something different we know that this is something fresh so with more concrete example so we have say some r DD 1 we map a function f over it to produce RTT to the signature of our DD two should be equal to the combined hash of the signature of our DD 1 and a hash of the function f so and for something where it doesn't have any dependencies on priority DS like a some kind of source our DD if you just load a text file from some uri that's you know just a string then you can just make a very really simple hash of a string that is basically just the uri + some kind of like constant that determines what kind of source what kind of source type it is like you might have a different one for parque files and fruit if you're loading something as a park a file versus text file or versus an object file you have a question dynamic configuration what do you mean oh so so if you're talking like if you have a path and the underlying data changes or eat so one of the things so I'll be I'll be kind of getting to this but yeah it doesn't know if your is basically considering the identity of something as just the path so if the underlying data changes it's not going to pick those up but in many cases when people make these stages or split their you know split their pipeline up in two stages they are also missing out on this so if you go and save some intermediate results obviously if the source changes out from under it you're not going to pass those along unless you explicitly replace them so this is not designed to to handle all those cases and in many cases you actually don't necessarily want to pass along every little update all the time in cases where this is mainly to in the process of development and debugging in order to produce a correct solution you are typically okay with using a particular snapshot of the data at some time so and for example it often makes sense to compose and specify your pipeline as something that you could run again later with the same arguments in order to produce the same answer and as an example of that what is something something people will do to kind of ensure kind of validity and so that you could like run it again to get the same answer you might set like a high water mark for example like if you if you specify as arguments to your pipeline and part part of the logic is use all data up to this point in order to produce some kind of like result that is valid at that time then you can even run that again later even if there's new data and you should get the same answer and it's you know sometimes people sometimes recommend doing doing in in that in that form because you know you don't have to worry about starting it at different times and getting different answers okay so so now I kind of just offhandedly mentioned oh yeah just hash a function no big deal right but it kind of is a big deal there's not really like a built-in like thing in Scala that just says oh yeah this is some kind of identifier you can use for the function that will change if the definition of the function is different that isn't included so but there are things we can do if you go and poke around in Scala and look at the the target directory when you kind of run things and compile things you might notice that there are all these little anon functions for you know dot class so every time you specify an anonymous function in Scala and also i should say this not valid for 2 12 but or so but there are things i'll get 2 i'll get to scala to 12 but let's just say at least in scala to 10 to 11 if you if you have this anonymous caller functions it will make a little class file for each one of them and those are made of you know Java bytecode and you can go and just read that file and hash those bites and you will get a pretty decent identifier for the function so if you go and take this and then later you change change that function so it's like you know x x 2 to x plus x times 3 that class file is going to change after you compile it and you can also you can figure out which class file goes to which function at runtime and enhance those bites and this is also pretty quick so now that was an easy case if it's just like x times 2 or x times 3 you just look at the class file for that function and you know it changes but if you have something that is you know not as trivial as that you there there a problem so you might have a function that depends on some kind of run time value and that might be different depending on different runs and you might also invoke some static method defined on somewhere else and who knows and you might also invoke other functions and what if any of those change is the is the class file for F going to be different if any of those change and the answer is no as long as long as those keep the same name you those things could be changed or things they depend on could be changed and the class file for F will have no idea it'll stay the same so what you have to do is you have to follow these dependencies recursively until you stop finding things or just you know only find things who already seen before and figure out all the class files that a particular function depends on because if any of them change then that function could easily produce different results it Mac czyli might not some of those things could change in a ways that it actually doesn't but there's not really great ways to know that so the safe thing to do is just assume that if any of them change that the lot it's logically different so we want to figure out all of those class files make a hash of them and also any kind of run time values like say something that like comes in not like you know passing in an argument on the command line you obviously won't those won't change those won't change the bytecode of any class files and also things like that just happened to be an answer you compute on the fly if you're using those and your functions then you know you obviously have to have to include those in the hash at runtime so if we have the ability to generate these these logical signatures associated with specific you know our db's and such we're still not done we need some kind of we basically specified there's different behavior that might happen depending on whether or not this exists so we need some kind of mechanism in order to switch things these things out and that is and that's a tricky question so how would you do this well you could try to change some things in spark there is existing checkpointing mechanisms they're mostly used for ml lib and streaming i should say the ml lib thing is mainly there and both of those are only really valid within a single execution and it's mainly there to trim long lineages they are not valid between different executions so this doing something like this is to make checkpoints valid between different executions and spark what how it checkpoints things is the checkpoints things by our DD IDs which is just an auto increment thing which you can't really rely on those being the same between different executions so um and also there are for a reason I will get into there are difficulties when you have definitions that depend on the results of actions it's kind of getting outside the lazy model which many spark things fall into button out everything so for that mainly for that reason and also because this is you know perhaps like a you know kind of an experimental change I I opted to kind of build an abstraction on top of spark which you know does mostly the same thing except allows you to have your things like this so i made an abstraction called a distributed collection or DC it pretty much does the same thing as our dd's data frames and data sets and in fact behind the scenes is just a data set and you can define these instantiate them without a spark context define your whole pipeline and then at any given point you can pass in a spark context to basically get your answer out to force the computation to actually happen but until then everything is lazy and it has logical signature so when when your check pointing is enabled it can resume from a previously saved thing instead of trying to recompute itself fresh so when it kind of materializes into getting the actual data spark data set our DD or data frame it will try to see if there's a previously saved result and use that so oh yeah now I mean a data set so well that's that's a we can definitely i can definitely tell you more after but basically the api is very similar between data sets and our DBS but not completely but this basically anything you can call an RDD and anything you can call in a data set you can call on a DC it encompasses both also data frames but also save data sets also encompass with data set data frames or data sets encompass data frames that is anything anything you can run onto data frame you can run on a data set but not everything you can run onto data set you can run out of data frame well I guess sort of because the data set is a bro but it's yeah it's a little little different so so there's not much difference in the API in fact is you see the first three lines of this thing are basically the same except here we basically start with a function defined in the library that kind of holds this thing and a you know spark is lit like this will be lazy for spark this is also lazy and this will force when you Kyle get when you call get our DD or get data set or get data frame for the passing in the spark context that's when you could actually just go back to spark land and in order to get your actual thing to in order to save results or kind of force the computation to happen so it's pretty much like copy paste code so so that's one of the things I was shooting for is it should be very easy to try this out it's just you might have to change if you if you include type signatures which I am just doing this for illustrative purposes you would obviously have to change type signatures where you do that but it's usually best to write your spark code in a way where you're not enforcing those you know only doing that as necessary so when you have something like this so you can to toss in a checkpoint and you know at a certain midpoint that it will switch in and out depending on what's already saved so now here's a problem that I kind of alluded to earlier this only works if we can define things lazily here's the thing this is this is regular spark and if you see this one of the things too we're starting with some kind of some numbers during them into doubles taking a sum and then using that some in the result in the definition of our transformation say we want these numbers to add up to one we can do that by dividing every number by the sum and the problem with this is this isn't this step is not lazy this will block and execute possibly a large amount of computation I mean in this case not a very large amount but if you imagine if you had a large data set and you're doing some kind of some on it it might take a while and a problem there is the definition of this function depends on having this sum which means if we're trying to hatch that we need to have it before we hash it and to have it we have to compute this sum and that is a problem because we're trying to avoid doing this computation so what do we do well here's another abstraction weaves call it a third result and it's defined by a function on our DD or kind of you know basically everywhere and spark where you would call an action this is something that is now it instead of returning like a double or you know kind of some some result it can train contains like it's almost like a future or except it's like it it's except it's called a deferred result because it doesn't have it yet in some ways and I think in like Scala said it's a like a task although in fact and but with this so this is also something that can be part of the dag the you know the directed it's a cyclic graph that specifies our pipeline and with this we can specify entirely lazy pipelines doing whatever whatever we want so for this this example it now if we change it so that we want it to be lazy using this abstraction it looks like this so basically the first three things are almost the same except now instead of having a double we have a dr of double and we can use it in like this and so this is instead of doing this is one step now it's two steps so we're basically zipping this with is this result in so that we have our original doubles and then we have this kind of some that we will have and then we have like a tuple so it's so it's the original number and then the sum that's being passed in so the definition of this function it doesn't depend on having the sum yet because it is we're defining the function so that is taking in the sun so we can go ahead and define this function to find all of these these steps here and it doesn't block it doesn't force the computation it's defined lazily which means that we can calculate a signature for any of these or anything that depends on them which means that if we check pointed them or something that depended on them we could calculate this signature you know calculating the signatures are very inexpensive goes pretty quick and then if we had any of them saved already we could avoid redundant computation and load from there so what's the point of all this the point is we would like to write straightforward comprehensible pipelines we want our workflow to be quick and efficient we don't want to do redundant computation when we're rerunning things and trying to debug our programs especially if that reads well it doesn't it's no big deal to do redundant computation except when that redundant computation takes a long time so we would like to save on really expensive redundant computation but we would also like to avoid writing complicated code so if we can if we can write things in a concise manner and have systems make them fast for us that's good so also as long as you know going to the trouble of making abstractions that are switching things out in order to be convenient for us we can do some other things without you know really a lot more effort and maybe maybe a little more effort we can do things such as automatic failure traps so as an example say you define you know you have your pipeline and you go and run it and something comes up where it blows up on one step and maybe it's only you know one out of a thousand things that it blows up on or work you know kind of even rarer than that what people will generally go and do is that okay I got some problems what exact look kind of email maybe the problem is obvious from the message that pops up but depending on the situation you might have to do some analysis to see like okay do I only have one kind of problem or are there a few or can I get a better sense of this and so something that people will often do if the answer isn't immediately obvious is to go and select all the items to all the elements that are you know throwing an exception and kind of do further analysis on those and maybe you can you'll find all the distinct error messages because you might have more than one kind of exception I've done so so we can do things like we get and that's replicating that behavior automatically is not a not a you know difficult thing to do it's basically it's general people generally do about the same thing every time which is you know throw a try except isolate the things that are failing and then also capture capture the message so that you could essentially have here's all the things that would fail on this function and here's the exception they would throw and you can have some to statistics and maybe generate a map of like here's the error message to the number of times it occurs with you know with these elements and you know maybe distinct on the elements and all that sorts of things there's a few things like that that can just happen every time instead of us and it can happen on the actual the the the execution what you know one execution instead of us having to run into problems write some new code run that again in order to see the problem it can also mean that the successful results you can have if your failures aren't you know aren't really common you could have successful things passing through normally and maybe you encounter several failures at the same time and I've able to to look at them instead of just encountering the first failure you run into generally better since these things can take a long time to run to see all of the problems and fix multiple things at once rather than fixing things one at a time if you have to wait a while to find them so can use this yes is recently open sourced its its requires spark 20 and it's still kind of early you have to build from source I'm working on getting it on spark packages and it's definitely not I would say this this is alpha you know you it but it's definitely something that if you're building it from source you can feel free to tweak it and you know if you're curious to add an automatic matic behavior of your own that's definitely something that you should be able to do if you care to do the sort of thing so now questions um there's another developer right here but it's generally there's there's perhaps than the three contributors in total it's been used for an internal project at Bloomberg awhile like a couple and it's it's generally useful if you are repeatedly running sparks submit to something that might take a while to run so the question is do you plan to add support for Python and are and the answer is so currently this is only a Scala library and this definitely could a similar thing could be done in Python the only the tricky thing is to basically be able to hash Python functions in a way that is consistent and i would say that for data frames this is actually a fairly easy thing to do for data frames except for capturing UDS UEF so you have to do the same kind of thing where you hash functions because in the particular language but for most of data frame operations that don't involve UDF it's just a matter of taking the string there is a when you for any kind of data frame operation it'll you can grab you know kind of that column expression you can grab it as a string and the that string will be unique for the logic except as it depends on any u 2 f's so for data frame things it is something something that's portable between languages or could be but I would say not yet for for Python and our let me it it does the same way as any you could you can import it into a notebook and use it like that or in the shell if you have more questions on we can go up to you later yes um so the deferred okay so so the question is is there a way to you know just kind of not use this deferred result thing and just just make your action happen and the answer is yes so the deferred result thing is really for completeness so that you can specify any arbitrary pipeline but if you don't need that you don't need to have to use it and if you so the point of this is to be able to specify everything that's lazy it would be able to specify everything lazily and if you just want to save your results the only thing you need to do is whatever kind of result you want you pass in the spark context to that DC you know you get your DB or your data set or your data frame and you can tell that to save your back into spark land now you can do whatever you want then so you can save it or or anything the point is is that this allows you to specify things in a lazy non-blocking define all your things and it will execute instantly but when you want to go and do your blocking operations do your save you can do that just by passing in a spark context and doing it okay so i think that's that's time if anyone has further questions come talk to me thank you everyone you