Apache Spark Core-Deep Dive-Proper Optimization Daniel Tomes Databricks
ฝัง
- เผยแพร่เมื่อ 5 มิ.ย. 2024
- Optimizing spark jobs through a true understanding of spark core. Learn: What is a partition? What is the difference between read/shuffle/write partitions? How to increase parallelism and decrease output files? Where does shuffle data go between stages? What is the "right" size for your spark partitions and files? Why does a job slow down with only a few tasks left and never finish? Why doesn't adding nodes decrease my compute time?
About: Databricks provides a unified data analytics platform, powered by Apache Spark™, that accelerates innovation by unifying data science, engineering and business.
Read more here: databricks.com/product/unifie...
Connect with us:
Website: databricks.com
Facebook: / databricksinc
Twitter: / databricks
LinkedIn: / databricks
Instagram: / databricksinc Databricks is proud to announce that Gartner has named us a Leader in both the 2021 Magic Quadrant for Cloud Database Management Systems and the 2021 Magic Quadrant for Data Science and Machine Learning Platforms. Download the reports here. databricks.com/databricks-nam... - วิทยาศาสตร์และเทคโนโลยี
This is so far the most informative and in-depth talk about spark job optimization I found on TH-cam. 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.
Thanks, your video helps me keep my job
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?
Excellent explanations.
Cleared so many wrong concepts of mine.
Thanks man!!!!
The best talk about spark optimizations in YT by far, thanks man!
thank you so much for explaining slat addition clearly.
Super talk Daniel and great insights, still waiting for the ebook though :)
Really very good session Dan.
It’s my pleasure to work with you.
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?
good insights, really helpful.
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...
Excellent talk Daniel 👍 I wish I saw it when I started with Spark :-) How's it looking with mentioned e-book?
4:45 1 Task = 1 Core can be changed using the property spark.task.cpu. The default is 1 Task = 1 Core.
Thanks, Daniel, great talk. Please share the ebook link.
great one
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.
How did you come up with the 16mb maxPartitionBytes? is there a general formula for it?
Great talk! Many Thanks.
Btw, where is the book, Daniel? :)
He mentioned that the slide deck would be available. Does anyone know where to find it?
This video is Gold stuff
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. :)
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
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.
@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 ?
Is the e-book available?
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
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.
what command do we use to use all 96 cores while writing instead of only 10?
Is there a github or some other place the data used for this exercise and the code?
is the ebook available?
@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.
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?
Awesome awesome awesome
Could you share the slides of this topic?
what's the link for range join optimization reference?
Lazy loading it is just a matter of adding a filter?
God Level
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
it is very helpful ! Can some one share the Ebook
Can anyone tell me why he reduced target size to 100 mB from 200mb
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 ?
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.
Slides - www2.slideshare.net/databricks/apache-spark-coredeep-diveproper-optimization?qid=68cf4b27-c5a2-48a1-96a1-faa3b411ebb4&v=&b=&from_search=1
Thank you!
Could you share slide?
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".
It's almost 2024 and the default 200 shuffle partitions are still wreaking havoc on pipelines.
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.
watched it
I can never do all of it
Checking if you can 😅
are all pandas UDF vectorized?