Hello folks, thanks for all the support. Sorry for the delay on the Ebook, it's still coming it was just delayed. I will share it here as soon as it's available. I'm hoping Q1 this year. :)
Hello Daniel, when i checked spark documnetation, i found that the cache is equal to persistent with MEMORY_ONLY option, did the rdd.cache is different than df.cache??? spark.apache.org/docs/latest/rdd-programming-guide.html#rdd-persistence Thanks
This is so far the most informative and in-depth talk about spark job optimization I found on RUclips. Before this, the opt I've been doing is mostly through blindly trial & error. Thank you so much Daniel, it's amazing to see someone who can break a (sometimes) overwhelming task into basic spark concepts and apply deductive & inductive analysis.
In the lazy-loading, he filtered the years from 2000-2001 , what if the calculation should be done for all the years? Can't use a filter in this case right ?
I believe he is just using filter to reduce the number of rows to join. I think you must have the knowledge regarding the input data and know that outside a particular range, the data will have a match in the join
50:49 what is the purpose of adding a constant to each row? They key will not change, it will still be computed on 1 executor. The salt column value should be different for different rows to split large key.
Can we get a link to the slides? There are tons of small details on the slides that will be easier to go through if we have the slides rather than pausing the video every time. :)
Hi...Great Presentation for understanding Spark optimizations. Is there any Presentation slides to go through..since in videos..its little difficult to read those numbers...
Yes, you are right I had the same question initially.This can be solved by running additional group by on the obtained dataset as this will run fast.We are using 2 group by to avoid skew.
Why is the first example a valid comparison? You reduced the size of the data you are working with so obviously it will run faster. What if you actually need to process all years instead of just two?
How to set the spark.sql.shuffle.partition by a variable instead of a constant..means if the shuffle input data size is less then it should automatically choose less number of SQL shuffle partition if input shuffle data stage is more then the job should programmaticaly be able to determine correct partition..rather then given a constant valuem
OK, got it now: Each core processes one 100 MB partition. We have 96 cores that need to process a total of 54 GB. At a given time or batch all 96 cores can process a maximum of 96x100=9600MB. That means after 5 batches the cluster processes 9600MBx5=48000MB. For the last batch the cluster needs to process the remaining 6000MB or 6000MB/100=60 partitions. Those 60 partitions will be processed by 60 processors which is 62,5% of the cluster. The remaining 36 processors which is 37,5% of the cluster will be idle in the last batch. The story looks different if we had 480 112,5 MB partitions. This gives us 10800 MB per batch. And all data are processed after 5 batches with 100% CPU utilization.
@@bikashpatra119 my guess is that just because it is less than the notional upper bound of 150-200 MB/partition. There doesn't seem to be a formula that will return the amount of spill, given a number of inputs (i.e. shuffle input size and list of tasks on this stage) but since the general logic is "smaller target partition sizes result in a reduction of spill, even if we cannot predict how much spill will result, let's just fastrack the spill reduction process by just halving the notional upper bound for the target partition size of 200 MB/partition".
@1:00:28 min line 22 i.e. save(histPerfPath_1y) does it work if I run it on cluster to save on hdfs path ? in foreach if i save and run it in yarn-cluster mode it would fail to save with error... hdfs temp file wont find to save.... how to solve this kind of issue ?
In my spark version 2.4.3 job after all my transformations,computations and joins I am writing my final dataframe to s3 in parquet format But irrespective of my cores count my job is taking fixed amount for completing save action For distinct cores count-8,16,24 my write action timing is fixed to 8 minutes Due to this my solution is not becoming scalable How should I make my solution scalable so that my overall job execution time becomes proportional to cores used
I was trying to load 600 million rows to a pandas dataframe from SQL Server. It was taking too long and then OOM error. I tought pyspark will solve that problem. But after 42 minutes watching this video, I see that's better using a cluster with more RAM and raise SQL Server processor. Most of requirements to setup pyspark are not known on my environment so pyspark is useless when you are working with data you don't know details about.
Hello folks, thanks for all the support. Sorry for the delay on the Ebook, it's still coming it was just delayed. I will share it here as soon as it's available. I'm hoping Q1 this year. :)
Thanks!
Thanks Daniel, great talk. Please share the ebook, once completed.
Hello Daniel,
when i checked spark documnetation, i found that the cache is equal to persistent with MEMORY_ONLY option, did the rdd.cache is different than df.cache???
spark.apache.org/docs/latest/rdd-programming-guide.html#rdd-persistence
Thanks
any update @daniel ?
Nice 👍, any updates on eBook?
This is so far the most informative and in-depth talk about spark job optimization I found on RUclips. Before this, the opt I've been doing is mostly through blindly trial & error. Thank you so much Daniel, it's amazing to see someone who can break a (sometimes) overwhelming task into basic spark concepts and apply deductive & inductive analysis.
Really very good session Dan.
It’s my pleasure to work with you.
The best talk about spark optimizations in YT by far, thanks man!
Redirected to this page from SO, class apart content. Glad to find this goldmine !!
Excellent explanations.
Cleared so many wrong concepts of mine.
Thanks man!!!!
Thanks, your video helps me keep my job
Super talk Daniel and great insights, still waiting for the ebook though :)
Great talk! Many Thanks.
Btw, where is the book, Daniel? :)
4:45 1 Task = 1 Core can be changed using the property spark.task.cpu. The default is 1 Task = 1 Core.
Excellent talk Daniel 👍 I wish I saw it when I started with Spark :-) How's it looking with mentioned e-book?
In the lazy-loading, he filtered the years from 2000-2001 , what if the calculation should be done for all the years? Can't use a filter in this case right ?
I believe he is just using filter to reduce the number of rows to join. I think you must have the knowledge regarding the input data and know that outside a particular range, the data will have a match in the join
+1. I think the idea is to filter as far upstream as possible. Don't do a cartesian product and then a filter if you can filter is ahead of time.
good job Daniel Tomes. It help lots
Great talk! Really learned alot, looking forward to the book!
What is the name of the book? When will it come out?
Great talk, with a lot takeaways!! Is there any references to the notebooks with datasets so I can recreate some of the optimizations?
50:49 what is the purpose of adding a constant to each row? They key will not change, it will still be computed on 1 executor. The salt column value should be different for different rows to split large key.
thank you so much for explaining slat addition clearly.
Can we get a link to the slides? There are tons of small details on the slides that will be easier to go through if we have the slides rather than pausing the video every time. :)
Hi...Great Presentation for understanding Spark optimizations. Is there any Presentation slides to go through..since in videos..its little difficult to read those numbers...
if you are adding "salt" column in groupBy it would give wrong results right ... if any groupBy function results we required ?
Yes, you are right I had the same question initially.This can be solved by running additional group by on the obtained dataset as this will run fast.We are using 2 group by to avoid skew.
Thanks, Daniel, great talk. Please share the ebook link.
If a system is created that you have to tweak it so much and understand it so much to get good performance I would say it should be redesigned
I am curious, if setting partition sizes that small, would cause a small file issue or maybe I am missing something. Can please someone answer?
Is the e-book available?
what's the link for range join optimization reference?
Why is the first example a valid comparison? You reduced the size of the data you are working with so obviously it will run faster. What if you actually need to process all years instead of just two?
I have the same question.. didn't get how reducing the number of input file is an optimization.
How did you come up with the 16mb maxPartitionBytes? is there a general formula for it?
Is there a github or some other place the data used for this exercise and the code?
good insights, really helpful.
How to set the spark.sql.shuffle.partition by a variable instead of a constant..means if the shuffle input data size is less then it should automatically choose less number of SQL shuffle partition if input shuffle data stage is more then the job should programmaticaly be able to determine correct partition..rather then given a constant valuem
He mentioned that the slide deck would be available. Does anyone know where to find it?
what command do we use to use all 96 cores while writing instead of only 10?
Could you share the slides of this topic?
Why are 540 partitions not good? It is explained at 24:42 but I didn't quite get it.
OK, got it now: Each core processes one 100 MB partition. We have 96 cores that need to process a total of 54 GB. At a given time or batch all 96 cores can process a maximum of 96x100=9600MB. That means after 5 batches the cluster processes 9600MBx5=48000MB. For the last batch the cluster needs to process the remaining 6000MB or 6000MB/100=60 partitions. Those 60 partitions will be processed by 60 processors which is 62,5% of the cluster. The remaining 36 processors which is 37,5% of the cluster will be idle in the last batch. The story looks different if we had 480 112,5 MB partitions. This gives us 10800 MB per batch. And all data are processed after 5 batches with 100% CPU utilization.
@@michailanastasopoulos1084 Thank you so much. A much needed explanation!
@@michailanastasopoulos1084 thanks, i was with the same doubt
@@michailanastasopoulos1084 can you please help me understand why the partition size is decided to be 100MB?
@@bikashpatra119 my guess is that just because it is less than the notional upper bound of 150-200 MB/partition.
There doesn't seem to be a formula that will return the amount of spill, given a number of inputs (i.e. shuffle input size and list of tasks on this stage) but since the general logic is "smaller target partition sizes result in a reduction of spill, even if we cannot predict how much spill will result, let's just fastrack the spill reduction process by just halving the notional upper bound for the target partition size of 200 MB/partition".
Lazy loading it is just a matter of adding a filter?
I will obviously come back for reference
This video is Gold stuff
@1:00:28 min line 22 i.e. save(histPerfPath_1y) does it work if I run it on cluster to save on hdfs path ? in foreach if i save and run it in yarn-cluster mode it would fail to save with error... hdfs temp file wont find to save.... how to solve this kind of issue ?
@45 min , why broadcast has 4 times 12 =48 ? it should be 3 times 12 = 36 right? as we have 3 executors ?
The memory will reduce to 36 GB once the GC kicks in,as the data is still lying in the memory of all the executors that is it is 48 GB.
Awesome awesome awesome
it is very helpful ! Can some one share the Ebook
what do mean by saying ... have array of table names and parallelize it ... wht you mean parallelize here ?
He meant kinda ideas of “multiple threads” for spark jobs for list of tables
Could you share slide?
It's almost 2024 and the default 200 shuffle partitions are still wreaking havoc on pipelines.
watched it
In my spark version 2.4.3 job after all my transformations,computations and joins I am writing my final dataframe to s3 in parquet format
But irrespective of my cores count my job is taking fixed amount for completing save action
For distinct cores count-8,16,24 my write action timing is fixed to 8 minutes
Due to this my solution is not becoming scalable
How should I make my solution scalable so that my overall job execution time becomes proportional to cores used
@22 min where did you get Stage 21 shuffle input size ?
God Level
I was trying to load 600 million rows to a pandas dataframe from SQL Server. It was taking too long and then OOM error. I tought pyspark will solve that problem. But after 42 minutes watching this video, I see that's better using a cluster with more RAM and raise SQL Server processor. Most of requirements to setup pyspark are not known on my environment so pyspark is useless when you are working with data you don't know details about.
great one
Can anyone tell me why he reduced target size to 100 mB from 200mb
to equally utilize all the cores
are all pandas UDF vectorized?
I can never do all of it
Checking if you can 😅
is the ebook available?