spark-学习笔记--29 shuffle2

shuffle2

 

spark shuffle 演进的历史：

 

Spark 0.8及以前 Hash Based Shuffle

Spark 0.8.1 为Hash Based Shuffle引入File Consolidation 机制

Spark 0.9 引入ExternalAppendOnlyMap

Spark 1.1 引入Sort Based Shuffle，但默认仍为Hash Based Shuffle

Spark 1.2 默认的Shuffle方式改为Sort Based Shuffle

Spark 1.4 引入Tungsten-Sort Based Shuffle

Spark 1.6 Tungsten-sort并入Sort Based Shuffle

Spark 2.0 Hash Based Shuffle退出历史舞台

 

 

File Consolidation   同一个节点：

并行执行的 shuffleMapTask  写不同的文件

并行执行后再运行的shuffleMapTask  复用文件  写到不同的 segment

 

 

 

 

 

org.apache.spark.shuffle.hash.HashShuffleWriter 

 

    判断是否需要本地聚合

若需要 则执行本地聚合 ： (hello,1)、(hello,1)  会聚合成  (hello,2)

    

 

    使用partitioner 进行分区  默认使用 hashPartitioner  

生成 bucketId

相同 bucketId 的数据写入到一个bucket

 

 

  /** Write a bunch of records to this task's output */

  override def write(records: Iterator[Product2[K, V]]): Unit = {

    val iter = if (dep.aggregator.isDefined) {

      if (dep.mapSideCombine) {

        dep.aggregator.get.combineValuesByKey(records, context)

      } else {

        records

      }

    } else {

      require(!dep.mapSideCombine, "Map-side combine without Aggregator specified!")

      records

    }

 

    for (elem <- iter) {

      val bucketId = dep.partitioner.getPartition(elem._1)

      shuffle.writers(bucketId).write(elem._1, elem._2)

    }

 

  }