Optimize Spark Workloads: Shuffling, Spilling & Skew Fixes

0:01hi today I have imanu with me as a guest he wrote uh a new book with Prashant few months ago about how to optimize your workloads hiu hi you thanks for inviting me so please can you walk me through uh the guide you wrote and show me how to

0:23find it absolutely yes let me share my screen okay I share my desktop can you see my screen you sir yes okay perfect so uh let's start with a quick uh Google search to find that

0:45optimization guide that USF has mentioned so all we need to do is just type data breaks optimization guide and here you go I think this is like in the that's that's the first uh uh link that you will find here so it's

1:08comprehensive uh guide to optimize data workloads and then you can land on the homepage of this guide yeah probably I can give a quick uh like work through of different sections of the guide and then we can jump into one particular section that Mak sense awesome so uh basically Bally

1:30this guide uh is kind of a I would say like it's a based on my and Prashant experience working with database customers over the years and this contains lots of uh tips and you know uh best practices to optimize your data engineering workloads whether you're using spark Delta Lake and data in general so this guide is uh kind of

1:52broken down into multiple let's say chapters or subsections uh we have sections around uh Delta Lake like how you can optimize uh daa like uh file layout we have a lot of uh you know tricks and tips on spark optimization like you know data shuffling spilling uh data exess data explosion then we also talk about how you can basically skip uh data when you are quaring different

2:15tables right uh we have a section on Dat caching uh how you can uh clean up your uh uh table or let's say the file underlying your tables which are not uh you know uh needed anymore and we have a section on Del Mar how how you can basically speed this up um and yeah like at the end of the book you will find some some information about Delta life tables and then then these are kind of a

2:38big chapter on uh datab cluster configuration and optimization so I think yeah that's uh that's it uh youf uh should we start with one of these sections yeah and I would choose to start with how to optimize my spark workloads makes sense I think that's the most important one so uh for spark optimization I I think

3:01there like couple of sections like shuffling spilling qess explosion and skipping so let's probably take like couple of these sections and I think rest will be like quite quite easy to kind of you know U go through later so start let's start with the shuffling so each section of the book of this book uh starts with a quick introduction of the

3:24problem statement like in this case what is shuffling why it happens so basically like we are like giving some context that whenever you are doing white Transformations like joints and aggregation uh spark basically needs to move data between workers you know so that kind of uh causes a shuffle uh and then we talk about like challenges like why shuffling is is bad or why we need to control it so basically like whenever there's a

3:47shuffle you have a lot of dis IO Network IO and all of these things is going to basically uh you know cause uh some delays in your job execution now and at the end of the section then then we have like lot of uh you I would say like tips to optimize uh for this particular problem statement like in case of shuffling uh obviously the first or the best optimization is to avoid the

4:10shuffle altoe and that you can do if you can broadcast one of the two tables that you are uh for example joining together right then we basically talk about different way to broadcast like using some you know hints in the in the in the SQL query or you can set up some threshold and based on this threshold if your table size is basically less or equal to this value spark is going to automatically broadcast your table

4:32similarly you can also use aqe to do that uh once the job is started basically AQ can uh work on the plan and optimize it uh on the go and then in the each section uh there are some you know you will see this this alert icon so this is more like some goas or some key takeaways uh of each section and in these in this area you will also find

4:54some Corner cases or caveats where you need some further tuning uh to basically achieve what you're trying to do like in this case I've talked about uh one corner case is that when your tables are uh like highly compressed on disk uh sometimes like broadcasting those tables can end up into out of memory acception because uh like these are too big when

5:18they are decompressed and you know converted to uh row format in the memory so for these kind of uh scenarios U either you can completely dis disable the broadcast or you can choose a smaller uh threshold so that like very big uh tables which are highly compressed are not you know automatically broadcasted uh again like some other configuration tuning that you

