scale.bythebay.io: Long Cao, Avoiding Spark Pitfalls at Scale
Recording: scale.bythebay.io: Long Cao, Avoiding Spark Pitfalls at Scale
right scale by the bay how's lunch time to work off that post lunch haze I know I'm sleepy and I'm speaking so it's not a very good combination so today I'll be talking about avoid avoiding spark pitfalls at scale what that what I mean this to be is more about sharing some of my production experience tips and tricks and you know some of the heartache and lost sleep I've had over the over the past year and a half with working with spark so a little bit about me I'm a software engineer data engineer on a data science team at co2 management we're a hedge fund located in New York we have a VC privates arm look in Menlo Park and so people always ask you know what it was a hedge fund doing with data science while we use it to inform traits come find me later if you want to talk more about that so I've been doing scala for the last five and a half years data engineering for last two years I live in Brooklyn by way of Texas I usually like to tell people I'm from Texas there's some for some reason I like to sink that in I'm very proud of where I grew up and a little bit about me I really like coffee and whiskey so you ever want to talk about that you can talk about that with me too these are pictures of me when I just went to Japan last month really great place so just off the bat first off what this talk is not about and I apologize it's not about machine learning it's about deep learning about AI not about real-time streaming or anything super fancy nor is it a wine fest about sparks API I know it's not perfect but this is more about you know practical things that we can do with it and and because it's a powerful framework and and something that we can actually use in production you know you have to know what you're up against what this Hawk is about is what I just mentioned before tips and tricks about spark production experience and some dirty subversive tricks that you know I'm a little ashamed to share but it works for us and it is centered on spark - - so I apologize if you're on an older distribution all I can say is it's a lot better when you upgrade so some of the major sections we'll talk about in this 20 minute talk and we're gonna try and go a little bit faster because there's just so much to discuss but you know we're talking gonna talk about knowing your data in code dealing with data skew and I'll define what that is how to tango with the execution engine there's there's some some oddities when you when you come up against what spark comes with it does under the hood configuration because there's probably like a thousand knobs in in spark that you know not all of them are surfaced in a way that's very obvious talk about the API itself and then I'm gonna give it a little plug about what's called type Philippe you guys will get we'll get to that at the end so most of these are probably truisms you know if you're an experienced data engineer you know these things to be true about working with large data sets so knowing your cardinality and the distribution of your data is important particularly because of anything that you're doing that involves shuffling data involves any sort of parallel massively parallel operate operations so GroupWise joins windowing operations partition keys I'm gonna talk a little bit more about what knowing your data means especially when it comes to in terms of skew or data skew but you know one practical tip I can give immediately since we know like spark and I'm assuming that most people have at least a little bit of experience will that spark is but if you have you know a data set you can apply a function from A to B across all the partitions what's something that's easily parallelizable this can be a problem if you have a slow function let's say it's you know calculating a bunch of things doing some heavy math or doing some heavy text extraction if you're applying this across you know n number of partitions you can run into what's called the straggler task problem you can you can at the end of your stage of applying this function because not a it's probably not true that every one of your partitions is completely homogeneous and so some partitions will finish they're mapping operation for others and so if you look at the CPU graph apologize I don't have a picture but over over time it's going to go down to zero while that stage is finishing well that's that's what we call a stereo chess problem you can get around this by repartition into finer grained partitions you kind of think of it like starting to chunk it in in a way that's more streamlined so I mentioned this before but but what is skew skew in the context of data engineering when when is is when certain keys appear more often than others in an uneven fashion I just threw up a graph that's like kind of what it what it illustrates imagine if this was a group by key and count if you have like one key that happens to show up a lot this gonna happen would say you know placeholder keys or null keys in this case it generally speaking it can adversely affect parallelization of tasks anything from group eyes or joins especially you will see an example in a bit so I can tell you from personal experience data excuse probably 90% of the heart problems we face and this is because real world data anything useful almost always has skewing it anything from like user level data geo data you know if it doesn't have skew in it it's probably been cleaned up a lot and someone has toiled and sweat to get get it to that place and so leaky when you're working on raw data sets this this happens a lot and the other 10% is serializable but you know that's a whole different story so what can happen with data skews let's say you're trying to do a joint across what's a skew key well on a high level what SPARC does is shuffles these your your records across different executor x' using that key and so if you have a very skewed key what you can do is actually cause your executor go out to throw out of memory errors and you know your join won't complete or or it will complete very very slowly you can use what's called spill thresholds which is something I'm going to talk about in the configuration section but but for the most part data skew can can either cause a catastrophic failure or just really really slow queries which is annoying right because you know big data has his promise of handling large datasets quickly you know in a way that should be elastic so so I mentioned that the joins can be can convey because of these because of data skew well you can you can do what's called a skew join which it's it's not actually in the SPARC API of any sort so you can actually just implement this on top of already existing operators but something you can do in SPARC is what's called an isolated broadcast join and this is enabled because spark has the concept of the broadcast wrong way you have one side of the data set this is very or so one side of the join that's very small you can actually create copies of that exact same data set across the entire cluster and just doing it in memory join so generally speaking I won't i won't you know prescriptively describe this but generally speaking what you're gonna do is split the data set into to split it into two sides where you you have your hot keys so you remember that graph from from back your hot side might be like that that that one key with a very large skew and you you you you split both sides of your join you join the hot side with broadcast joins the cold side with just a regular whatever join you want left joint inner join and then you Union the results so this works around that whole problem of what happens if your sort sort merge showing causes out of memory errors for the very very few key so I'm breezing through these things but these are some of you know the some of the the biggest problems we've ran into with production datasets so with sparks execution engine on a high level what it does is it takes your data set API calls compiles them into a logical plan then into a physical plan and that actually gets handed over to spark to execute so this is a separate part of what normal you know developer might might work with but the errors from spark tend to bubble up from from here so one thing we ran into is broadcast timeouts this is something that we thought was really odd because you know we would notice our data set stages were way up way up in the call chain but we were seeing that broadcasts were timing out even before it was getting to the point where he was broadcasting a dataset and joining with it so what we noticed is that you know as soon as if you have very slow stages upstream and you have a broadcast joint and you materialize that action the the timer for that broadcast actually starts much earlier than then then you would think I'm not entirely sure this is a bug or visit us by design that you have to do a bit more research about that but it's a little buried in the spark documentation but you should use you should increase the broadcast I think it's to the default 5 minutes or 20 minutes I don't remember but that's that's a tip that we we took up half data to figure out another thing is because the the spark data set API is is lazy meaning you can separate between transforms and actions one is strict and one is one is lazy actions are strict transforms are usually lazy you have to be aware of how partitioning is and how coalescing works coalescing is when you want to condense things down into your data set down into a number of partitions this is useful especially if you're doing some sort of aggregation where let's say your result at the end of the day is just 20 20 rows right you don't want that to be spread across a thousand partitions especially if you're saving that say in a blob store or HDFS because now it means you have you know thousand empty started I don't know 19 1980 empty empty partitions and then 20 partitions with one row each there's no point in that so what you normally do is call s it down to one and then save that into a blob source of some sort well that that's a problem when it comes to lazy api n and this action because this coalesce operator will actually preempt what you're partitioning seam is like and so your upstream computations will then start to execute on one partition and in in in sparklin partitions usually map to tasks and tasks usually map to a core so most people don't want their aggregations running on one task if it's specially very large large cluster you're just gonna ruin your your utilization what's recommended on the scala doc and said is to use repartition of one this will trigger a full shuffle but it will get you the semantics that you're looking for here without sacrificing performance so this one's a bit more vague I'm going to hand wave this because there's a whole class of issues here but because SPARC is lazy and it's got these semantics you can you can stack a lot of transforms on top before you actually materialize it this can cause a lot of issues when it comes to catalyst optimization or Janee no code generation you'll get some really arcane errors and it's just really hard to reason about the performance of you know where things are what your stages are doing so the dirty trick that we like to use is to cache this data set and account which essentially forces in action and forces the evaluation of that data set but because you're caching it or persisting it to disk whatever you want you at least save save the results of that computation and then move forward it remains the it remains to be said whether this is more efficient or not I just know it gets us around a lot of issues when it comes to very large query plans so next section is a configuration this is not too fun to dig around because you know if you into some errors that you don't you don't know what's going on you you tend to copy the stack trace and you know go digging in JIRA which if you're a developer like me cheers not always your favorite thing so the tip I can give you honestly is get comfortable with digging through spark docs spark source code especially if it matches with your version the spark JIRA there's a lot of gold in those hills because you know I discovered most of these things through digging through JIRA so like I mentioned before you know partitions usually map two cores something that's actually recommended is to set your parallelism and and partition settings to 2x the number of course this is actually listed in the it's written in the spark documentation but my defaults part sets your partitioning to 200 so it's something that you it's probably the first thing you need to first not be needed tuned when you play with spark spill thresholds is something that I mentioned earlier this is something that recently changed I think in spark to two but you can actually tune specific execution spill thresholds especially when it comes to any sort of operations that you if you don't have a choice and actually you know joining or windowing on some of that skewed this is really useful because you'll start spilling to disk instead of filling up your executor memory this is this is something that I can't give a blanket recommendation for for tuning but you just should just know the general size of your records how many of them are on those skewed partitions and how much memory you have on your executor is also how much disk space you have it's you should just tune it enough so that it's not spilling to disk too often but not so low that it our memories and something became a little bit more production-ready as I think sparked to one as a spark speculation blacklisting and task reaping these are all I won't go into details as a how how these work but these are all really useful for for the for the long-term health of a large cluster if you have like say one badly behaving note on your cloud provider than smart will then monitor things and not schedule tasks or or kill tasks before their before they're finished it's a really useful suite of settings to have one thing that we noticed is for for very large clusters since most people in order to track like that the progress of their jobs they look for the spark UI if you have very large very large stages with a lot of partitions say like you know fifty thousand partitions this event queue can actually fill up on the driver it's something that is pretty simple to increase it increases the reliability of your spark UI reporting so I mentioned before the the API of spark is sometimes a little a little opaque it's not very clear sometimes when you know some things are performing or not so one thing we noticed is that map versus select map is typically if you're working on a normal Scala program let's say have a case class with three fields I just want one of them well what are you gonna do when you're in a constant say that a monadic context right you're gonna map on it you're gonna take your function from A to B and so and and select that field well you know spark has that promise of doing that the same thing but because plain Scala functions aren't optimizable by catalyst is actually less efficient and say just selecting those columns so if you're just selecting columns I try just using select rather than map you may notice it's a lot more performant another thing is instance members are reinitializing crossing serialization boundaries this is a it's actually a trap that we fell into you know because we we we structured all of our data around case classes as a very nice you know record record type but and then we would stuff you know expensive members into those classes but if you're if you're paying for this cost every single time you cross these boundaries well you know you're gonna make your jobs that much slower for each time you you you you materialize these things is that a question that is a very tricky question because of maybe I I can't give a blanket recommendation there either because there's some tricky things with lazy valves as well we've run into issues with lazy valves and when it comes to initialization order so best I can tell you maybe so a quick plug since you know since the spark API isn't super super tight full by that I mean you know there's a lot of dealing with string columns a lot of dealing with even even the lambdas that we deal with with spark datasets aren't aren't optimized well there's there are other open source projects looking to make things better a quick plug for frameless which is a shapeless a shapeless driven API on top of on top of spark I actually gave an intro talk about this maybe like nine months ago check it out I think it what I would want to see is more users because more users means more contributors more contributors means better quality code we're we're definitely interested in or I'd definitely be interested in chatting with you if you're if you're interested in that and I don't know if probably the mid junk of you have had attended Fabio's talk about quilt but you know I I just recently discovered this maybe two weeks ago this is also a really interesting project that directly compiles your your monadic code into spark sequel I think this is a really interesting interesting concept because you know as I just mentioned before you know spark functions aren't introspected buys by catalyst but if you can use spark that spark sequel as a target language you can do you can do you can allow catalyst to do optimizations a bit more so all in all that was just just constant hits of tips and tricks if you're and all interested in about the work we do again I work on a data engineering team of three ATCO team management very small team but what you get to do is tremendous and the budget we have to do it is is is amazing amazing we work on alternative data sets to turn that into signal for to inform our our trades we are Terra by CL lots of billions of rows of data we're implementing a lambda architecture heavy users of functional programming we've got cast shapeless FS to HP for us SPARC on the stack or on AWS if you're interested chat with me or tweet at me I'm very active on Twitter for better or worse so with that if you have any questions I think we have time for one or two one one question all right then well if you're if you got any other questions or wanna chat come find me I'll be here all weekend [Applause]