一.Spark中的運算元總結（原理） 

Spark運算元
        1).Transformations ,轉換運算元，懶執行，需要Action類運算元觸發。
            map/mapToPair,flatMap,filter,reduceByKey,sample,sortBy/sortByKey,groupByKey,join,leftOutJoin,rightOuterJoin,fullOuterJoin,distinct,union,intersection,subtract,repartition,coalesce,zip,zipWithIndex,mapPartitions,
            mapPartitionWithIndex,cogroup,mapValues,aggreagateByKey,combineByKey
        2).Action，行動運算元，觸發Action類運算元執行。Spark應用程式中(Spark Application)有一個Action運算元就有了一個job。
            take,frist,foreach,count,collect,reduce,foreachPartition,countByKey,countByValue
        3).持久化運算元。
            a).cache
                預設將資料持久化到記憶體，cache()=persist()=persist(StorageLevel.MEMORY_ONLY)
            b).persist
                可以手動指定資料持久化級別。
                MEMORY_ONLY
                MEMORY_ONLY_SER
                MEMORY_AND_DISK
                MEMORY_AND_DISK_SER
                "_2"代表有副本數，儘量避免使用"DISK_ONLY"級別。
            c).checkpoint
                將資料可以持久化到磁碟，指定的checkpoint目錄中，切斷checkpointRDD之前的依賴關係，使之後的RDD依賴於checkpoint目錄中的資料。需要設定checkpoint路徑。
                RDD lineage 非常長，每一個RDD之間邏輯複雜，計算耗時。對一個RDD進行checkpoint之前最好先cache下。
                    
            注意：
            a).cache和persist注意事項：
                i).cache和persist是懶執行，需要Action運算元觸發。
                ii).對一個RDD進行cache/persist之後，可以賦值給一個變數，下次直接使用這個變數就是使用的持久化的資料。
                iii).cache/persist之後不能緊跟Action類運算元。
            b).checkpoint執行流程：
                i).Spark任務執行完成之後，會從後往前回溯，找到CcheckpointRDD做標記。
                ii).回溯完成之後，重新計算標記RDD的資料，將資料放入checkpoint目錄中。
                iii).切斷RDD之間的依賴關係。

————————————————————————————————————————————————————————

二.運算元例項 

package com.bjsxt

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import scala.collection.mutable.ListBuffer

