Deep Dive: Spark Memory Management

data.bythebay.io: Andrew Or - Deep Dive: Spark Memory Management

Recording: data.bythebay.io: Andrew Or - Deep Dive: Spark Memory Management

thanks for the introduction so hi my name is Andrew today I'll take you guys through a deep dive into how spark manages its memory um so I'd like to point out upfront that there will be no demo associated with this talk unfortunately however I'll make it as exciting as if it had one so first before I begin I'd like to take a quick PLL uh how many of you here are contributors to spark not many okay uh how many of you use it in production most people okay uh you're evaluating it you know it's like you've read about it you're not so sure all right how many of you have nothing to do with spark come on it's okay all right anyway so in case not in case not everyone is super familiar spark is a general data processing platform for so-called Big Data it is fast because of many optimizations that are built into it so first of all um the most commonly cited one is perhaps in memory data sharing so when spark first began in the amplab at UC Berkeley uh hadu map ruce was kind of like the big thing in the industry so everyone's talking about how massively scalable it was what eventually brought spark into the spotlight however was that it realized you don't actually have to process all the intermediate results to dis say like after every single reduce instead you could just put them in memory and keep reading it from there and for many workloads that actually much faster uh the spark guys also realized that if you express the data Transformations as a general graph as opposed to you know the strict like map reduce model then uh you can do a lot of optimizations like pipelining and lazy evaluation and so on so uh oops I have a quick so spark is also Al fast not just because uh it's fast to run uh in particular it's also fast to write smart code and this is because of the simple uh simple functional apis that we offer in a variety of languages and the fact that you can incrementally test out your application in the Shell so other than being fast spark is also General so over time spark has sort of evolved into a platform rather than just an execution engine so users have found many different use cases that can be expressed on top of spark elegantly so for instance uh the user might decide to run olab queries on top of the query execution engine built on double spark known as spark SQL U they may decide to run real time analysis using spark streaming Etc spark is also General in the sense that it interoperates with a variety of other systems uh in the Big Data ecosystem so for example you can read data from htfs or Cassandra uh you can run spark using the yarn cluster manager you can run it on top of Amazon you know ec2 and so on the company that I work for is called Data bricks we care a lot about spark because we are the people who created spark in the first place at UC Berkeley and we offer a product that looks something like this uh our goal is to make spark or more generally Big Data easy to use and with our product uh it has like a you know nice notebook interface you can make fancy visualizations that look like that and you don't have to worry about like cluster setup and all of that complexity so we handle all that for you uh that's my company and a little bit about me I am a committer of the project and as I mentioned before I'm a software engineer at data breaks in the past I've spoken a little bit at uh Hado Summit spark Summit and other related conferences about spark so you may have noticed that the title of this talk is deep dive memory management in spark now for those of you who said you had nothing to do with spark fortunately There Were None of none of you so that was good um you might be thinking to yourself well why do I care about you know a system that I barely barely use it's like all the internal details like I don't really care about that well it turns out that for any distributed computing system out there memory is an extremely critical resource and I would even go as far as to say that go as far to say that efficient memory use is critical to good performance and everyone cares about good performance because that translates directly to business value so to illustrate this I'd like to borrow from Jim Gray's famous analogy About Storage latency so to summarize all the distracting graphics on the screen essentially what he's trying to say is if reading some piece of data from memory is like going to Sacramento which takes 90 minutes from from from here or brookley then picking up the same piece of data from disk is like going all the way to Pluto which is is really really F comparison so how this applies to spark is that uh where possible we try really hard not to write any data to disk if we didn't have to instead if we just keep reading things from memory then we can bypass the trip to Pluto and in an environment like spark memory is always a constrained resource simply because if you had more memory then you can process more data while spilling less to dis so uh however unfort everyone wants memory you know uh within a system like spark there's a lot of sources of many sources of contention for memory that makes the whole uh system for managing memory in spark very difficult in particular uh there's memory contention between the two main use cases of uh memory and Spark execution and storage in the next slide I'll go into what these meaning accrate there's also a lot of memory contention across task running and parall uh in spark a task is a small unit of execution that represents a partition in your data so for example if your data set has 200 partitions then your spark application is going to launch 200 tests and each worker is going to try to run as many of those tasks in parallel as possible so while the tasks are running at the same time they're going to try to grab memory uh each of them is going to try to grab memory and you have to decide how much memory to give to each one of them uh further with even within the task there's memory contention across operators I'm going to also going to come back to this uh in a little bit but for now you can think of operators as uh something you do to your data so for example uh sort or a group buy all right so I mentioned earlier that there are two main uh usages of memory and Spark and they are execution and storage so in a nutshell execution is kind of like the memory that you use to process data right now uh so for example when you're doing a sort you have to read all the data uh in before you can return anything because uh you need to find the minimum before you can you know start returning results so a common way to do it is to read it into a buffer maybe an array or something and then uh and then sort it there so the memory that you use there is called execution memory uh storage refers to the memory that you you use to cach things uh that will be used in the future rather than right now so storage is kind of like an optimization but it's a very important one that distinguishes spark from other big data e uh other Big Data Systems so to illustrate this more completely in an example suppose you're trying to sort a bunch of integers um and what you have is an iterator so an iterator can only be traversed once so you have to buffer everything either in memory or if it doesn't fit in memory then you have to spill some of it to dis right okay so the memory use here is called execution memory and then you might run some sort algorithm like Tim sort for example that's what we use in smar we sorted it and then we return an iterator so that's execution memory and then let's say we take this iterator and we apply some transformation to it so uh let's say we add one to every single item in this iterator and we produce a new iterator and we might try to consume this iterator perhaps by summing up all the values um so the next time however uh the next time you try to get the same iterator you're going to have to Traverse the entire chain of computation right there's just no other way to to get uh the iterator because you know by definition they can only be traversed once and this could be very expensive because you know you'd be sorting it again even though you've already sorted it in the past so usually when um when you know that a particular data set will be used again in the future it's generally a good idea to cach it in memory so for instance here we cach the sorted values in memory and all subsequent competitions will just read from memory instead of you know doing the Sorting again so the memory used to cach the iterator here is called storage memory is that clear to everyone yeah okay oops oops so uh the first challenge then uh to manage memory in smark is to arbitrate memory between execution and storage so how do we decide how much memory to you know give to the sort or um or to caching so perhaps the simplest solution here is simply you know static allocation and what this means effectively is you have uh here's your total available memory you assign a chunk only for execution and and then uh you reserve another fraction only for storage and they can't use each other spaces right this is what we've been we had been using since part 1.0 like two years ago so as an example let's say I'm sorting so you know I'm like acquiring more memory here and then some other task is acquiring some memory for storage and so on right so there's a lot of Acquisitions here and finally uh the task that was sorting realizes that there's no more memory for execution so it has no choice but to Spill the dis and remember this could be an expensive operation so um it's better to try to avoid this where possible so we spilled that to dis did some aisle and then let's say storage you know keeps going and similarly when storage hits the boundary it's going to evict the least to dis um in spark the unit of storage is a block which you can think of as a partition of your data right so so I mentioned earlier that uh efficient memory use is critical to good performance however the flip side is also true that inefficient memory use effectively means bad performance so uh back in this example what's the problem with static allocation well if we're not using any of the storage memory then really the execution memory could have used the entire jvm Heap but that's not actually the case here uh it's because simply that we reserved a chunk of memory only for storage in the beginning and execution cannot borrow from it so uh doing something like a sort will end up having to spill to dis and do a lot more iOS much more often than it's necessary so you might say well what if you know I'm I'm the user I know my application best what if I'm not caching anything so I tell spark that allocate no memory for storage because I'm not going to use it uh that's okay I mean that's actually the solution that we have been using since spk 1.0 however it's not super user friendly because the user now has to think about execution memory versus storage and they have to understand all the lowlevel details so let's fast forward to um to May 2016 what are some things that we could have done better well so the problem originally was that execution could not use the entire jbm Heap even if there's no caching so what if uh so here sorting what if we just let the sort operator you know borrow from the storage space like this okay um and then after it's done it's going to release some memory so uh so this is what I call unified memory management uh which effectively says there's only one unified uh space that execution and storage share so it's not like preserved in advance reserved in advance uh which one gets how much and this has been the case since spark 1.6 so uh the critical thing to to note here is that uh in this particular example the sort didn't actually have to do any dis iio at all it didn't have to spill to dis because there's no need to you know the memory is big enough to fit all the sort of data but in the previous case it would have spilled to dis and that would have been much slower so this is kind of like a silly example where there's no storage uh now let's say there is so what happens so you know we're sorting we're sorting we're trying to acquire more memory for sorting uh and then we try to acquire more memory but there's not uh there's simply not enough so we're going to evict some blocks to disk right and then we're going to keep going and evict more blocks if necessary okay so this is what happens with unified memory management what about the other way around so let's say there's like some memory on each side and then we're putting in more blocks we're putting in more blocks and uh we try to acquire more memory for for putting in a block but there is not enough so instead of spilling execution here however uh the storage will spill itself so effectively you're freeing up space um by kicking out older blocks to accommodate your new block so there are some design considerations here right um you notice that in both cases we decided to evict storage as opposed to execution memory why do we you know make that decision well so the main reason is is because spilled execution data is always going to be read back from disk right so you're always going to incur some disc IO there whereas cash data may not be so anecdotally I can tell you that I've seen many users you know they they just abuse the cash operator in spark you know they put cash after every single line thinking that it would make things faster but at worse it would you know destroy your cash locality and it's really not recommended uh perhaps more common people cash a lot of things and they never never uncache it so you know you have this uh data in your memory that you're never going to use and meanwhile that space is not available to you know to computation so you're going to have to like spill to this unnecessarily so uh so this is why we prefer to EV storage rather than execution it's because uh for storage you know you may never have to read them back uh well you might say well my graph my my application is like some graph heavy uh comput involves a lot graph heavy computation uh which relies on caching so I'm going to have to uh keep reading from the same data set over and over again so that you know if we do exactly what I did there earlier then you'd destroy my you know my application so to fix that uh we allow the user to specify a minimum amount of cach data that will never be evicted you know this is not a reservation uh which is different from the static Al loation model uh in the sense that it's not reserved in the beginning so if you don't end up catching any data you can still use that space for for execution all right so perhaps that was the hardest uh challenge to solve um however once we figure that out we still have to figure out how to arbitrate memory across task running in parallel um so as an example of this uh perhaps uh suppose my worker has four cores uh perhaps the simplest approach we can do here once again is static allocation which says each task gets exactly 1/4 of the total memory and the amount of memory that each task uh actually uses is completely independent of uh what other tasks use so for instance I have four tasks you know they they just use some arbitrary amount of memory let's say the task in slot three finishes so the problem here now is there only three active tasks but uh but neither but none of the running tasks can use any of the memory that was allocated for slot 3 even though you know there's like it's just not being used so that's an that's another source of memory inefficiency so instead what sparked us is essentially allow the share of each task to depend on the number of actively running tasks rather than like the total of uh maximum you know possible running tasks in the system so for for instance here I only have run one task running it can use up all the space that it wants right but as soon as another task comes in this first task is going to have to spill to this uh this is because we want we want the memory allocation to be fair in addition to efficient so it's going to spill to disk and essentially now we have two slots as before so they're going to take up some memory in their slots and so on right if more tasks come in we're going to force the existing ones to spill so uh if you compare oh sorry um so the benefit here is that uh let's say all the other task finished so task three is the last remaining one it actually gets all the memory because the number of active active task is now equal to one so this is uh this is a solution for mitigating stragglers which are essentially tasks that uh take so long that they block the entire execution from completing and this has been the case in spark 1.0 two years ago uh it's been used by a lot of in production we haven't had much problem with it so if you look if you try to compare these two solutions uh you quickly notice that there are more similarities than differences uh in particular it's easy to know that both of them are fair and starvation free for static allocation is pretty much the definition of fair like you can't you literally can't get more fair than that um as as for What spark does it's pretty similar to the static allocation case in terms of fairness uh because you know if is essentially doing the same thing except the slots are determined dynamically rather than uh rather than yeah they're determined dynamically uh an obvious advantage of static allocation is that it's simpler but then what spark does handles stragglers which are like the last remaining test a little better finally uh the last challenge that I wanted to talk about today is how to arbitrate memory across operators running in the same task so to give you a quick example of what I mean by that suppose I'm running a simple query um you know I'm Computing the average heights per age and ordering them by the by that average height right so there are two ways of expressing this first you can write a simple SQL query like the one at the top or you can use Sparks data frame syntax um like the bottom to express it in a more sort of functional way so these two both produce use the same uh query execution plan on the on the left that you see right there so it's going to start by scanning your data you know projecting some columns and then Computing the average that's the Aggregate and finally sorting it so let's say in this example uh the worker has six pages of memory so a page here is a fixed size of uh is fixed chunk of memory that your jvm has and it's going to allocate them to the operators so scan and project they don't really use much memory they use some but like compared to Aggregate and sort they really don't use anything at all so let's say in this particular example aggregate is like super memory intensive it grabs all the pages very quickly and what is it using the memory for well to build a map that looks like this um because we're Computing average we could technically have stored uh the sum of all the Heights and the total number of heights to to compute the average later but let's say for Simplicity you know we just naively store all the Heights and then finally when we get to sort uh while the system doesn't have any more memory so you know sort just can't do anything it doesn't even have a page of working memory to work with so this is what I mean by memory contention between in this example sort and aggregate so there are several Solutions here right the first simpler one is to Simply reserve a page for each operator so in this example uh I would reserve a page for Aggregate and sort because those actually use memory right um so this does work however there's one problem uh if we keep going with this then aggregate is going to get five pages right and then sort it's going to have one uh it is now starvation free but it's still not really fair so sorts going to keep spilling to dis and you know that's going to eventually make your application run slower so it's still not fair uh perhaps the bigger problem here is it's not really scalable either like what if I had more operators in this particular example I only had two but like if I had seven then we just can't simply reserve a page for every operator so by the way uh this is not just like some silly approach that I I thought of like to make the real one look better this is actually a real approach that we uh decided and Spark and then later decided to revise it because of uh uh real workloads that we observed so um that's why I'm talking about it so now for the slightly better approach that is in fact in spark um what we could do instead is something called Cooperative spilling so let's say aggregate once again takes up all the pages and then when sort tries to acquire memory it realizes that there are none so what it's going to do is force aggregate to spill a page to free memory so effectively it steals a page from aggregate so then let's say you know a sort is like U expanding its array so it's trying to to get more memory so what it's going to do now is keep forcing aggregate to spill um until it finishes so let's say it finishes with three pages so now um aggregate doesn't have to spill its remaining pages so this kind of ties back to the principle I talked about earlier where we try to keep as many things resident in memory where possible so we don't have to go all the way to Pluto right in this example you can see that U it is fair in the sense that you know both sort and NRA end up with three pages but even if sort takes up all the pages from aggregate it's still fair and this is because aggregate must have already finished by the time sort started so uh it doesn't actually need new pages so it can spill at most you know six pages right and this is uh implemented since spark 16 which is available uh the be since the beginning of this year all right so as a quick recap uh there are three major sources of contention that makes uh memory management difficult in spark in particular they are arbitrating memory between the two main usages execution and storage across task running in parallel and across operators running in the same task so perhaps the U the central motiv motivating theme across all three of them is that instead of statically reserving memory and events simply allow everyone to you know take a of it whenever they want however um when there is memory contention when someone's getting more than their fair share uh impose you know fairness by forcing them to spill so this is the heart of essentially the memory management in spark so uh I can't give a talk about spark memory without also mentioning project tungsten how many of you have heard of tungsten that's more than I thought actually so um what it is essentially is uh a project to Leverage The lowle you know bite level semantics of data in spark so to push U computation to you know closer to the hardware where possible so this has three major parts I'm only going to talk about the first two since the third one is kind of unrelated to this talk so the first one is uh storing data as like bytes you know in memory as compactly as possible and the second one is uh taking advantage of the fact that uh you know memory is a lot faster in dis but l123 caches are a lot faster than memory so let me talk about the first one first um so Java objects in general they have very large overheads so for a simple string like ABCD you really only need four byes to actually encode it uh but then in Java this string becomes 48 bytes it has like a bunch of unnecessary things that you really don't care about um so if we use Java objects to represent a row you know that's even worse because now we have all these wrappers you know we box integers and other Primitives um we have a lot of random overhead and hash code is more expensive because you have to Traverse all the pointers independently so instead What spark does using tungsten is represent each row uh as you know just bites so essentially for primitive types we know exactly how big they are so we we just like encode them in the row um directly for variable length data like strings we store a pointer that points to uh first the length of the string and then finally the data itself so you don't have to worry about the details here but uh the the takeaway here is if you store things as compactly as possible you know you have more space to work with that reduces garbage collection overheads you have better cach locality there are a lot of other benefits uh one other benefit that's not really related to memory here is that you don't really have to serialize and deserialize your data all the time anymore because it's already you know this is already serialized it's like in bites and also you can work uh directly on top of the serialized form of the data because you know exactly uh where the fields are because the offsets are determined by you not by some random you know compiler and the second part of tungsten that I want to talk about is uh taking level taking advantage of the uh l123 caches as opposed to just memory so for example if you want to sort a simple list of Records uh a very common way to do it is to just sort the pointer itself so this is however a naive layout in the sense that uh every time you want to dreference every time you want to compare two pointers you're going to have to make two memory accesses so you're going to have to go to Sacramento twice that's actually quite expensive what you could do better instead is to Simply store the key that you're sorting um or at least part of it right alongside with a pointer so you don't actually have to dreference a pointer you could directly compare the two key prefixes so this is a cache aware layout leads to better cache locality all right so uh that's actually all I have I ended up speaking much quicker than I thought I than I than I planned uh I guess we're going to break out into questions now you talk about the off heat nature of things yeah actually had a slide in there but I I thought that hey repeat the questions oh sorry um the question was can you talk about the offhe nature of things um yeah I thought that actually I wouldn't have time to talk about it but I guess I do now um so in spark you can uh you can both do execution and Storage off Heap in the sense that uh when you catch something you know you can you don't have to put it in the in the Heap and like when you do sorting also you can acquire memory from outside of the Heap as well so uh the few ma major advantages here are that uh first of all um you can scale the size of your working memory without regard to problems with large heaps like you know crazy garbage collections that could occur for minutes you know when you have hundreds of uh gigabytes so that's that's usually very expensive um so that's perhaps the biggest thing uh other smaller motivations for off Heap you know could be zero copy IO when you um when you try to write something that's already in memory directly to dis or like to um to some other uh over or when you're trying to send it over the network uh you do you don't have to do another copy when you do that um so so both of these combined will likely make your application run faster that's the high level point question yeah so the so the constraint of 32 gig previously what's the best practice now with emploees um used to be uh so off Heap is available since 1.6 yeah so um even before that however your keep size can be however large you want just like you're going to suffer from garbage collection issues so like with a then you want to study like your cash um then you be able to use like in that would it be comp you want to talk to it be easier the memory model yeah so currently so in the future that could be the case where you store something off he and then another process maybe not spark some something else other than spark could read that data um currently however the format in which we store them is like only spark knows about it so other things can't just go ahead and read it um but you know like off he is like the first step towards that you can sure did it more efficiently what about like process yeah so um integration when Tac so sorry um it's now called alexio U integration with alexio is also something that a lot of people want um so from databas side at least there are no active U you know developments towards that um however alexio the company they might have something out of Ena yeah there's definitely a lot of motivation to do it open yeah so actually um open source spark is only a small part of data brakes data brakes uh essentially its goal is to make spark easy to use and to to achieve that it has like a whole product where you know um you go there you ask for a couple of nodes and then you have a nice interface in which you can run spark jobs and then the results are going to come back to you in uh user understand in a user friendly format and that's like you know ideal for people like data scientists and data Engineers sorry can I monit execution and storage memory that's a good question um for storage you can yes EX for execution you currently cannot yes the same [Music] question the usage metric of yeah um so sorry I realize I haven't been repeating questions so if I understand correctly your questions are there metrics or um you know visualizations to show like how uh storage and execution you know how much of those are used not necessar visualization even in the Raw text format some yeah um so I'm not sure if you or the audience is familiar with the spark UI which is yeah which is um basically like a a user interface for knowing more about your application so currently uh the spark UI stores information about how much how many blocks are stored in memory on which node all of that information is in the spark UI however that only applies to storage uh execution currently there's I believe there's no information on spark UI about that yet but we we could add that that inform yes um that is currently not there sh all right sure okay to andc that I see I I'm personally not aware of any active PLS to integrate with arrow um so unfortunately I don't think I can yeah I just don't know yes you talked about all of this uh memory sharing between operators has there been any thought about sharing memory between tasks so if I had four par tasks three which use very little memory a lot share memory across put a task borrow memory from other tasks yeah so oh the question is is there any plans or or like any memory sharing across tasks um So currently the only memory the only so-called memory sharing cross task is uh is not like the contents of the memory but rather than than but but rather the space itself uh yeah so um so I don't know if uh so let's see yeah so so the the memory sharing model across the task uh in spark right now is if another task comes in you know like forces existing task to Spill and so um it does like arbitrate fairness that way I'm not sure that answers your question yes so Ides the memory layout you mention weu some scientist told me that want to use def go there to Pi memory sets I'm not yeah so oh um well the question is very long so I'm not going to repeat it but essentially uh so we don't want people to have to uh have to tweak the memory settings to get the ideal workload what we want at least in spark 2.0 onwards and probably in spark6 onwards is that you just start your application it just picks the you know the best settings for you and it just like runs without any memory problems so that's the goal that we want um however if before say like 16 that you do have to tweak various settings to avoid like out of memory errors or or even like very slow application you know that's like that's not intentional yes yes so memory management is very closely associated with Cony right and we have faced lot of issues with concurrency of spark so will the improved memory management address some concurrency issues for example can you uh if we are hitting say 500 concr users in the system right now SP doesn't scale for concurrency so will this improved memory management will help for concurrency issue that do you mean 500 people sharing the same like driver or like same cluster um and all 500 of them are like running jobs in parallel us maybe ,000 and even if you consider 10% or 1% is the concurrent users 500 is the concurrent us H Sameer or different quer yeah so right uh so the question is roughly sorry um does the new me unified memory management model do anything to address the concurrency issues see in like shared clusters U so uh to address that I think uh the biggest thing that might be relevant here is perhaps the fact that previously if you casat something that's basically always going to be there unless like someone also caches other things um whereas now whereas now uh even if you don't explicit and cach whatever you cached then it'll be thrown out of the cash uh because extion will like kick it out so this guards against people who misuse you know the caching operator for example so I believe that's time but if I didn't answer any of your questions fully or if you have you know more questions feel free to come to me happy to talk thank [Applause] you