Scale By The Bay 2019: Rose Toomey, Moonshot Spark: serverless with GraalVM

Scale By The Bay 2019: Rose Toomey, Moonshot Spark: serverless with GraalVM

Recording: Scale By The Bay 2019: Rose Toomey, Moonshot Spark: serverless with GraalVM

[Music] all right getting started this is a talk about doing some rather improbable things with Apache spark so how many people is anyone in the room right now running Apache spark sir furless right one person is it replicating what Delta Lake does no Wow okay that's that's pretty impressive so the reason why the talk is called moonshot spark is because surfer lifts is really strange and hostile and remote environment for Apache spark so just to give you a bit of background about me I thought I was a software engineer but I kept falling back down into the data so finally I realized that I was trying to fix everything at the wrong end of the data pipeline and it was giving me a situation I can only call resting so I decided to go work with data and that's how I wound up and coach you and if you want to see something about like what I really do day to day you could see my spark AI summit presentation linked here so just in general what could be less well suited to a server less environment than apache spark it loves a lot of memory a lot of cpu and having fast local storage and having fine-grained control over all of its resources what does sir furless offer this list has a lot of great stuff on it but none of those things pretty much I mean really the reason why serverless is so seductive is because it's easy to reason about and it's easy to compose and it's supposed to be cheaper I mean it probably will be cheaper someday for some people but you know it's micro service a nice idea right I mean if you've ever dealt with a sufficiently large mono repo the thought of turning it all into micro services is amazing you know and then also just you know turning all of this awful infrastructure into code that is somebody else ops is also amazing so you know thinking of what I would want from Apache spark and serverless I want the ability to do what I do now but I want it to be faster and cheaper and less troublesome and I don't want to worry about having to granularly control my resources and as long as I'm asking for a pony I decided to ask that it should just integrate with popular logging and monitoring tools without effort on my part right that's probably not going to happen so looking just in general at the paradigm shift of serverless the whole idea is that it's supposed to let us focus on the application and be agnostic of where that application is going to run that it's going to be a stateless function and if you think of it a spark executor could be conceived of as a stateless function right I mean in reality it's not at all but that's just because that's the way it's been written that's not because it has to be that way right so you know if you think about what you control right now in serverless functions that list is relatively small you can decide to time out sooner than your cloud provider times you out and you can also choose how much memory that you allocate up to a maximum of what maybe three gig so that's that's compared with what most people running spark clusters control number of course per executor amount of memory shuffle partition like all of this stuff that's not very much right but it's kind of refreshing to write it's I mean loads of people if you try to Google about optimizing spark or how you should optimize your spark jobs you find so much it's a very complicated topic wouldn't it be great if you had far fewer knobs to turn or if that were handled for you in some way so as far as what you don't control it's pretty much everything else you know how long the you're going to depend on external surfaces and you don't have very much control over how long those take to respond you don't have any control really over the reliability of the underlying infrastructure except maybe going to a different cloud provider and you know really there's a variety of different times that cloud providers will allow functions to run so and there's a lot of obstacles here like memory efficiency and that just as a whole JVM applications have a slow startup relative to the lifetime of a serverless function and then we get into some sparse specific obstacles like scheduling and when I say scheduling that's really kind of one in the same with with resource allocation and with cluster management because you're talking about because you're really talking about what's controlling the life cycle of the the spark the spark cluster what's like yarn or maysa or now lately kubernetes you know of course the big gun shuffle all about state where are you going to where you're going to keep that state and the answer if you look on online is people really want that state so you'll see a variety of different implementations putting it on s3 message queues in key value stores and you know as far-fetched as this may seem the impulse clearly shows that people want a pluggable shuffle service as part of Apache spark and right now that almost involves either forking spark and adding it yourself or going into something like spark 3 and going into some blinking lights code to start getting that added so it's by no means easy to get this done so taking a closer look at stateless functions the very simplest apache spark serverless apache spark that you could imagine would be like standalone a standalone spark server running in a stateless function just imagine that it pops up and does something really simple and then it goes away within whatever 5 minutes 10 minutes I've just got here like a summary of what some of the current limitations are and you can pretty much see from the memory limitations that that you're gonna have to turn sideways and suck it in to make it to make your deployable under the under the limitations here I've got also here like open whisk and Oracle cloud like these are all like these are all it doesn't matter back and forth you can see like to choose your poison it's going to be a little difficult to get this working so and so essentially you've taken something that's used - that was developed a process very large datasets and now you are introducing the idea of time and memory constraints of and now we have to worry about the size of the deployable which if you build a fat spark jar like lots of people do you know it's a big jar and really but at the same time once you have a spark ecosystem for processing large datasets why wouldn't you want to use that to process small datasets - I mean why not it's only that spark wasn't evolved to do that now so let's let's let's consider an example of some embarrassingly parallel operation so just imagine that you were using AWS step function or something like that and just launching a number of stand-alone spark services to deal with that like let's say that you're running some type of a very parallel ml model or something something like that so you've just got it so you've just got it running in parallel everything's you don't really even have a cluster manager and I think everything's just running standalone so that could work right I mean every mo model for itself that could be there are actually some implementations in the wild that do this now so but most SPARC jobs are spark jobs because they're a lot more complicated to this so what if it doesn't fit into a single function what are you going to do that so you need to consider what if you've got multiple stages in order you know like that is really what a smart cluster manager exists to do we're talking about scheduling about allocation about managing tasks and making sure that they succeed or failing the entire spark job you can't really do that very easily with a step function right or sorry you can't really do that very easily with serverless workflow management so considering that like what is the simplest thing that could possibly achieve this it's not good but it's kind of an idea right so we just we have the same idea that we had before that we're just launching a bunch of standalone spark services standalone spark servers and stateless functions but now we're going to use step orchestration like and we're just going to kind of launch them and make sure that they and if they don't succeed we'll watch them again so that's really very primitive and we can also use that to split that up if we have to and this is terrible but it kind of gives you and it kind of gives you an idea that well maybe maybe you can maybe you can do this so really more viable options would be to run it on a server list container like AWS Fargate because that doesn't have a time limit and you've got significantly more resources so in practice this workflow usually involves having the trigger hit a server list function which then kicks off the server list container but or you could have used surveillance workflow orchestration you could use a purpose-built cloud provider offering so essentially there's a variety of data processing pipelines that cloud providers just offer so that you know like Legos you just stack up the different services I don't hear very good reviews of those in person but I do read blog posts that sound enthusiastic about them or finally the most interesting one you can delegate a cluster management so for server list containers you've got a variety of options now you know Fargate is alright it's kind of expensive now but it will come down in cost over time and Google Cloud has an offering so does deserve it some serverless orchestration is another hybrid approach because it's the coordination of the pipeline that is serverless so the actual lambdas could be launching anything really and if you look at accounts online you find pretty much just a lot of really interesting things being nailed together yeah that's what I love about data engineers they're determined to get to the end of the pipeline one way or another so all right and then you know you've got services like AWS glue which is fully managed and Google Cloud Data proc is actually from a service perspective the most interesting one because they actually do have a working external shuffle and what they're using for that is distributed NFS so there was a presentation I've got linked in the resources at the end where they explained how they were putting where they explain how they evaluated different external shuffle data store options and came to the answer distributed NFS so and it just came out running containerized on kubernetes like month ago maybe two months ago so this is this is really technically interesting and also Microsoft is not far behind they're offering they're offering Microsoft data accelerator for Apache spark which works with data bricks and also Azura's own HD insight so serverless offerings are coming up on on the cloud which certainly cut down on the you know it's still a bit experimental doing it yourself so and then finally delegating cluster management sorry delegating cluster management so kubernetes offers his kubernetes no is natively supported in SPARC and this is an opportunity to take the I think everyone was using yarn until pretty recently and even if you were unhappy with yarn it was what were you going to replace it with missus but but kubernetes has had alright it's not fully baked if you're still using spark submit but it's starting to be it's starting to become a bit smoother especially in spark three they are looking at there they're looking at making dynamic allocation actually useful and that is a huge hot topic right now because the problem with dynamic allocation is it's assuming that when the cluster is up those nodes stay up forever if if you if dynamic allocation identifies that the node is unused it doesn't spin the note down it just makes it available to a different job but in a server list world what I want to do is get rid of you know if that note is unused I I want it to go away you know and downscaling is a hard hard problem so you can kind of see some tickets here and sprinkled about a lot of discussion how to make downscaling seriously work so that nodes can be shut down and kind of an extension of serverless would be well wouldn't it be great if you could turn a spark executor into a stateless function and that way when you're done with it just it goes away you know if you've ever stared at a cluster that was running a job with skew and seen two nodes active out of a thousand then I think you'd probably understand the desire to do something like this it could save a lot of money on cluster costs so kubernetes doesn't support that right now and it doesn't support external shuffle right now for reasons that you know if you start looking at shuffle it's not pluggable in SPARC right now so but if you look at the spark if you check out apache spark and look at branch three you can see some interesting developments people are trying to fix this Palantir has some contributions it's yeah it's it's really it's not something that you can just use right now easily so so what can you use right now while cluster management that is build on top of cloud providers so data bricks have been doing serverless since 2017 they offer optimized auto scaling and that's a feature where you can just take a box and enter some parameters cubile is probably from a technical perspective the most interesting one because they've figured out how to run a whole spark cluster on Atlanta so that's obviously you have to hack up spark quite a lot to make that work because for one thing serverless functions can't address each other so how does this spark you know so yeah it's it's they I've got a link to some of their code in the at the end of the presentation because just seeing some of the things that they did to make that work is just an interesting look at the underside of spark so now going back into the engine room just taking a look in detail changes in Apache spark to support server lists the shuffle problem is that shuffle blocks are stored locally on the node so if if you're running a single executor per node your executor and your node are synonymous but it becomes a bit blurred when you're using something like kubernetes or you you don't really understand you don't really necessarily understand or control per physical node or per container what executor czar are there so you can run into issues where executives can be shut down but they're sprinkled across different different containers so can the container be shut down no because you have three nodes running in three different containers not three notes on one container allowing you to shut down two different containers another problem that you can run into is that at least if you were running smart inside a server list function they all have very deep in the documentation limitations on the essentially I ops so since shuffle generates a lot of intermediate files you could actually blow your resource limitations if you were shuffling a lot so the most interesting approaches aren't looking at writing externally all the time is really expensive right I mean even even the fastest key value store is going to be a lot slower than writing to local storage assuming that you have it so they're taking looks at running with attached or accessible local storage what what what have you EBS volumes SSDs mounted whatever and then only move it externally off the node when you absolutely have to so there's some of that in the SPARC 3 branch so now for auto scaling yeah sorry I kind of covered some of this stuff on on this slide but something something which doesn't seem quite ready for primetime but is super interesting is Apache Crale it's an incubator project where you just use direct memory access so no more external storage just everything and everything directly accessing everything else's memory it sounds super cool runs super fast and from what I've heard is not super stable at scale yet but imagine that I mean that could just get rid of the external storage problem entirely so now finally the scheduler problem like why these things keep converging is because it's because when you touch resource allocation you know like you're talking about it you're touching so the SPARC scheduler is is hidden deep down into the stock to change the scheduler you have to fork spark and it's you know the kubernetes contribution was a major open source improvement but you know different cloud providers could offer different optimizations specifically for their own kubernetes services or what what have you if if the scheduler could be made pluggable then you can choose the best one for you for instance the kubernetes scheduler in spark 2 for is known to have issues with starvation there is actually a presentation by Palantir which referred to it as the clown car scheduler so I've got a link to that presentation at the end but I think being able to choose your scheduler to choose the scheduler that's best for your cloud provider and for your job would be a necessity for a serverless spark now if you're interested in knowing where's the ongoing discussion about this here are some links to some tickets about that and also more abstraction why should you even be tied to spark Apache beam which is another layer of abstraction on top of on top of data processing jobs has this idea all right spark is just one runner maybe you want to use different runners and that's actually available now on Google Cloud dataflow so looking ahead besides decoupling storage and compute which is really at the end of the day what we're talking about with serverless spark I want to talk about bravia because if we can solve the shuffle problem and we can solve the scheduler problem then what can we do to make spark run startup more quickly use fewer resources run faster and part of the answer in my opinion is that is running with Grell vm and getting to a native image now is everyone here familiar with Braille vm have you heard about it you've seen some I mean there's been a lot of very cool presentations in the past two or three years especially about it but essentially you can use it to compile JVM code it's much faster you can use it to run JVM code depending on what you're doing it's kind of faster but what a native image does is it it gives you the ability to run machine code from startup onward and what we're really talking about here is the difference between just-in-time vs. ahead of time so what you have with a normal vm with a normal JVM with a normal compilation process is all right you've got your code and you know when it starts up everything just gets loaded up its pars verified whatever and then when you start running your application that is when the just-in-time compiler is going to start looking at the stuff that runs a lot it is quote unquote hot and then eventually it's going to turn that into optimized native code but that eventually could be a very long time for the lifetime of a serverless function so ahead of time compilation is everything is native up front at compile time and it can conform perform optimizations which just take too long for JIT compilation so the trade-off from that you'll see a link here to certain limitations I've been trying to I would really like to see a native image of spark and if you have any idea of what the limitations are namely reflection is a huge one code generation all right at compile-time that can be dealt with but there there are just a number of limitations with some of the older and weirder code in the stack so I'm it's you know I think it's definitely going to be an effort to get a native image but it's it's something that I really want to do because it's something that I really want to do because that that opens up a possibility of using far fewer resources to run so imagine even just if you're running spark conventionally imagine how much you could save over repeated runs of the same job so not exclusively serverless but another possible improvement is you know I'm a scholar developer and but actually a lot of people who use smart our Python developers it's it's an m/l world out there and you know conventional wisdom has it that you know alright so fine it is slow the cost the interoperability cost of using Python with smart is something but everyone always says oh it's not the slowest part or what have you which may be true but we don't really know that and what about developer experience is that is that cost nothing I don't think it's nothing so I I think that being able to use growl polygons support could drastically improve the Python experience on spark so honestly I don't know really I I think that that at least a couple of people or companies are working on this but I don't think anyone has significantly progressed yet but I have seen a blog post from from Oracle mentioning specifically compiling sparks so I think someone might be working on this so so that gives me that gives me some hope because my endgame here would be turning spark into turning spark executor z' into serverless functions which would somehow once the shuffle problem is solved be able to be able to persist the shuffle state wherever wherever you want you can choose distributed NFS sounds pretty good to me if everything old is new again I suppose so and and and then you can run it where you want you can choose the cluster management but you can spin up and down these these service functions as you want let's just say that the limitation increases lambda used to be five minutes now it's ten eventually it could be long enough to support the lifetime of an executor for the state you know for an average SPARC stage so and the compute well I mean serverless functions could be a bit too optimistic but let's say containers I mean the whole idea would be that when you're done with an executor it doesn't hang about just because another executor is doing something you know it doesn't sit I'll it could really it could bring about scheduler improvements which would allow us to keep clusters running hotter so that's anyway that's that's that's my idea with serverless SPARC so I'd like to open up the floor to questions [Applause] my slides are already up on speaker duck and if you're interested come and find me do you think to achieve is that the farc team will need to come up with a form of moving the scheduler that computer so that they can hold three houses rather than the push tap in the current education all yes it's essentially essentially the way that's being gone around right now is that the executor czar kind of being launched with parameters that with with some type of parameter that tells the app you're running on this chunk of data so it's kind of yeah I I don't always rewrite the spark master but when I do I do it in a job I do it in JavaScript you know yeah I will say I anecdotally heard an account of spark spark stannis spark executors which communicate using sqs and Amazon for this purpose so that's another way you know in other words if you make serverless functions not addressable people will find a way to make them to to communicate so with shuffle sort of portability to hack around it or do you think those people are gonna be waiting both the shuffle services actually apply the winter faith I I think there are so many different solutions anecdotally I've I've heard about when I attended spark summit I was kind of asking everyone you know aren't you running are you running spark serverless and and I heard back a lot of interesting answers you know kind of some of the work is non-public just because you essentially have to form if you want to do your own thing just to scratch an itch to run server list either you can go through a very extended process with making an open-source contribution to spark or you can fork it write your own very lightly reviewed implementation of shuffle server and go now so I think that there's a lot of that in the wild right now but in spark 3 you see some kind of larger companies which have the time and you know kind of focused open source developer effort to make contributions back to spark you know through an open source channel kind of adding in some really interesting ideas and I got to tell you every single one of these is marked do not use this will change blinking light stop but it's just just look for the external shuffle service and look for implementations of it you'll see Palantir in particular has a big change set around that so the whole idea volunteers whole and game is to is to make dynamic allocation work on kubernetes without requiring an external shuffle service so which is an interesting direction you know I'd like to take that in the other direction and say hey once you're putting the files externally why does the external shuffle service have to live on each individual node why can't it be centralized the way the spark master is scale it out have a bigger RPC pool something like that have a couple of them pull it you know like I I think I I think that the communications in the spark cluster are very adaptive to a certain point in time a thought of having an always up cluster or a single-purpose cluster that runs for a long time and it's kind of a paradigm shift to think of hey you need to decouple having all of these resources located on one box so and you know it's not it's not criticism to say that the people who wrote it didn't foresee this happening it's just kind of an evolution yes [Music] like you know what I read something about that recently but I would have to I'd have to go look that back out to be honest with you I think that they are serial they're serializing the state externally I can tell you that much so oh okay all right sorry we're running we're running over a little bit I'm sorry but I will be happy to answer any any other questions