5:41might uh that you will require for example this parameter is very important because uh broadcast happens through driver and by default driver driver can basically hold up to one uh gigabyte of result uh so you need to increase this value if you want to basically broadcast some bigger tables so yeah so that's that's about uh broadcasting I I believe just one something I wanted to add that for the a handles uh

6:06let's say 80% of the optimizations by by by Design and if want to go further with this 20 remain percent then you need to be experience because if you do modify uh any configuration it can improve the efficiency but it can also uh it can also take your job down so we

6:29need to be very CC and this is what like this is what we are trying to uh share through this guide that these are the things that we have battle tested like for example uh by default if you have a doubt like don't go over 200 megabytes of threshold because we know that after this point probably broadcasting is going to be more costly than Shuffle right so these are the basically I would

6:51say the configuration values that we have kind of figured out over the years and that's what we are trying to share uh through this Skype similarly like uh in case of Shuffle you can uh prefer Shuffle hash over SAR join which might be bit cheaper in terms of execution uh you know we can leverage a cost based Optimizer uh which can basically reorder the join operations

7:15like if you are doing multiple joins operations it's better to join like smaller table first and then the bigger table that will give you more you know like faster join operation because you skipping a lot lot of data from the beginning so uh yeah these are the things that that you need to think about when you are again uh trying to optimize your sh and I have just something to add regarding the analyze table so this is something uh I

7:38think uh people should keep in mind that in case for example you added a new column or multiple columns you'd better uh run this analyz table because even if Delta capture the statistics by default for the first 32 columns uh it won't do it for the new columns you have added so yeah is analyze yeah as you can see

8:01yourself like there's a clear guideline like like analyze command is like a maintenance command you have to run it regularly probably on daily basis because otherwise you will be like analyze table generates statistics which are stored in I meta store for example uh and these stats are used by spark during the physical uh planning When It prepares the plan so if you don't have updated uh stats obviously you're going

8:24to work with the stale values and Spark will not be able to generate the optimized plan right so anyway you have to like run this command regularly otherwise obviously you are not taking advantage of CDO in a way that uh you know it should be perfect uh let's move on to another section which is data spilling so uh spilling again we start with the problem definition like whenever we are doing Transformations

8:47like shuffles uh we might end up seeing some data being spilled to disk because we cannot like spark cannot fit that data into memory okay now spill is spilling is bad because obviously if you are putting data to dis you have to serialize it then write it to disk then once the memory is kind of freed up you have to read back from disk then dilize it so all these uh cycle wasted

9:09obviously going to slow down your job so the idea would be to avoid spilling as much as possible okay and uh again we talk about different ways to do that like you can just go for aqe as you you have mentioned like aqe take cares of like most of the things uh in your spark job automatically like for example you can like the like the root cause of

9:31spilling uh one of the let's say the reason is not having enough Shuffle partitions right that means you have less number of Shuffle partitions than required so each code is doing a lot of basically data shuffling and basically taking care of a lot of data and that's like it doesn't have that much of memory and that's why it kind of spills some part of it into into this right so if

9:53you can increase the number of Shuffle partitions that problem will be taken care but again fine tuning uh number shff partition is it takes some some experience so if you can't do that you can OB obviously go for this Auto uh parameter and it will like uh do that calculation for you and again as I said like each section would have some caveats like in this case I'm explaining again the same case uh for highly

10:15compressed tables uh like which are 20 to 40x compressed on disk uh this parameter might not give you exactly the right value that you are looking for uh so again this section explains how you can find out first like whether your table is highly compressed on disk or not once you have kind of figured that out uh in that case if aqe is not giving you the right value by default there's a way to further twe sorry tweak aqe with

10:41this uh parameter by default this parameter is set to 128 MB but you can reduce this value to for example 16 MB in that case aqe will give you more Shuffle partitions and at the end let's say even if with that you feel like you need to go a mile ahead and do the manual file tuning then again again we have a section on that which explains clearly like uh what are the metrics

