Downloaded 1,297 times
 => U)) 3. Fn to produce results 14](https://image.slidesharecdn.com/stratadebuggingtalk-150221160850-conversion-gate02/75/Tuning-and-Debugging-in-Apache-Spark-14-2048.jpg)
 => U)) 3. Fn to produce results 15](https://image.slidesharecdn.com/stratadebuggingtalk-150221160850-conversion-gate02/75/Tuning-and-Debugging-in-Apache-Spark-15-2048.jpg)
 => U)) 3. Fn to produce results 16](https://image.slidesharecdn.com/stratadebuggingtalk-150221160850-conversion-gate02/75/Tuning-and-Debugging-in-Apache-Spark-16-2048.jpg)
 => U)) 3. Fn to produce results 17](https://image.slidesharecdn.com/stratadebuggingtalk-150221160850-conversion-gate02/75/Tuning-and-Debugging-in-Apache-Spark-17-2048.jpg)
![Seeing this on your own scala> counts.toDebugString res84: String = (2) ShuffledRDD[296] at reduceByKey at <console>:17 +-(3) MappedRDD[295] at map at <console>:17 | FilteredRDD[294] at filter at <console>:15 | MappedRDD[293] at map at <console>:15 | input.text MappedRDD[292] at textFile at <console>:13 | input.text HadoopRDD[291] at textFile at <console>:13 25 (indentations indicate a shuffle boundary)](https://image.slidesharecdn.com/stratadebuggingtalk-150221160850-conversion-gate02/75/Tuning-and-Debugging-in-Apache-Spark-25-2048.jpg)
![Example: take(N) action class RDD { def take(n: Int) { val results = new ArrayBuffer[T] var partition = 0 while (results.size < n) { result ++= sc.runJob(this, partition, it => it.toArray) partition = partition + 1 } return results.take(n) } } 27](https://image.slidesharecdn.com/stratadebuggingtalk-150221160850-conversion-gate02/75/Tuning-and-Debugging-in-Apache-Spark-27-2048.jpg)
![Choice of Serializer Serialization is sometimes a bottleneck when shuffling and caching data. Using the Kryo serializer is often faster. val conf = new SparkConf() conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") // Be strict about class registration conf.set("spark.kryo.registrationRequired", "true") conf.registerKryoClasses(Array(classOf[MyClass], classOf[MyOtherClass])) 33](https://image.slidesharecdn.com/stratadebuggingtalk-150221160850-conversion-gate02/75/Tuning-and-Debugging-in-Apache-Spark-33-2048.jpg)
Uploaded byPatrick Wendell
Tuning and Debugging in Apache Spark
The document discusses tuning and debugging of Apache Spark, detailing its execution model, RDD API, and the importance of understanding Spark's internal mechanisms for optimizing performance. It emphasizes tasks, stages, jobs, and data shuffling while suggesting strategies to improve execution speed, such as using built-in operators and managing data partitioning effectively. The talk is geared towards participants familiar with Spark's core API, and mentions further resources and a conference for deeper insights.
In this document
Powered by AI
Introduction to tuning and debugging in Apache Spark, speaker background, overview of Databricks, and talk objectives.
Outline of Spark's execution model emphasizing key concepts like jobs, stages, and tasks.
Overview of RDD API, illustrating how RDD transformations work and their representation as a Directed Acyclic Graph (DAG).
Process of evaluating a DAG in Spark using the runJob method and the importance of actions.
Illustration of the runJob method's operation, including the need to compute dependencies back to parent RDDs.
Description of the stage graph, detailing the tasks involved in processing RDDs and their execution relationship.
Understanding Spark operations with RDD actions like count() and take(), and how they relate to runJob.
Discuss factors affecting Spark performance including data shuffling, parallelism, serialization, caching formats, and hardware considerations.
Advocacy for using higher-level Spark APIs, such as DataFrame and Spark SQL, for improved processing efficiency.
Information about references for further learning, upcoming event schedules, and concluding remarks.
Details on internal mechanics of RDD interface, examples including Hadoop RDD, Filtered RDD, and Joined RDD within complex DAG structures.
Explains narrow and wide transformations within RDDs, using examples of joined and filtered RDDs to illustrate performance differences.
Tuning and Debugging in Apache Spark
- 1. Tuning and Debuggingin Apache Spark Patrick Wendell @pwendell February 20, 2015
- 2. About Me Apache Sparkcommitter and PMC, release manager Worked on Spark at UC Berkeley when the project started Today, managing Spark efforts at Databricks 2
- 3. About Databricks Founded bycreators of Spark in 2013 Donated Spark to ASF and remain largest contributor End-to-End hosted service: Databricks Cloud 3
- 4. Today’s Talk Help youunderstand and debug Spark programs Related talk this afternoon: Assumes you know Spark core API concepts, focused on internals 4
- 5.
- 6. 6 The key totuning Spark apps is a sound grasp of Spark’s internal mechanisms.
- 7. Key Question How doesa user program get translated into units of physical execution: jobs, stages, and tasks: 7 ?
- 8. RDD API Refresher RDDsare a distributed collection of records rdd = spark.parallelize(range(10000), 10) Transformations create new RDDs from existing ones errors = rdd.filter(lambda line: “ERROR” in line) Actions materialize a value in the user program size = errors.count() 8
- 9. RDD API Example //Read input file val input = sc.textFile("input.txt") val tokenized = input .map(line => line.split(" ")) .filter(words => words.size > 0) // remove empty lines val counts = tokenized // frequency of log levels .map(words => (words(0), 1)). .reduceByKey{ (a, b) => a + b, 2 } 9 INFO Server started INFO Bound to port 8080 WARN Cannot find srv.conf input.txt
- 10. RDD API Example //Read input file val input = sc.textFile( ) val tokenized = input .map(line => line.split(" ")) .filter(words => words.size > 0) // remove empty lines val counts = tokenized // frequency of log levels .map(words => (words(0), 1)). .reduceByKey{ (a, b) => a + b } 10
- 11.
- 12. DAG View ofRDD’s textFile() map() filter() map() reduceByKey() 12 Mapped RDD Partition 1 Partition 2 Partition 3 Filtered RDD Partition 1 Partition 2 Partition 3 Mapped RDD Partition 1 Partition 2 Partition 3 Shuffle RDD Partition 1 Partition 2 Hadoop RDD Partition 1 Partition 2 Partition 3 input tokenized counts
- 13. Transformations build upa DAG, but don’t “do anything” 13
- 14. Evaluation of theDAG We mentioned “actions” a few slides ago. Let’s forget them for a minute. DAG’s are materialized through a method sc.runJob: def runJob[T, U]( rdd: RDD[T], 1. RDD to compute partitions: Seq[Int], 2. Which partitions func: (Iterator[T]) => U)) 3. Fn to produce results 14
- 15. Evaluation of theDAG We mentioned “actions” a few slides ago. Let’s forget them for a minute. DAG’s are materialized through a method sc.runJob: def runJob[T, U]( rdd: RDD[T], 1. RDD to compute partitions: Seq[Int], 2. Which partitions func: (Iterator[T]) => U)) 3. Fn to produce results 15
- 16. Evaluation of theDAG We mentioned “actions” a few slides ago. Let’s forget them for a minute. DAG’s are materialized through a method sc.runJob: def runJob[T, U]( rdd: RDD[T], 1. RDD to compute partitions: Seq[Int], 2. Which partitions func: (Iterator[T]) => U)) 3. Fn to produce results 16
- 17. Evaluation of theDAG We mentioned “actions” a few slides ago. Let’s forget them for a minute. DAG’s are materialized through a method sc.runJob: def runJob[T, U]( rdd: RDD[T], 1. RDD to compute partitions: Seq[Int], 2. Which partitions func: (Iterator[T]) => U)) 3. Fn to produce results 17
- 18. How runJob Works Needsto compute my parents, parents, parents, etc all the way back to an RDD with no dependencies (e.g. HadoopRDD). 18 Mapped RDD Partition 1 Partition 2 Partition 3 Filtered RDD Partition 1 Partition 2 Partition 3 Mapped RDD Partition 1 Partition 2 Partition 3 Shuffle RDD Partition 1 Partition 2 Hadoop RDD Partition 1 Partition 2 Partition 3 input tokenized counts runJob(counts)
- 19. Physical Optimizations 1. Certaintypes of transformations can be pipelined. 1. If dependent RDD’s have already been cached (or persisted in a shuffle) the graph can be truncated. Once pipelining and truncation occur, Spark produces a a set of stages each stage is composed of tasks 19
- 20. How runJob Works Needsto compute my parents, parents, parents, etc all the way back to an RDD with no dependencies (e.g. HadoopRDD). 20 Mapped RDD Partition 1 Partition 2 Partition 3 Filtered RDD Partition 1 Partition 2 Partition 3 Mapped RDD Partition 1 Partition 2 Partition 3 Shuffle RDD Partition 1 Partition 2 Hadoop RDD Partition 1 Partition 2 Partition 3 input tokenized counts runJob(counts)
- 21. How runJob Works Needsto compute my parents, parents, parents, etc all the way back to an RDD with no dependencies (e.g. HadoopRDD). 21 input tokenized counts Mapped RDD Partition 1 Partition 2 Partition 3 Filtered RDD Partition 1 Partition 2 Partition 3 Mapped RDD Partition 1 Partition 2 Partition 3 Shuffle RDD Partition 1 Partition 2 Hadoop RDD Partition 1 Partition 2 Partition 3 runJob(counts)
- 22. How runJob Works Needsto compute my parents, parents, parents, etc all the way back to an RDD with no dependencies (e.g. HadoopRDD). 22 input tokenized counts Mapped RDD Partition 1 Partition 2 Partition 3 Filtered RDD Partition 1 Partition 2 Partition 3 Mapped RDD Partition 1 Partition 2 Partition 3 Shuffle RDD Partition 1 Partition 2 Hadoop RDD Partition 1 Partition 2 Partition 3 runJob(counts)
- 23. Stage Graph 23 Task 1 Task2 Task 3 Task 1 Task 2 Stage 1 Stage 2 Each task will: 1. Read Hadoop input 2. Perform maps and filters 3. Write partial sums Each task will: 1. Read partial sums 2. Invoke user function passed to runJob.Shuffle write Shuffle readInput
- 24. Units of PhysicalExecution Jobs: Work required to compute RDD in runJob. Stages: A wave of work within a job, corresponding to one or more pipelined RDD’s. Tasks: A unit of work within a stage, corresponding to one RDD partition. Shuffle: The transfer of data between stages. 24
- 25. Seeing this onyour own scala> counts.toDebugString res84: String = (2) ShuffledRDD[296] at reduceByKey at <console>:17 +-(3) MappedRDD[295] at map at <console>:17 | FilteredRDD[294] at filter at <console>:15 | MappedRDD[293] at map at <console>:15 | input.text MappedRDD[292] at textFile at <console>:13 | input.text HadoopRDD[291] at textFile at <console>:13 25 (indentations indicate a shuffle boundary)
- 26. Example: count() action classRDD { def count(): Long = { results = sc.runJob( this, 1. RDD = self 0 until partitions.size, 2. Partitions = all partitions it => it.size() 3. Function = size of the partition ) return results.sum } 26
- 27. Example: take(N) action classRDD { def take(n: Int) { val results = new ArrayBuffer[T] var partition = 0 while (results.size < n) { result ++= sc.runJob(this, partition, it => it.toArray) partition = partition + 1 } return results.take(n) } } 27
- 28. Putting it AllTogether 28 Named after action calling runJob Named after last RDD in pipeline
- 29. 29 Determinants of Performancein Spark
- 30. Quantity of DataShuffled In general, avoiding shuffle will make your program run faster. 1. Use the built in aggregateByKey() operator instead of writing your own aggregations. 2. Filter input earlier in the program rather than later. 3. Go to this afternoon’s talk! 30
- 31. Degree of Parallelism >input = sc.textFile("s3n://log-files/2014/*.log.gz") #matches thousands of files > input.getNumPartitions() 35154 > lines = input.filter(lambda line: line.startswith("2014-10-17 08:")) # selective > lines.getNumPartitions() 35154 > lines = lines.coalesce(5).cache() # We coalesce the lines RDD before caching > lines.getNumPartitions() 5 >>> lines.count() # occurs on coalesced RDD 31
- 32. Degree of Parallelism Ifyou have a huge number of mostly idle tasks (e.g. 10’s of thousands), then it’s often good to coalesce. If you are not using all slots in your cluster, repartition can increase parallelism. 32
- 33. Choice of Serializer Serializationis sometimes a bottleneck when shuffling and caching data. Using the Kryo serializer is often faster. val conf = new SparkConf() conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") // Be strict about class registration conf.set("spark.kryo.registrationRequired", "true") conf.registerKryoClasses(Array(classOf[MyClass], classOf[MyOtherClass])) 33
- 34. Cache Format By defaultSpark will cache() data using MEMORY_ONLY level, deserialized JVM objects MEMORY_ONLY_SER can help cut down on GC MEMORY_AND_DISK can avoid expensive recompuations 34
- 35. Hardware Spark scales horizontally,so more is better Disk/Memory/Network balance depends on workload: CPU intensive ML jobs vs IO intensive ETL jobs Good to keep executor heap size to 64GB or less (can run multiple on each node) 35
- 36. Other Performance Tweaks Switchingto LZF compression can improve shuffle performance (sacrifices some robustness for massive shuffles): conf.set(“spark.io.compression.codec”, “lzf”) Turn on speculative execution to help prevent stragglers conf.set(“spark.speculation”, “true”) 36
- 37. Other Performance Tweaks Makesure to give Spark as many disks as possible to allow striping shuffle output SPARK_LOCAL_DIRS in Mesos/Standalone In YARN mode, inherits YARN’s local directories 37
- 38. 38 One Weird Trickfor Great Performance
- 39. Use Higher LevelAPI’s! DataFrame APIs for core processing Works across Scala, Java, Python and R Spark ML for machine learning Spark SQL for structured query processing 39
- 40. 40 See also Chapter 8:Tuning and Debugging Spark.
- 41. Come to SparkSummit 2015! 41 June 15-17 in San Francisco
- 42. Other Spark HappeningsToday Spark team “Ask Us Anything” at 2:20 in 211 B Tips for writing better Spark programs at 4:00 in 230C I’ll be around Databricks booth after this 42
- 43.
- 44.
- 45. Internals of theRDD Interface 45 1) List of partitions 2) Set of dependencies on parent RDDs 3) Function to compute a partition, given parents 4) Optional partitioning info for k/v RDDs (Partitioner) RDD Partition 1 Partition 2 Partition 3
- 46. Example: Hadoop RDD 46 Partitions= 1 per HDFS block Dependencies = None compute(partition) = read corresponding HDFS block Partitioner = None > rdd = spark.hadoopFile(“hdfs://click_logs/”)
- 47. Example: Filtered RDD 47 Partitions= parent partitions Dependencies = a single parent compute(partition) = call parent.compute(partition) and filter Partitioner = parent partitioner > filtered = rdd.filter(lambda x: x contains “ERROR”)
- 48. Example: Joined RDD 48 Partitions= number chosen by user or heuristics Dependencies = ShuffleDependency on two or more parents compute(partition) = read and join data from all parents Partitioner = HashPartitioner(# partitions)
- 49. 49 A More ComplexDAG Joined RDD Partition 1 Partition 2 Partition 3 Filtered RDDPartition 1 Partition 2 Mapped RDDPartition 1 Partition 2 Hadoop RDDPartition 1 Partition 2 JDBC RDD Partition 1 Partition 2 Filtered RDDPartition 1 Partition 2 Partition 3 .count()
- 50. 50 A More ComplexDAG Stage 3 Task 1 Task 2 Task 3 Stage 2 Task 1 Task 2 Stage 1 Task 1 Task 2 Shuffle Read Shuffle Write
- 51. 51 RDD Partition 1 Partition 2 Partition 3 Parent Partition 1 Partition 2 Partition 3 Narrow and WideTransformations RDD Partition 1 Partition 2 Partition 3 Parent 1 Partition 1 Partition 2 Parent 2 Partition 1 Partition 2 FilteredRDD JoinedRDD