什么是RDD

• RDD（Resilient Distributed Dataset）叫做彈性分布式數(shù)據(jù)集镇辉，是Spark中最基本的數(shù)據(jù)抽象，它代表一個(gè)不可變、可分區(qū)习劫、里面的元素可并行計(jì)算的集合。RDD具有數(shù)據(jù)流模型的特點(diǎn)：自動容錯(cuò)嚼隘、位置感知性調(diào)度和可伸縮性诽里。RDD允許用戶在執(zhí)行多個(gè)查詢時(shí)顯式地將工作集緩存在內(nèi)存中，后續(xù)的查詢能夠重用工作集飞蛹，這極大地提升了查詢速度谤狡。

什么是DataFrame

• DataFrame引入了schema和off-heap
  schema : RDD每一行的數(shù)據(jù), 結(jié)構(gòu)都是一樣的灸眼，這個(gè)結(jié)構(gòu)就存儲在schema中。 Spark通過schema就能夠讀懂?dāng)?shù)據(jù), 因此在通信和IO時(shí)就只需要序列化和反序列化數(shù)據(jù), 而結(jié)構(gòu)的部分就可以省略了墓懂。

什么是DataSet

• DataSet結(jié)合了RDD和DataFrame的優(yōu)點(diǎn)焰宣，并帶來的一個(gè)新的概念Encoder。
  當(dāng)序列化數(shù)據(jù)時(shí)捕仔，Encoder產(chǎn)生字節(jié)碼與off-heap進(jìn)行交互匕积，能夠達(dá)到按需訪問數(shù)據(jù)的效果，而不用反序列化整個(gè)對象榜跌。Spark還沒有提供自定義Encoder的API闪唆，但是未來會加入。

  
  
  三者之間轉(zhuǎn)換

RDD的寬依賴和窄依賴

• 由于RDD是粗粒度的操作數(shù)據(jù)集斜做，每個(gè)Transformation操作都會生成一個(gè)新的RDD苞氮，所以RDD之間就會形成類似流水線的前后依賴關(guān)系；RDD和它依賴的父RDD（s）的關(guān)系有兩種不同的類型瓤逼，即窄依賴（narrow dependency）和寬依賴（wide dependency）笼吟。如圖所示顯示了RDD之間的依賴關(guān)系。

• 窄依賴：是指每個(gè)父RDD的一個(gè)Partition最多被子RDD的一個(gè)Partition所使用霸旗，例如map贷帮、filter、union等操作都會產(chǎn)生窄依賴诱告；（獨(dú)生子女）

• 寬依賴：是指一個(gè)父RDD的Partition會被多個(gè)子RDD的Partition所使用撵枢，例如groupByKey、reduceByKey精居、sortByKey等操作都會產(chǎn)生寬依賴锄禽；（超生）

依賴關(guān)系流下面的視圖

在spark中，會根據(jù)RDD之間的依賴關(guān)系將DAG圖（有向無環(huán)圖）劃分為不同的階段靴姿，對于窄依賴沃但，由于partition依賴關(guān)系的確定性，partition的轉(zhuǎn)換處理就可以在同一個(gè)線程里完成佛吓，窄依賴就被spark劃分到同一個(gè)stage中宵晚，而對于寬依賴，只能等父RDD shuffle處理完成后维雇，下一個(gè)stage才能開始接下來的計(jì)算淤刃。

因此spark劃分stage的整體思路是：從后往前推，遇到寬依賴就斷開吱型，劃分為一個(gè)stage逸贾；遇到窄依賴就將這個(gè)RDD加入該stage中。因此在圖2中RDD C,RDD D,RDD E,RDDF被構(gòu)建在一個(gè)stage中,RDD A被構(gòu)建在一個(gè)單獨(dú)的Stage中,而RDD B和RDD G又被構(gòu)建在同一個(gè)stage中。

在spark中耕陷，Task的類型分為2種：ShuffleMapTask和ResultTask掂名；

簡單來說，DAG的最后一個(gè)階段會為每個(gè)結(jié)果的partition生成一個(gè)ResultTask哟沫，即每個(gè)Stage里面的Task的數(shù)量是由該Stage中最后一個(gè)RDD的Partition的數(shù)量所決定的！而其余所有階段都會生成ShuffleMapTask锌介；之所以稱之為ShuffleMapTask是因?yàn)樗枰獙⒆约旱挠?jì)算結(jié)果通過shuffle到下一個(gè)stage中嗜诀；也就是說上圖中的stage1和stage2相當(dāng)于mapreduce中的Mapper,而ResultTask所代表的stage3就相當(dāng)于mapreduce中的reducer。

在之前動手操作了一個(gè)wordcount程序孔祸，因此可知隆敢，Hadoop中MapReduce操作中的Mapper和Reducer在spark中的基本等量算子是map和reduceByKey;不過區(qū)別在于：Hadoop中的MapReduce天生就是排序的；而reduceByKey只是根據(jù)Key進(jìn)行reduce崔慧，但spark除了這兩個(gè)算子還有其他的算子拂蝎；因此從這個(gè)意義上來說，Spark比Hadoop的計(jì)算算子更為豐富惶室。

RDD編程API(Transformation和Action)

• Transformation(轉(zhuǎn)換)

• Action(動作)

WordCount代碼編寫

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

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

    val conf = new SparkConf()
    /**
      * 如果這個(gè)參數(shù)不設(shè)置，默認(rèn)認(rèn)為你運(yùn)行的是集群模式
      * 如果設(shè)置成local代表運(yùn)行的是local模式
      */
    conf.setMaster("local")
    //設(shè)置任務(wù)名
    conf.setAppName("WordCount")
    //創(chuàng)建SparkCore的程序入口
    val sc = new SparkContext(conf)
    //讀取文件 生成RDD
    val file: RDD[String] = sc.textFile("E:\\hello.txt")
    //把每一行數(shù)據(jù)按照逾滥，分割
    val word: RDD[String] = file.flatMap(_.split(","))
    //讓每一個(gè)單詞都出現(xiàn)一次
    val wordOne: RDD[(String, Int)] = word.map((_,1))
    //單詞計(jì)數(shù)
    val wordCount: RDD[(String, Int)] = wordOne.reduceByKey(_+_)
    //按照單詞出現(xiàn)的次數(shù) 降序排序
    val sortRdd: RDD[(String, Int)] = wordCount.sortBy(tuple => tuple._2,false)
    //將最終的結(jié)果進(jìn)行保存
    sortRdd.saveAsTextFile("E:\\result")

    sc.stop()
  }