11:03that you have to pick from spark UI uh and once you have those metrics just put them in this nice and simple formula and it will give you like exactly the right number of Shuffle partitions required for your you know shuffle stage yeah yeah and again some some yeah and just something I wanted to add because sometimes people tend tend to think that if they have uh this uh data

11:28spelling they increase the memory of the cluster but not the first that will not help that will not help like even if you have have thousand uh CES and gigabytes of memory by default the shuffle partition value is set to 200 right so anyway you are not going to be using more than 200 CPU CES and that's a problem right so you need to fix your Shuffle partitions uh to take care of

11:51this problem okay yeah and let's move on to data qess so again we start with the problem definition what is data CU so basically whenever you are doing for example joints or aggregations uh using some business column and your data is basically unevenly distributed uh across different values of that that business

12:13column that will obviously cause some iscue because as long as soon as you Shuffle there will be one CPU code that will have a lot of data and others will not have that much of data to process so this basically what will happen is that all the tasks or course will finish very quickly but there will be probably one or two course which are basically doing a lot of processing will kind of uh stay stucks for minutes or hours and that is

12:36basically wasting of resources uh losing a lot of execution time so yeah that's that's the skew uh you can identify that skew very easily on the spark UI like if you look through this uh you know uh task uh distribution uh metrics you you would see like there's like one one one bar which is really bigger than other bar so that's obviously a task which is doing a lot of Fab lifting same thing

12:59can be seen from this uh you know metrix as well like 199 of 200 task are done and like we are waiting for one task for eternity so obviously that's a skew and lastly like same thing can be reverified if you go into these detailed summary metrics of uh different task like if there's a huge difference in the shuffle read size for a minimum and the maximum uh you know uh task obviously like

13:21there's one task which is like doing 2.3 uh gigabyte of data processing compared to the one the smallest one which is doing couple of megab right so yeah these are the ways to identify skewes now once you have identified the skes okay like you can also run this query at the end if you want to be like really really sure yeah and um like to

13:43fix skes uh again we have few different ways and I think what is important is also kind of uh try to use this in this order if possible like if you can filter out the values that are causing SK the is gone for example in some in many cases if you are joining with a column which has a lot of null values you are going to end up into SK but do you

14:06really need to join with null I don't think that makes sense like in many cases doesn't make sense so if you can filter out those uh problematic values uh that basically fixes your SQ automatically otherwise you can use SQ hints uh if not aqe also comes up with the automated uh SK optimization so basically I think this is by default activated as long as you enable your

14:29like your AQ is enabled uh you can further tune aqe like by default any partition which is five time bigger than the median partition size and which has at least 256 MB data inside is considered as skew partition but you can always uh basically lower down these values if you want to make uh like you want to further click equ equ and lastly like the manual technique to handle

14:51uh skew which we have been using for Generations is called salting yes so I've given a nice example here for salting we you can basically add some random value and kind of suffix to with your uh uh value column values where you have the qess so that you kind of can get some more even distribution and then you kind of join with these uh randomly generated uh uh keys right so

15:16yeah I think uh that's pretty much it for skes and there's some other sections about data explosion data skipping so just to kind of uh keep this video short I'm not going to go through this uh today but yeah feel free to browse through these sections and again as you will find like uh this book is designed in a way that you don't have to spend too much time like each section is targeted for a particular problem statement it

15:39takes like hardly five minutes to go through the Section and understand how you can fix it so yeah give it a try and um just one more thing I would like to add like this was released last year uh uh Prashant and I and te are also working on a refresher or a new version of this book uh that will have like few other things like a lot of content on streaming how you can optimize spark structure streaming

16:02autoloader streaming uh how you and all the new features that came out lately like in data breakes for example liquid clustering delion vectors productive optimization uniform all those things are coming up in in few months so I I Believe imanu by last year it means uh 2023 so we will have a new Rel in 2024 yes absolutely yes yes thank

16:26you thank you imanu for walking us through the spark optimization and see you in the next uh chapter thank you s bye