最近專案不是很忙，把一些東西整理出來，當作筆記，主要是為了養成一個好的習慣。 這個主要介紹MLlib原始碼主要出現的treeAggregrate

package com.lm.spark.ml

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

object Treeaggreate {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("test").setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")
    val  rdd = sc.parallelize(1 to 12).repartition(6)

//    rdd.mapPartitionsWithIndex((index:Int,it:Iterator[Int])=>{
//      Array((s" $index : ${it.toList.mkString(",")}")).toIterator
//    }).foreach(println)

    rdd.mapPartitionsWithIndex((index:Int,it:Iterator[Int])=>{
      Array((s" $index : ${it.toList.mkString(",")}")).toIterator
    }).foreach(println)

    val res1 = rdd.aggregate(0)(seq,opt)

    println("----------")

    val res2 = rdd.treeAggregate(0)(seq,opt)

    sc.stop()
  }

  def seq(s1:Int,s2:Int):Int = {
    println("seq  "+s1+":"+s2)
    s1+s2
  }


  def opt(s1:Int,s2:Int):Int = {
    println("opt  "+s1+":"+s2)
    s1+s2
  }

}

 

原始碼結構：