Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Advanced Spark @ Spark Summit 2014

Advanced Spark @ Spark Summit 2014

Reynold Xin

July 02, 2014
Tweet

More Decks by Reynold Xin

Other Decks in Technology

Transcript

  1. This Talk Formalize RDD concept Life of a Spark Application

    Performance Debugging * Assumes you can write word count, knows what transformation/action is “Mechanical sympathy” by Jackie Stewart: a driver does not need to know how to build an engine but they need to know the fundamentals of how one works to get the best out of it
  2. Reynold Xin Apache Spark committer (worked on almost every module:

    core, sql, mllib, graph) Product & open-source eng @ Databricks On leave from PhD @ UC Berkeley AMPLab
  3. Example Application val sc = new SparkContext(...) val file =

    sc.textFile(“hdfs://...”) val errors = file.filter(_.contains(“ERROR”)) errors.cache() errors.count() Resilient distributed datasets (RDDs) Action
  4. Quiz: what is an “RDD”? A: distributed collection of objects

    on disk B: distributed collection of objects in memory C: distributed collection of objects in Cassandra Answer: could be any of the above!
  5. Scientific Answer: RDD is an Interface! 1.  Set of partitions

    (“splits” in Hadoop) 2.  List of dependencies on parent RDDs 3.  Function to compute a partition" (as an Iterator) given its parent(s) 4.  (Optional) partitioner (hash, range) 5.  (Optional) preferred location(s)" for each partition “lineage” optimized execution
  6. Example: HadoopRDD partitions = one per HDFS block dependencies =

    none compute(part) = read corresponding block preferredLocations(part) = HDFS block location partitioner = none
  7. Example: Filtered RDD partitions = same as parent RDD dependencies

    = “one-to-one” on parent compute(part) = compute parent and filter it preferredLocations(part) = none (ask parent) partitioner = none
  8. RDD Graph (DAG of tasks) HadoopRDD" path = hdfs://... FilteredRDD"

    func = _.contains(…)" shouldCache = true file: errors: Partition-level view: Dataset-level view: Task1 Task2 ...
  9. Example: JoinedRDD partitions = one per reduce task dependencies =

    “shuffle” on each parent compute(partition) = read and join shuffled data preferredLocations(part) = none" partitioner = HashPartitioner(numTasks) Spark will now know this data is hashed!
  10. Dependency Types union groupByKey on" non-partitioned data join with inputs

    not" co-partitioned join with inputs co-partitioned map, filter “Narrow” (pipeline-able) “Wide” (shuffle)
  11. Recap Each RDD consists of 5 properties: 1.  partitions 2. 

    dependencies 3.  compute 4.  (optional) partitioner 5.  (optional) preferred locations
  12. Spark Application sc = new SparkContext f = sc.textFile(“…”)" "

    f.filter(…)" .count()" " ... Your program (JVM / Python) Spark driver" (app master) Spark executor (multiple of them) HDFS, HBase, … Block manager Task threads RDD graph Scheduler Block tracker Shuffle tracker Cluster" manager A single application often contains multiple actions
  13. Job Scheduling Process rdd1.join(rdd2) .groupBy(…) .filter(…) .count() RDD  Objects  

    build  operator  DAG   Scheduler   (DAGScheduler)   split  graph  into   stages  of  tasks   submit  each   stage  as  ready   DAG   Executors   execute  tasks   store  and  serve   blocks   Block manager Threads Task  
  14. DAG Scheduler Input: RDD and partitions to compute Output: output

    from actions on those partitions Roles: >  Build stages of tasks >  Submit them to lower level scheduler (e.g. YARN, Mesos, Standalone) as ready >  Lower level scheduler will schedule data based on locality >  Resubmit failed stages if outputs are lost
  15. Scheduler Optimizations Pipelines operations within a stage Picks join algorithms

    based on partitioning (minimize shuffles) Reuses previously cached data join   union   groupBy   map   Stage  3   Stage  1   Stage  2   A:   B:   C:   D:   E:   F:   G:   =  previously  computed  partition   Task  
  16. Task Unit of work to execute on in an executor

    thread Unlike MR, there is no “map” vs “reduce” task Each task either partitions its output for “shuffle”, or send the output back to the driver
  17. Shuffle Stage  1   Stage  2   Redistributes data among

    partitions Partition keys into buckets (user-defined partitioner) Optimizations: >  Avoided when possible, if" data is already properly" partitioned >  Partial aggregation reduces" data movement
  18. Shuffle Disk   Stage  2   Stage  1   Write

    intermediate files to disk Fetched by the next stage of tasks (“reduce” in MR)
  19. Recap: Job Scheduling rdd1.join(rdd2) .groupBy(…) .filter(…) .count() RDD  Objects  

    build  operator  DAG   Scheduler   (DAGScheduler)   split  graph  into   stages  of  tasks   submit  each   stage  as  ready   DAG   Executors   execute  tasks   store  and  serve   blocks   Block manager Threads Task  
  20. Performance Debugging Distributed performance: program slow due to scheduling, coordination,

    or data distribution) Local performance: program slow because whatever I’m running is just slow on a single node Two useful tools: >  Application web UI (default port 4040) >  Executor logs (spark/work)
  21. Stragglers due to slow nodes sc.parallelize(1 to 15, 15).map {

    index => val host = java.net.InetAddress.getLocalHost.getHostName if (host == "ip-172-31-2-222") { Thread.sleep(10000) } else { Thread.sleep(1000) } }.count()
  22. Stragglers due to slow nodes Turn speculation on to mitigates

    this problem. Speculation: Spark identifies slow tasks (by looking at runtime distribution), and re-launches those tasks on other nodes. spark.speculation true
  23. Stragglers due to data skew sc.parallelize(1 to 15, 15) .flatMap

    { i => 1 to i } .map { i => Thread.sleep(1000) } .count() Speculation is not going to help because the problem is inherent in the algorithm/data. Pick a different algorithm or restructure the data.
  24. What if the task is still running? To discover whether

    GC is the problem: 1.  Set spark.executor.extraJavaOptions to include: “-XX:-PrintGCDetails -XX:+PrintGCTimeStamps” 2.  Look at spark/work/app…/[n]/stdout on executors 3.  Short GC times are OK. Long ones are bad.
  25. jmap: heap analysis jmap -histo [pid] Gets a histogram of

    objects in the JVM heap jmap -histo:live [pid] Gets a histogram of objects in the heap after GC (thus “live”)
  26. Reduce GC impact class DummyObject(var i: Int) { def toInt

    = i } sc.parallelize(1 to 100 * 1000 * 1000, 1).map { i => new DummyObject(i) // new object every record obj.toInt } sc.parallelize(1 to 100 * 1000 * 1000, 1).mapPartitions { iter => val obj = new DummyObject(0) // reuse the same object iter.map { i => obj.i = i obj.toInt } }
  27. Local Performance Each Spark executor runs a JVM/Python process Insert

    your favorite JVM/Python profiling tool >  jstack >  YourKit >  VisualVM >  println >  (sorry I don’t know a whole lot about Python) >  …
  28. Example: identify expensive comp. def someCheapComputation(record: Int): Int = record

    + 1 def someExpensiveComputation(record: Int): String = { Thread.sleep(1000) record.toString } sc.parallelize(1 to 100000).map { record => val step1 = someCheapComputation(record) val step2 = someExpensiveComputation(step1) step2 }.saveAsTextFile("hdfs:/tmp1")
  29. Local Debugging Run in local mode (i.e. Spark master “local”)

    and debug with your favorite debugger >  IntelliJ >  Eclipse >  println With a sample dataset
  30. What we have learned? RDD abstraction >  lineage info: partitions,

    dependencies, compute >  optimization info: partitioner, preferred locations Execution process (from RDD to tasks) Performance & debugging