How are stages split into tasks in Spark?

by

You have a pretty nice outline here. To answer your questions

  • A separate task does need to be launched for each partition of data for each stage. Consider that each partition will likely reside on distinct physical locations – e.g. blocks in HDFS or directories/volumes for a local file system.

Note that the submission of Stages is driven by the DAG Scheduler. This means that stages that are not interdependent may be submitted to the cluster for execution in parallel: this maximizes the parallelization capability on the cluster. So if operations in our dataflow can happen simultaneously we will expect to see multiple stages launched.

We can see that in action in the following toy example in which we do the following types of operations:

  • load two datasources
  • perform some map operation on both of the data sources separately
  • join them
  • perform some map and filter operations on the result
  • save the result

So then how many stages will we end up with?

  • 1 stage each for loading the two datasources in parallel = 2 stages
  • A third stage representing the join that is dependent on the other two stages
  • Note: all of the follow-on operations working on the joined data may be performed in the same stage because they must happen sequentially. There is no benefit to launching additional stages because they can not start work until the prior operation were completed.

Here is that toy program

val sfi  = sc.textFile("/data/blah/input").map{ x => val xi = x.toInt; (xi,xi*xi) }
val sp = sc.parallelize{ (0 until 1000).map{ x => (x,x * x+1) }}
val spj = sfi.join(sp)
val sm = spj.mapPartitions{ iter => iter.map{ case (k,(v1,v2)) => (k, v1+v2) }}
val sf = sm.filter{ case (k,v) => v % 10 == 0 }
sf.saveAsTextFile("/data/blah/out")

And here is the DAG of the result

enter image description here

Now: how many tasks ? The number of tasks should be equal to

Sum of (Stage * #Partitions in the stage)

More Related Contents:

  1. Find maximum row per group in Spark DataFrame
  2. While writing to hdfs path getting error java.io.IOException: Failed to rename
  3. Multiple Aggregate operations on the same column of a spark dataframe
  4. How to read multiple text files into a single RDD?
  5. How to handle categorical features with spark-ml?
  6. How to access element of a VectorUDT column in a Spark DataFrame?
  7. What is the difference between map and flatMap and a good use case for each?
  8. Which operations preserve RDD order?
  9. Python Spark Cumulative Sum by Group Using DataFrame
  10. Link Spark with iPython Notebook
  11. Default Partitioning Scheme in Spark
  12. What is the maximum size for a broadcast object in Spark?
  13. Any performance issues forcing eager evaluation using count in spark?
  14. How does createOrReplaceTempView work in Spark?
  15. Debugging “Managed memory leak detected” in Spark 1.6.0
  16. Spark: how to get the number of written rows?
  17. Why are Spark Parquet files for an aggregate larger than the original?
  18. Save ML model for future usage
  19. Determining optimal number of Spark partitions based on workers, cores and DataFrame size
  20. Apache spark dealing with case statements
  21. pyspark: count distinct over a window
  22. Mind blown: RDD.zip() method
  23. How spark read a large file (petabyte) when file can not be fit in spark’s main memory
  24. Serialize a custom transformer using python to be used within a Pyspark ML pipeline
  25. Not able to cat dbfs file in databricks community edition cluster. FileNotFoundError: [Errno 2] No such file or directory:
  26. PySpark: How to fillna values in dataframe for specific columns?
  27. How to optimize shuffle spill in Apache Spark application
  28. How to calculate Median in spark sqlContext for column of data type double
  29. PySpark error: AttributeError: ‘NoneType’ object has no attribute ‘_jvm’
  30. Understanding Spark’s caching

Leave a Comment