Spark: disk I/O on stage boundaries explanation

by

It’s a good question in that we hear of in-memory Spark vs. Hadoop, so a little confusing. The docs are terrible, but I ran a few things and verified observations by looking around to find a most excellent source: http://hydronitrogen.com/apache-spark-shuffles-explained-in-depth.html

Assuming an Action has been called – so as to avoid the obvious comment if this is not stated, assuming we are not talking about ResultStage and a broadcast join, then we are talking about ShuffleMapStage. We look at an RDD initially.

Then, borrowing from the url:

  • DAG dependency involving a shuffle means creation of a separate Stage.
  • Map operations are followed by Reduce operations and a Map and so forth.

CURRENT STAGE

  • All the (fused) Map operations are performed intra-Stage.
  • The next Stage requirement, a Reduce operation – e.g. a reduceByKey, means the output is hashed or sorted by key (K) at end of the Map
    operations of current Stage.
  • This grouped data is written to disk on the Worker where the Executor is – or storage tied to that Cloud version. (I would have
    thought in memory was possible, if data is small, but this is an architectural Spark
    approach as stated from the docs.)
  • The ShuffleManager is notified that hashed, mapped data is available for consumption by the next Stage. ShuffleManager keeps track of all
    keys/locations once all of the map side work is done.

NEXT STAGE

  • The next Stage, being a reduce, then gets the data from those locations by consulting the Shuffle Manager and using Block Manager.
  • The Executor may be re-used or be a new on another Worker, or another Executor on same Worker.

So, my understanding is that architecturally, Stages mean writing to disk, even if enough memory. Given finite resources of a Worker it makes sense that writing to disk occurs for this type of operation. The more important point is, of course, the ‘Map Reduce’ implementation. I summarized the excellent posting, that is your canonical source.

Of course, fault tolerance is aided by this persistence, less re-computation work.

Similar aspects apply to DFs.

More Related Contents:

  1. Using a column value as a parameter to a spark DataFrame function
  2. Split Spark Dataframe string column into multiple columns
  3. DataFrame join optimization – Broadcast Hash Join
  4. Overwrite specific partitions in spark dataframe write method
  5. How to control partition size in Spark SQL
  6. Avoid performance impact of a single partition mode in Spark window functions
  7. How to import multiple csv files in a single load?
  8. Convert date from String to Date format in Dataframes
  9. What is the meaning of partitionColumn, lowerBound, upperBound, numPartitions parameters?
  10. Spark DataFrame: count distinct values of every column
  11. How to split a list to multiple columns in Pyspark?
  12. pyspark dataframe filter or include based on list
  13. Use collect_list and collect_set in Spark SQL
  14. What is the maximum size for a broadcast object in Spark?
  15. Why does format(“kafka”) fail with “Failed to find data source: kafka.” (even with uber-jar)?
  16. How to find mean of grouped Vector columns in Spark SQL?
  17. Generate a Spark StructType / Schema from a case class
  18. How to loop through each row of dataFrame in pyspark
  19. How does createOrReplaceTempView work in Spark?
  20. Fill in null with previously known good value with pyspark
  21. How to group by common element in array?
  22. PySpark – get row number for each row in a group
  23. Keep only duplicates from a DataFrame regarding some field
  24. Using windowing functions in Spark
  25. PySpark: How to fillna values in dataframe for specific columns?
  26. Why spark application fail with “executor.CoarseGrainedExecutorBackend: Driver Disassociated”?
  27. What does “Correlated scalar subqueries must be Aggregated” mean?
  28. How to calculate Median in spark sqlContext for column of data type double
  29. spark.ml StringIndexer throws ‘Unseen label’ on fit()
  30. PySpark error: AttributeError: ‘NoneType’ object has no attribute ‘_jvm’

Leave a Comment