How does coalesce works internally

[subham0611]

Hi Databricks team,

I am trying to understand internals of spark coalesce code(DefaultPartitionCoalescer) and going through spark code for this. While I understood coalesce function but I am not sure about complete flow of code like where its get called and how coalescedRDD gets passed to executor. If you can provide a sample flow it would be great.

def coalesce(maxPartitions: Int, prev: RDD[_]): Array[PartitionGroup] = {
  val partitionLocs = new PartitionLocations(prev)
  // setup the groups (bins)
  setupGroups(math.min(prev.partitions.length, maxPartitions), partitionLocs)
  // assign partitions (balls) to each group (bins)
  throwBalls(maxPartitions, prev, balanceSlack, partitionLocs)
  getPartitions
}

I wanted to understand the code flow. Which service internally calls this function and how coalesced partitions get distributed acorss executors etc.

[raphaelblg · Accepted solution]

Hello @subham0611 ,

The coalesce operation triggered from user code can be initiated from either an RDD or a Dataset, with each having distinct codepaths:

• RDD:
  https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/rdd/RDD.scala
• Dataset: https://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/Dataset.sca...

Both the RDD and Dataset classes contain a coalesce function.

The coalescing logic is relatively straightforward:

The driver node determines the Spark plan for the coalesce operation. When using the Dataset API, this operation results in a narrow dependency. For instance, if you reduce the number of partitions from 1000 to 100, there will not be a shuffle. Instead, each of the 100 new partitions will claim 10 of the current partitions.