object Suanzi {
  def main(args: Array[String]): Unit = {
    val conf=new SparkConf().setAppName("test").setMaster("local");
    val sc=new SparkContext(conf)
            val rdd2=sc.parallelize(Array(
        "love1","love2","love3","love4",
                "love5","love6","love7","love8",
                "love9","love10","love11","love12"         
                ),3)
                rdd2.foreach(println)
                println("_________________________________--")
                /**
                 * repartition重新分割槽
                 * 增加分割槽
                 * 是一個shuffle類的運算元
                 */
                val repartition1=rdd2.repartition(5)
                repartition1.foreach(println)
                println("+++++++++++++++++++++++++++++++++++++")
                /**
                 * repartition重新分割槽
                 * 減少分割槽
                 * 是一個shuffle類的運算元
                 */
                val repartition2=rdd2.repartition(1)
                repartition2.foreach(println)
                /**
                 * Coalesce
                 * 可以增多，也可以減少分割槽
                 * 底層有shuffle，但是需要設定，預設是false關閉的
                 * 
                 */
                println("##########################################")
                val coalesce1=rdd2.coalesce(6, false)
                coalesce1.foreach(println)
                            /**
                 * Coalesce
                 * 可以增多，也可以減少分割槽
                 * 底層有shuffle，但是需要設定，預設是false關閉的
                 * 
                 */
                println("%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%")
                val coalesce2=rdd2.coalesce(6, true)
                coalesce2.foreach(println)
                            /**
                 * Coalesce
                 * 可以增多，也可以減少分割槽
                 * 底層有shuffle，但是需要設定，預設是false關閉的
                 * 
                 */
                println("PPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPPP")
                val coalesce3=rdd2.coalesce(2, true)
                coalesce3.foreach(println)    
                println("*****************************************************")
                /**
                 * mapPartitionsWithIndex
                 * 根據索引進行分割槽
                 */
                    val    mapPartitionsWithIndex = rdd2.mapPartitionsWithIndex((index,iter)=>{
                  val list=ListBuffer[String]()
                  while(iter.hasNext){
                    val next=iter.next()
                        println("parallelize index=["+index+"],value=["+next+"]")
                        list.+=(next)
                  }
                  list.iterator              
                },false).collect()
                println("^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^")
                /**
                 * reduce,將value的值進行累計求和
                 */
    val rdd1=sc.parallelize(Array(("a",200),("b",200),("a",300),("d",400),("a",200),("b",200),("a",300),("d",400)))
    val result=rdd1.reduce((t1,t2)=>{
      val key=t1._1+"~"+t2._1
      val value=t1._2+t2._2
      (key,value)
    })
    println(result)
    println("?????????????????????????????????????????????????")
    /**
     * countbyvalue
     * 根據value值不同進行求和計算
     */
    val result1=rdd1.countByValue()
    result1.foreach(println)
    /**
     * countbykey
     * 根據value值不同進行求和計算
     */ 
    println("^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^")
    val result2=rdd1.countByKey()
    result2.foreach(println)
    /**
     * groupByKey
     * 按照key的值不同進行分組
     * 將value的值放在一起
     */
    println("""""""""""""""""""""""""""""""""""""""""""""""""")
       rdd1.groupByKey().foreach(println)
       /**
        * zip的意思就是合併成（k,v)格式
        * 即第一個rdd的值為key，第二個RDD的值為value
        */
       println("zip$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$$")
        val rdd3=sc.parallelize(Array(
        "love1","love2","love3","love4",
                "love5","love6","love7","love8",
                "love9","love10","love11","love12"         
                ))
                val rdd4=sc.parallelize(Array(
        "love1","love2","love3","love4",
                "love5","love6","love7","love8",
                "love9","love10","love11","love12"         
                ))
      val zip=rdd3.zip(rdd4)
      zip.foreach(println)
    /**
     * zipWithIndex
     */
      println("parallelize##########@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@")
    val parallelize=sc.parallelize(Array(
        "love1","love2","love3","love4",
                "love5","love6","love7","love8",
                "love9","love10","love11","love12"         
                ),3)
                val zipWithIndex= parallelize.zipWithIndex()
                zipWithIndex.foreach(println)
      /**
       * mapPartitionsWithIndex和coalesce的混合使用
       */
      println("mapPartitionsWithIndex和coalesce的混合使用")
        val    mapPartitionsWithIndex1 = rdd3.mapPartitionsWithIndex((index,iter)=>{
                  val list=ListBuffer[String]()
                  while(iter.hasNext){
                    val next=iter.next()
                        println("parallelize index=["+index+"],value=["+next+"]")
                        list.+=(next)
                  }
                  list.iterator              
                },false)
                val coalesce=mapPartitionsWithIndex1.coalesce(7,true)
                val result0=coalesce.mapPartitionsWithIndex((index,iter)=>{
                  val list=ListBuffer[String]()
                  while(iter.hasNext){
                    val next=iter.next()
                        println("parallelize index=["+index+"],value=["+next+"]")
                        list.+=(next)
                  }
                  list.iterator    
                  
                },true).collect()
                result0.foreach(println)
  }
}
 

用心做事，只為方便你我，鼓勵一下我唄！

————————————————————————————————————————————————————————

Repartitions運算元：

重新分割槽

重新分割槽，可以增多分割槽，可以減少分割槽

是一個有shuffle類的運算元

重新分割槽，可以增多分割槽，可以減少分割槽

是一個有shuffle類的運算元

Coalesce也是重新分割槽，可以增多，也可以減少分割槽，還可以有一個過載的方法

可以產生shuffle，也可以不產生shuffle，預設是不產生shuffle的

Repartition底層封裝的是coalesce

Coalesce叢小的分割槽到大的分割槽，同時不讓產生shuffle，這樣是不起作用的。

增多分割槽經常repartition

減少分割槽經常用coalesce

用coalesce，同時不設定true產生shuffle，這樣子是不產生分割槽的。

Repartition重新分割槽，可多可少，預設有shuffle

coalesce重新分割槽，可多可少，可以設定分割槽是否產生shuffle，coalesce（numpartition，shuffle=true/false），預設是false

由少的分割槽到多的分割槽，不讓產生shuffle，是不起作用的

Mappartition with index

Groupbykey：

將tuple型別的list轉換成用（K,V）格式的RDD用parallelizepares

依據key的值，將所有的value的值放在一起。

zipwithIndex:轉換成（k,v）格式的RDD

Zip就是將兩個RDD壓縮成一個RDD

Tuple型別的RDD也可以壓縮成k,v

Groupbykey

Reduce:累加求總數

Countbykey:數一數相同的key有幾個，按照pairRDD,統計相同的key有幾個

Countbyvalue，可以作用到（k,v）格式的RDD上，也可以作用在非（K,V）格式的RDD上