scala.bythebay.io: Holden Karau, Beyond Shuffling: Scaling Apache Spark
Recording: scala.bythebay.io: Holden Karau, Beyond Shuffling: Scaling Apache Spark
I you so yeah beyond snuffaluffagus but i guess the actual title of the talk has already been given away so it's beyond shuffling tips and tricks for scaling spark and I'm going to throw in a sneak preview of structured streaming because it's new and everyone likes new exciting shiny things they distract us from the sadness that is life ok so I'm holdin my preferred pronouns are she or her I have a tattooed on my wrist in case you forget and I'm a software engineer at IBM spark Technology Center in San Francisco just like almost all of the companies that are here we are always looking for good people and I haven't figured out how our referral system works but if you are interested in working with me I would love to chat with you after this talk and I've previously worked at a bunch of interesting places that were all a lot of fun sort of doing search and recommendation problems and those problems very quickly become big data problems hence how I am the person I am today I'm a co-author of learning spark and some other spark books that will try and get you to buy several copies of you can follow me on Twitter for a really really depressed Canadian watching america what did you all do I thank you um I've got slides they're not as depressing and business numbers and code and when the videos that I get don't suck I put them on YouTube and you can check them out here there's there's some videos of similar talks and sort of more intro talks if the stuff goes over your head or if this is boring I've got some more advanced talks to and just to make it really clear yeah I'm friends I'm queer I'm Canadian I'm in part of the leather community and i'm really sad with what y'all did and if y'all need a hug today you can come find me and i'll give you a hug if y'all did mean things please don't do them again if your family didn't mean things please tell them not to I I like living here and I want to keep living here and okay anyways so happy software time yay we're gonna be excited and optimistic and I have graphs that imply optimism so 2016 hasn't been all [ __ ] we've written a whole bunch of really awesome spark code look it goes up and to the right that is a good thing on the spark technology center which employs me does a whole bunch of awesome things n is involved with the Scala foundation and a bunch of other places as well so if you're like a really excited Scala person I think you should come and chat with us even if I can't recruit you I love just chatting with more skyla people and I have a graph which strongly implies that we do a lot of work strongly implies and this graph may or may not have been selected because it's the one where my employer is in the first position but more realistically I think it's it's fair to say we're in the top three your top four and four starting from nothing that's pretty awesome and so if you by IBM support contracts or have a mainframe keep doing that I like getting paid and if you don't buy support contracts I don't get paid I think I don't know one's made it super clear where the money comes from oh so just keep that party rolling okay so more seriously we're going to talk about a bunch of spark stuff and not just random [ __ ] I'm going to start with sort of my assumptions about y'all as an audience will talk about reusing data and spark we'll talk about some considerations for working with key value data will talk about how sparks equal and data frames can be really awesome we'll also talk about where they're going to be really sad so that we know when to use them and we don't repeat 2016 and we'll also do a brief introduction to structured streaming and so we can essentially think of this talk is going from this pissed off cat which is going to claw her eyes out to this kind of happy unicorn and in case you can't read it it says I can't believe it's not MapReduce so I'm hoping you're all nice people I'm hoping you don't object to pictures of cats or David Hasselhoff I gave this talk in Germany once David Hasselhoff did not go over so well oh how many people know some Apache spark okay and now I know this is going to be really intimidating given the answer to the previous question how many people don't know some Apache spark and it's okay I just want to know cool that's awesome so parts of this talk aren't going to be super amazing for you too but we can chat afterwards and there will be some good follow-up resources that if I think trick you into thinking this is fun you can you can use to help understand more of its and you're probably no longer distracted with Pokemon go except if you are come find me after the talk because I need to find a Lapras so what is spark well poor for those two it's a general-purpose distributed system it has really nice api's it's written in Scala but it also has a Python API and our API it's a really active Apache project and while I have some feelings about how the apache software foundation runs I think it's really good to to have sort of more than just one company involved and have this arching governance so that's kind of important and useful so you don't have to worry about us like running off and trying to like make this closed source or something you can keep the lying on it and most of you came to this because it's faster than MapReduce for those of you who may be started working with large data problems after MapReduce sort sort of had its heyday go down you may be king just because your problems were no longer fitting on a single machine and you were like so what what do I do now like seriously so cool rtds are sparks primary obstruction we can think of it as a distributed collection and it works pretty similar to distributed collection except for the times when a dozen data frames if you're from a Python background you need to lower your expectations but they're really awesome and data sets give us a compile-time type version of data frames and they're fun spark context is sort of our window through the world and for anyone upgrading from spark 16 to spark 20 whenever I say spark session you can just replace it with sequel context in your head and it will be close enough yay different pieces of spark um but this is this is important we've got structured streaming now exciting we can have screaming data with a sequel engine and I'm lazy so that's great um the the language api is although here we're going to focus on scala because you all like Scala hopefully the graph tools part is a little sad a little sad don't use them they don't like they're not maintained but for some reason we still ship the artifacts so please don't use them and then there's community packages which contains alternatives to this stuff that isn't maintained anymore as well as a whole bunch of other exciting things like tools to test your spark code and stuff like that and I was really excited that there was a talk on on Scylla check because I think scholars check is really awesome and there's community packages to make spark work with Scala check so you can like automatically generate data and it's exciting but the the sort of take away from the slide is spark is more than just a distributed system it's a whole bunch of tooling built on top that distributed system to make it really easy and it all works together so rather than having to like install hive and then install my out and like a whole bunch of different tools you can just install spark and like give it a list of packages and you don't have to copy your data between systems so I mean most of you seem to already know spark so that was probably boring just like this cat okay so you may have thought that by coming to this talk you were going to get away from the word count example the dreaded word count example no no I have a big data license and I don't want my big data license taken away so I have word count in every single one of my talks not a guarantee so I can maintain my big data licensing but let's let's look at this big data license approved word count now some of you might be suspicious because it looks like that cat over there is going to have a bad time and the error code for 17 is also like not a good number either so okay this is maybe not what we want to do but if you're looking at this code it looks pretty reasonable right we're loading in some data were you know tokenizing it with the stringtokenizer of spaces it's kind of shitty but whatever we're making these word pairs and we're grouping them together by key and then we're just computing the the number of words for for each individual word and then we also go ahead and we're like well what's interesting but I crap i should have changed those two errors whatever I also you know I'm processing some log files so I'm actually really interested in knowing how many of the individual lines contain the word error because like maybe I'm worried about how many times I'm getting stack traces you know but okay so this this looks pretty reasonable but what's wrong with it yeah group by key so group by key is going to be the second one and we'll talk about the other one first so what would I skip the slide so you might have already seen it what is the second thing wrong with this slide what oh the flat map at the flat maps okay like flat maps fine right we don't catch the RTD so that's the problem in spark when we when we use our data twice we need to give spark a hint to let it know that we're using it twice and we can specify all of these different cash levels really for the most part you you probably memory and memory and disk or the two that you care about the most and also the underscore to and checkpointing are the things that other people don't often think about how many of you run in a shared cluster environment that's a lot of people do you ever experienced job preemption anyone okay I'm seeing some nodding heads but not a lot of hands so we'll pretend that you do so if you experience job preemption that can be really challenging if you have our d DS which are really expensive to compute right because spark spark will handle that just fine right like our job will complete but we'll spend a lot of time recomputing things especially the really expensive things if we have like a stage that takes like eight hours it's going to be really painful when we have to recompute like parts of that our stage so we can we can actually just you know tell spark like yeah this data is actually super important put it on two machines for me just in case but if your data is like actually incredibly important and like it took you eight hours to compute and involved a bunch of shuffles so that like spark can't do it's really awesome part where it just recompute San arrow slice of the data it actually I asked us to recompute the entire data you might find yourself considering using checkpointing and checkpointing is not really labeled in the documentation is something that you would use for this but it works really well what it does is it sort of goes back to the old MapReduce days and says like I'm going to write this out to persistent storage and you know if my cluster disappears in the middle of the night it's going to be okay I'm just going to go back to persistent storage and read this back in and when you check point you should always cash because if you just check point you'll have to read it back in from persistent storage right away because checkpointing is an action which actually forces materialization of your rtd but if you cash and check point you're really expensive data you know it'll go a little bit better awesome so the other problem is that we were doing group by key and group by key is is bad but there's there's a bunch of other bad things that we can do with key value data and how many people in the audience would describe themselves as engineers like data engineers okay that's most people how many people in the audience would describe themselves as data scientists it's like five cool so for those five you probably got this right you're used to thinking about what the distribution of your data looks like right like that's your bread and butter for the data engineers we're often used to like building really cool pipelines and being like nah I got a bunch for data it's going from A to B what's you're talking about distribution no it's cool um but I think as data engineers it's important for us to steal some things from the geoscientists and as we're working just be more aware of what the distribution of our keys look like because we can really sort of screw ourselves up if we get into places where we have large amounts of key skew and and this won't always be obvious right like sometimes our jobs will just start running much slower than they used to and the problem can be that the distribution of our data has changed but I like we had some local marketing in one place yeah so key skew is important because it occurs naturally in a lot of data right if you've got stuff group by zip code or really anything with humans you tend to get a lot of key skew even you know a lot of other distributions end up having a lot of skew towards one value especially if that value happens to be null null tends to be the most popular value of Mike Eve because I work with the cleanest of data and so so and and this can be confusing right because we're used to dropping null from from our data when we're thinking about what the distribution looks like so we can be like no it's evenly distributed it's fine but it's like no oh crap eighty percent of my records are actually no [ __ ] um and group by key will explode it's easy to make group by key explode but what this key skew could actually impact our other operations right our unbalanced partitions could cause sort by key to fail right and like you don't think about sort by key failing right like I'm just sorting my data what could go wrong a lot of things even even if you don't have like failures you can end up with stragglers right you're going to have a lot of data is being processed on a single machine and that's kind of not the point spark that the point of spark is to split up artwork so most people already know group by key is evil just like this cat and the part about group by key let's evil is that it groups all the records with the same key into a single record and and part of what makes group by keys so confusing to people about it being evil is that in a lot of cases when we ask spark to do a bunch of things spark is able to be like oh you did these three things in a row i'm just going to like push them together and like i can optimize the hell out of this right like part of the awesome part about spark is we don't have to think about like i did a map and then I need another map can I actually put these two together or like put these two pieces of data together and do a bunch of things in a single stage like spark does that for us but there's this catch it's not able to help us all the time and and that's the the problem is it can't see inside of our lambdas right so when we did our group by key and then afterwards we did a map values with a sum function inside of it spark can't see inside of what we've told it to do inside of map values it doesn't know that we want a computer some all it knows is that it wants this giant list and we want to do something on that giant list with the natural exception of sparks equal and data frames where instead of taking arbitrary lambda expressions we take sequel expressions and this is really important because when we are working with sequel expressions spark is actually able to understand what we're doing and it can it can actually pipeline these operations for us and group by key become see and so this is my handlebar mustache example so let's say that I've given up on tech because of reasons that do not bear going into and I've decided I'm going to open a mustache wax store in San Francisco and to do that I need to know where the most handlebar mustaches are so naturally I want to compute the distribution of handlebar mustaches across our fine city but the problem is I called group by key and then I got this giant record and red is bad red means failure so this record is too big to fit in our single machine and that's that but it's okay we can we can actually just compute this another way right we can give spark the opportunity to use reduce by key so instead of our group by key here we're going to go ahead or an i use reduce by key and really any time you find yourself doing group by key you should wonder if you can solve it with reduced by key or aggregate by key and in those cases spark can actually see oh this is the operation you want to perform right like i can start aggregating my data together i don't have to make this giant list i can aggregate it sort of incrementally and this is genuine big data 48 kilobytes but we can see here this is this is group by key and i wanted to actually succeed so we could see the difference so this is 48 kilobytes and then if we do reduce by key we see our shuffle greed is 11 kilobytes and 11 is smaller than 48 by a factor of four so that's good well not quite but close enough so so why is that it's because we get this free map side reduction and what that means is when we use reduced by key or aggregate by key not only do we get the benefit of like having our job actually work even if it would have worked otherwise we get this benefit that we combine the data on each individual worker before we start sending it across the network so if you've got some duplicate keys on some of your workers it'll get reduced a lot beforehand wonderful what I say to that distribution of the keys are not yeah yeah so if you if you have a most like five thousand records for okay so the question is I I'm gonna rephrase your question because I think you meant something else and if I didn't you can you can correct me so the question is like if my distribution isn't too skewed like can I just use group by key is there anything wrong with using group by key and then there's like no you can use group by key the question is like is there any benefit to using group by key for your operation like if it's a lot simpler to write with group by key and you know for sure there's only five thousand records for for each key then you're good sure yeah yeah so that's fine yeah if you're building something where you have like you need to build a separate lr for each thing and you've got like a few thousand points for each key like it's okay you can do this I'm just gonna be giving you some side I and like don't be surprised when your data source changes and everything goes to hell sorry I'm a bit cynical taking a few years but yeah right yeah no I've dealt with too many data sources where someone's like no our distribution is totally one hundred percent not skewed I assure you and then like eighty percent of your records are now ed what the hell right right like and and what works today may not work tomorrow right like data sources change frequently without my involvement hopefully because like I'm really lazy and my desire to get paged at two o'clock in the morning is pretty low I stopped working at amazon for a reason um shout out to anyone working at Amazon with a pager I feel your pain sorry so right okay even if group by key isn't going to crash right you should still consider using reduced by key or aggregate by key because you're going to get this free map side combined right so even though group by key would be successful if you can reduce your data on each worker locally first you know you'll just have less Network transfer unless disk access as a whole and that's just generally a good thing you know there's there's no magic inside of aggregate by Keith oh right like if the operation you're doing fundamentally involves creating a list of all of your records yeah okay fine you can use group by key and it's probably going to break one day but that's a problem for future you and now we come to the oh crap I forgot to update this slide oh well and now we come to the exciting sorting our data can fail sly yay so there's this unfortunate partition size limit of two gigs that we've worked to get rid of a few times and then always just sort of in like wow that's a lot of work you know what we'll just tell people to make smaller partitions and that part's worked out really well for us uh-huh in that it's no longer my problem but and it also gives me a slide to talk about so that's awesome right um so if you find yourself like running into really weird problems that make no sense during a shuffle stage you should be like huh I wonder if one of my partitions after my like shuffle is like larger than two gigs and if you find that's the case well we know what your jobs failing i abdicate all responsibility and tell you to make more partitions but there's a catch right if we're sorting by key there's sort of this limitation or even if we're using a hash partitioner if a lot of our data has the same key spark uses these these partition errs concepts which are which are really awesome but they have to be deterministic and so what that means is all of the data with the same key always has to go to the same partition so if I've got more than two gigs of data with the input key spark isn't going to be able to split it up amongst two partitions for me even if I tell it to like shuffle to 10,000 partitions it's going to send all of the data about my handlebar mustaches to that one partition and I won't know where to open my artisanal handlebar mustache workshop and then this is the sadness right like I can't do my direct mailing campaign about handlebar moustache wax and that's just a terrible thing for the economy but it's okay there's a solution and you too can send spam about handlebar mustache wax okay right and here is the bug which most people wouldn't notice but one person did once and I forgot to fix it this key oh wait I fixed it nevermind everything is perfect everything is fine no questions no questions so more more seriously more seriously sort of the solution to our handlebar mustache problem which is a much more fun name for search by key is we just need to allow sparks partition errs the opportunity to introduce splits right so we need to give it some sort of extra noise which it can partition on I'm the easiest source of noise is just like grabbing part of our value and using it as noise and that's fine and that works really well and you might be going like yeah but holding this this data is really not good right like I don't actually want to send direct mail 294 110 underscore a the post office is not going to deliver my direct mail ads right and that's okay because after you've done your shuffle you can immediately do a map and you can get rid of this junk part and you'll end up with having 94 110 across two partitions and I told you that was impossible but it's just impossible when it's been sent there by a specific partitioner and it gets sent by specific essentially when we do a sort by key e or a repartition spark uses a partitioner to figure out where to send the data but when we do something like a map spark is just like well I'll leave the data where it is because that's cheap and then it just doesn't know where to find the records anymore and then it also removes all of the restrictions about having to have all of the records with the same key on the same machine and so we can have our success and happiness and i can send my handlebar moustache wax related ads so that is it for handlebar mustache related jokes but there's some secret inside of the shuffle and it's fun secret magic that occasionally catches on fire and explodes so yeah and there's there's a partial fix but I argued against making it a correct fix because I was wrong um but I'm I can admit that anyways so there's this thing when we shuffle our data in spark spark implicitly creates these shuffle files and they go into our tamp directory and if we shuffle a whole bunch of data like a lot of times we could have a lot of shuffle files and normally this isn't that bad right because our machines tend to be pretty big right the only problem is SSDs became really popular and they sort of like screwed with this whole mentality that I don't have to ever worry about disk space because now people are just buying machines with lots of RAM and really small SSDs and now I'm running out of disk space and I didn't do a really good job fixing that one so this is also on you this talk is essentially things we should have done right that we didn't that you have to work around um that would be much better title anyways so there's some cool things though about the shuffle files right I was telling you that whenever we reuse data inside a spark we have to tell spark that we want to reuse it the exception is if we shuffled our data and then we're reusing it right afterwards is okay spark will just be like oh no it's cool I wrote that to disk on like that other machine we got this right so we essentially get disc level caching implicitly on shuffles for free the only problem is we get disc level caching that is completely unmanaged on shuffle files for free and even if our our dd's go out of scope right like even if we've stopped keeping track of them spark won't clean them up until there's a garbage collection event on the driver notably your driver running out of memory is not related to your workers running out of disk space and that's unfortunate right so we have this problem where spark could write a whole bunch of files and we could fill up our disks and our jobs will fail now lovely question so the question is if the driver goes down does that mean that the temp files don't get cleaned up hmm [Music] right that actually might have something to do with it indeed um yeah so when your driver program crashes bad things happen and I make no guarantees so try not to do that and you'll have a much better time I had this conversation with someone earlier this week where they're like so I keep getting an out of memory exception in my native code how do I work around that I'm like well fix your native code cuz I don't know how to make that work I'm sorry um yeah really there's there's no graceful clean up if you're running inside of yarn though it should get cleaned up unless we did terrible things which we might have done because honestly I didn't pay a lot of attention so we can reuse these files occasionally these files will run out of space and then we'll be sad but if you find that happening you can just add like a system GC event in your driver program and hope that the JVM listens to you probably will what could go wrong alternatively if you look inside of the code for the ALS algorithm in spark you can find the part where it explicitly cleans up the shuffle files it's not publicly available because it's a little flaky and cleaning up the shuffle files can have some unintended side effects so use with caution and I'm a no guarantees yay more things that are gonna go wrong who does anyone use map partitions oh we're like five people okay six people well okay that's not a lot of you but for those of you that I've ever consider using map partitions you need to be really careful with it and if you're not going to use map partitions you don't have to like worry too much but there's this thing right sometimes we want to process a lot of data and it's going to be more damn more data that's things going to fit in the JVM easily right and we want spark to be able to go like okay I have processed like half of this partition I'm just going to write it out to disk so I've got some room to parse us the rest of my partition and if you write a map partitions incorrectly you can accidentally make it so that spark isn't able to do that so essentially map partitions gives you an iterator interface and you always need to make sure that you're returning an iterator not like an iterable and certainly not a sequence and there's a lot of implicit conversions that that will just make our lives terrible and actually take these things and turn them into sequences behind our back and make our code fail it'll be set so be careful when you're using map partitions otherwise it'll get an out of memory exception one day and be sad um restructuring your code to avoid turning things into lists can be kind of painful but it's worth it trust me okay yeah this is David Hasselhoff the Germans didn't like it either um so data sets data sets are awesome they're new in spark 16 and they give us a whole bunch of really great performance and we added streaming to them in spark 20 yay um oh I should change that oh whatever there's still like experimental if the API is probably going to change as we discover things we did wrong with them how many people really like writing udfs no one awesome so data sets save you from the terribleness of having to write like high view DFS and instead you can just use like a map option on it and like have your arbitrary Scala code just mixed in with your relational transformations and I was awesome so let's look at how to do that okay well this this is like interesting I've got a data set of pandas will rob pandas and I want to find the happy ones I don't care about the sad ones and I'm interested in how fuzzy they are and that stored is attributes ero so I select that and then I'm like oh I don't remember how to call the word some because let's pretend I was drinking so I call reduce with this reduced function to compute the song okay i mean that's that's pretty easy but but the interesting thing is we've got these two sort of relational queries followed by some arbitrary Scala code and this like happy equals equals equals true thing is really awesome because if our data happens to be partitioned by which pandas are happy in which pandas aren't spark will go oh yeah is cool I'm just going to read the data about the happy pandas and the fastest data to process is the data that I never bother loading from disk that [ __ ] is real fast um yeah arbitrary Scala code having to specify the type is a little ugly we probably could have done this with like macros and happiness but we didn't anyways awesome things so this is the part that's actually way more awesome right like calling reduce on the end of like some transformations like and whatever right that's not all that exciting right but like this part is really cool right like I have a functional map on a data set and after this map right like I could actually just like register this data set as a sequel table and write a sequel expression against it right like I don't have to save it out to disk copy it to another system I can just like in your mixes in the middle or if like some business analysts like gave me some stuff and it was like oh it's kind of slow I just want to rewrite this one piece I can just go ahead and rewrite that one piece with my like happy Scala code and then you know I can put their sequel above it and below it and it will work it'll be awesome and I don't have to rewrite their entire thing to make it go vroom vroom I just have to rewrite this low bit I don't have to write a UDF either because UDF sir sad and they can be really fast this is in Scala in the JVM so we all know that group by key is bad but the the interesting thing is that data frame operations outperform our DD produced by key right and actually even if we're doing the same operation many times and we give the JIT a chance to warm up they still outperform it and there's a few reasons why we essentially get this sorta bolt rate for free and essentially what that means is that our serialized data can be sorted without having to deserialize Andrey serialize it so like shuffling data and data sets and data frames is much faster than it is traditionally in our dd's we also get more efficient like storage so like it's going to take up less space and that's just generally a good thing there's going to be less network traffic so that's also going to help us go fast and it's generally good if there's any heretics in the room using Python you're really going to want to use this because your your your group by key is still really bad but reduced by key is still not that good right and that's because there's this sort of like extra overhead of working with non JVM languages inside of spark but data frames and data sets let us get out of this sort of extra overhead by taking our sequel expressions and just evaluating them in the JVM has a wonderful question data sets lower than data frames ? answer they're not the same thing no worries so whenever I talk about data frame you can just think of data set where I was too lazy to talk about the types um but there's going to be a bunch of places where it explodes so like sadness just be aware of these things whenever we've got new happy things there will be problems you need to be aware of them this is a workaround for lineage explosion so if you do really really like era of algorithms data sets don't handle it super well you need to like sort of round trip it through an RDD the reasoning is if you want to talk about it afterwards with me we can have a drink or two and now we have streaming with them yay streaming so streaming is the new happy thing right like we're all like yeah I'm gonna take my batch etl and turn it into magic streaming code so let's look at how to do that okay before we do this cat wants to remind you that it's alpha code so we can look at how to do it but if you use this in production it's on you all of you and I gave this talk once and someone came up to me afterwards and was like so I want to use this in production don't be that person okay so it's not super important to like look at this code because I'm running short on time but the interesting thing is while we don't have data type inference for streaming data just yet because we can't see and or sample the whole data to figure it out the API is really similar to just the regular data frame or data set API and in fact doing our operations this same code would compile against the regular dataframe API right I can figure out how related coffee count and happiness are and I think it's just strictly increasing but there's other people that say it's like a weird function I don't know why maybe you have six coffees it's bad but at eight it'll go back up to happy join me in the research and then we can just start it and we've we've got this awesome stuff and we've got streaming data and there's just this slide of gotchas that will go by super fast this is just to remind you really really don't use this in production there's a lot of reasons not to cool um there's a whole bunch of spark resources if you're interested in it if you're interested in unit testing please test your code I don't want another financial meltdown I don't want to be related software that I've worked on so test it if anyone works in banking please don't [ __ ] it up I've lost a lot of faith in America let me keep some uh huh and there's some like resources I'll do office hours and IBM can buy you coffee and if no one shows up I can't expense my coffee so please show up so I can expense my own coffee and answer your spark questions you should also buy several copies of all of my books except for this one which is out of date but you should buy many copies of this one because it's the one where I receive the most royalties and it doesn't even exist yet the first seven chapters are available in digital format from O'Reilly and if you give us a bunch of money if we promise to eventually give you the real book um videos I'm running low on time so if you have questions come find me in the hallway otherwise thank you so much you laughed it most of my jokes if not David Hasselhoff so thank you