Coalesce reduces parallelism of entire stage (spark)

by

Actually it is not because of SparkSQL’s optimization, SparkSQL doesn’t change the position of Coalesce operator, as the executed plan shows:

Coalesce 1
+- *Project [value#2, UDF(value#2) AS udfResult#11]
   +- *SerializeFromObject [input[0, double, false] AS value#2]
      +- Scan ExternalRDDScan[obj#1]

I quote a paragraph from coalesce API’s description:

Note: This paragraph is added by the jira SPARK-19399. So it should not be found in 2.0’s API.

However, if you’re doing a drastic coalesce, e.g. to numPartitions =
1, this may result in your computation taking place on fewer nodes
than you like (e.g. one node in the case of numPartitions = 1). To
avoid this, you can call repartition. This will add a shuffle step,
but means the current upstream partitions will be executed in parallel
(per whatever the current partitioning is).

The coalesce API doesn’t perform a shuffle, but results in a narrow dependency between previous RDD and current RDD. As RDD is lazy evaluation, the computation is actually done with coalesced partitions.

To prevent it, you should use repartition API.

More Related Contents:

  1. Encoder error while trying to map dataframe row to updated row
  2. How to define schema for custom type in Spark SQL?
  3. Specifying the filename when saving a DataFrame as a CSV [duplicate]
  4. How can I change column types in Spark SQL’s DataFrame?
  5. NullPointerException in Scala Spark, appears to be caused be collection type?
  6. how to make saveAsTextFile NOT split output into multiple file?
  7. What does setMaster `local[*]` mean in spark?
  8. Spark unionAll multiple dataframes
  9. How can I pass extra parameters to UDFs in Spark SQL?
  10. How to split a dataframe into dataframes with same column values?
  11. Spark performance for Scala vs Python
  12. How to use regex to include/exclude some input files in sc.textFile?
  13. How to define a custom aggregation function to sum a column of Vectors?
  14. How to write spark streaming DF to Kafka topic
  15. Stackoverflow due to long RDD Lineage
  16. Better way to convert a string field into timestamp in Spark
  17. Apache Spark how to append new column from list/array to Spark dataframe
  18. How to find spark RDD/Dataframe size?
  19. Provide schema while reading csv file as a dataframe in Scala Spark
  20. Apache Spark, add an “CASE WHEN … ELSE …” calculated column to an existing DataFrame
  21. Replace null values in Spark DataFrame
  22. Calculate Cosine Similarity Spark Dataframe
  23. How can I connect to a postgreSQL database into Apache Spark using scala?
  24. Spark Dataframe Random UUID changes after every transformation/action
  25. Spark column string replace when present in other column (row)
  26. How to vectorize DataFrame columns for ML algorithms?
  27. How to set hadoop configuration values from pyspark
  28. Spark DataFrames when udf functions do not accept large enough input variables
  29. Filter spark DataFrame on string contains
  30. Spark textFile vs wholeTextFiles

Leave a Comment