Spark为什么数据在UserMemory中放不下也能跑成功?

import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

object TestSparkExecutionPlan {

    def main(args: Array[String]): Unit = {

        val sparkConf = new SparkConf()
            .setMaster("local")
            .setAppName("Test")
        val sparkContext = new SparkContext(sparkConf)

        val rdd = sparkContext.parallelize(Seq("1", "2", "3"))
        rdd.repartition(1).map {
            case item => {
                item -> item // Breakpoint here
            }
        }.aggregateByKey(mutable.Set[String]())(
            (result: mutable.Set[String], item: String) => {
                result += item // Breakpoint here
            },
            (result1: mutable.Set[String], result2: mutable.Set[String]) => {
                result1 ++= result2
            }
        ).foreach(println(_))

    }

}

/**
  * Insert the given iterator of keys and values into the map.
  *
  * When the underlying map needs to grow, check if the global pool of shuffle memory has
  * enough room for this to happen. If so, allocate the memory required to grow the map;
  * otherwise, spill the in-memory map to disk.
  *
  * The shuffle memory usage of the first trackMemoryThreshold entries is not tracked.
  */
 def insertAll(entries: Iterator[Product2[K, V]]): Unit = {

   // >>>>>>>>>>>>>>>>>>>
   log.debug("-------------- calling insertAll")
   // <<<<<<<<<<<<<<<<<<<

   if (currentMap == null) {
     throw new IllegalStateException(
       "Cannot insert new elements into a map after calling iterator")
   }
   // An update function for the map that we reuse across entries to avoid allocating
   // a new closure each time
   var curEntry: Product2[K, V] = null
   val update: (Boolean, C) => C = (hadVal, oldVal) => {
     if (hadVal) mergeValue(oldVal, curEntry._2) else createCombiner(curEntry._2)
   }

   while (entries.hasNext) {
     curEntry = entries.next()
     val estimatedSize = currentMap.estimateSize()
     if (estimatedSize > _peakMemoryUsedBytes) {
       _peakMemoryUsedBytes = estimatedSize
     }
     if (maybeSpill(currentMap, estimatedSize)) {
       currentMap = new SizeTrackingAppendOnlyMap[K, C]
     }
     currentMap.changeValue(curEntry._1, update)
     addElementsRead()
   }
 }

/**
 * Spills the current in-memory collection to disk if needed. Attempts to acquire more
 * memory before spilling.
 *
 * @param collection collection to spill to disk
 * @param currentMemory estimated size of the collection in bytes
 * @return true if `collection` was spilled to disk; false otherwise
 */
protected def maybeSpill(collection: C, currentMemory: Long): Boolean = {

  // >>>>>>>>>>>>>>>>>>>
  log.debug("---------- calling maybeSpill, currentMemory: " + currentMemory)
  log.debug("---------- _elementRead: " + _elementsRead)
  log.debug("---------- numElementsForceSpillThreadshold: " + numElementsForceSpillThreshold)
  log.debug("---------- elementReads: " + elementsRead)
  log.debug("---------- myMemoryThreshold: " + myMemoryThreshold)
  // <<<<<<<<<<<<<<<<<<<

  var shouldSpill = false
  if (elementsRead % 32 == 0 && currentMemory >= myMemoryThreshold) {
    // Claim up to double our current memory from the shuffle memory pool
    val amountToRequest = 2 * currentMemory - myMemoryThreshold
    val granted = acquireMemory(amountToRequest)
    myMemoryThreshold += granted
    // If we were granted too little memory to grow further (either tryToAcquire returned 0,
    // or we already had more memory than myMemoryThreshold), spill the current collection
    shouldSpill = currentMemory >= myMemoryThreshold
  }
  shouldSpill = shouldSpill || _elementsRead > numElementsForceSpillThreshold

  // >>>>>>>>>>>>>>>>>>>
  log.debug("--------- shouldSpill: " + shouldSpill)
  // <<<<<<<<<<<<<<<<<<<

  // Actually spill
  if (shouldSpill) {
    _spillCount += 1
    logSpillage(currentMemory)
    spill(collection)
    _elementsRead = 0
    _memoryBytesSpilled += currentMemory
    releaseMemory()
  }
  shouldSpill
}