一、RDD的概述

1.1　什么是RDD？

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

1.2　RDD的屬性

（1）一組分片（Partition），即數(shù)據(jù)集的基本組成單位。對(duì)于RDD來說，每個(gè)分片都會(huì)被一個(gè)計(jì)算任務(wù)處理，并決定并行計(jì)算的粒度。用戶可以在創(chuàng)建RDD時(shí)指定RDD的分片個(gè)數(shù)，如果沒有指定，那么就會(huì)采用默認(rèn)值。默認(rèn)值就是程序所分配到的CPU Core的數(shù)目。

（3）RDD之間的依賴關(guān)系。RDD的每次轉(zhuǎn)換都會(huì)生成一個(gè)新的RDD，所以RDD之間就會(huì)形成類似于流水線一樣的前后依賴關(guān)系。在部分分區(qū)數(shù)據(jù)丟失時(shí)，Spark可以通過這個(gè)依賴關(guān)系重新計(jì)算丟失的分區(qū)數(shù)據(jù)，而不是對(duì)RDD的所有分區(qū)進(jìn)行重新計(jì)算。

（3）RDD之間的依賴關(guān)系。RDD的每次轉(zhuǎn)換都會(huì)生成一個(gè)新的RDD，所以RDD之間就會(huì)形成類似于流水線一樣的前后依賴關(guān)系。在部分分區(qū)數(shù)據(jù)丟失時(shí)，Spark可以通過這個(gè)依賴關(guān)系重新計(jì)算丟失的分區(qū)數(shù)據(jù)，而不是對(duì)RDD的所有分區(qū)進(jìn)行重新計(jì)算。

（4）一個(gè)Partitioner，即RDD的分片函數(shù)。當(dāng)前Spark中實(shí)現(xiàn)了兩種類型的分片函數(shù)，一個(gè)是基于哈希的HashPartitioner，另外一個(gè)是基于范圍的RangePartitioner。只有對(duì)于于key-value的RDD，才會(huì)有Partitioner，非key-value的RDD的Parititioner的值是None。Partitioner函數(shù)不但決定了RDD本身的分片數(shù)量，也決定了parent RDD Shuffle輸出時(shí)的分片數(shù)量。

（5）一個(gè)列表，存儲(chǔ)存取每個(gè)Partition的優(yōu)先位置（preferred location）。對(duì)于一個(gè)HDFS文件來說，這個(gè)列表保存的就是每個(gè)Partition所在的塊的位置。按照“移動(dòng)數(shù)據(jù)不如移動(dòng)計(jì)算”的理念，Spark在進(jìn)行任務(wù)調(diào)度的時(shí)候，會(huì)盡可能地將計(jì)算任務(wù)分配到其所要處理數(shù)據(jù)塊的存儲(chǔ)位置。

1.3　WordCount粗圖解RDD

image

其中hello.txt

image

二、RDD的創(chuàng)建方式

2.1　通過讀取文件生成的

由外部存儲(chǔ)系統(tǒng)的數(shù)據(jù)集創(chuàng)建，包括本地的文件系統(tǒng)，還有所有Hadoop支持的數(shù)據(jù)集，比如HDFS、Cassandra、HBase等

scala> val file = sc.textFile("/spark/hello.txt")

image

2.2　通過并行化的方式創(chuàng)建RDD

由一個(gè)已經(jīng)存在的Scala集合創(chuàng)建。

scala> val array = Array(1,2,3,4,5)
array: Array[Int] = Array(1, 2, 3, 4, 5)

scala> val rdd = sc.parallelize(array)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[27] at parallelize at <console>:26

scala>

image

2.3　其他方式

讀取數(shù)據(jù)庫等等其他的操作。也可以生成RDD。

RDD可以通過其他的RDD轉(zhuǎn)換而來的。

三、RDD編程API

Spark支持兩個(gè)類型（算子）操作：Transformation和Action

3.1　Transformation

主要做的是就是將一個(gè)已有的RDD生成另外一個(gè)RDD。Transformation具有l(wèi)azy特性(延遲加載)。Transformation算子的代碼不會(huì)真正被執(zhí)行。只有當(dāng)我們的程序里面遇到一個(gè)action算子的時(shí)候，代碼才會(huì)真正的被執(zhí)行。這種設(shè)計(jì)讓Spark更加有效率地運(yùn)行。

常用的Transformation：

3.2　Action

觸發(fā)代碼的運(yùn)行，我們一段spark代碼里面至少需要有一個(gè)action操作。

3.3　Spark WordCount代碼編寫

使用maven進(jìn)行項(xiàng)目構(gòu)建

（1）使用scala進(jìn)行編寫

查看官方網(wǎng)站，需要導(dǎo)入2個(gè)依賴包

image

詳細(xì)代碼

SparkWordCountWithScala.scala

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()
  }

運(yùn)行結(jié)果

image

（2）使用java jdk7進(jìn)行編寫

SparkWordCountWithJava7.java

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;

public class SparkWordCountWithJava7 {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("WordCount");
        JavaSparkContext sc = new JavaSparkContext(conf);
        JavaRDD<String> fileRdd = sc.textFile("E:\\hello.txt");

        JavaRDD<String> wordRDD = fileRdd.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String line) throws Exception {
                return Arrays.asList(line.split(",")).iterator();
            }
        });

        JavaPairRDD<String, Integer> wordOneRDD = wordRDD.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String word) throws Exception {
                return new Tuple2<>(word, 1);
            }
        });

        JavaPairRDD<String, Integer> wordCountRDD = wordOneRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer i1, Integer i2) throws Exception {
                return i1 + i2;
            }
        });

        JavaPairRDD<Integer, String> count2WordRDD = wordCountRDD.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
            @Override
            public Tuple2<Integer, String> call(Tuple2<String, Integer> tuple) throws Exception {
                return new Tuple2<>(tuple._2, tuple._1);
            }
        });

        JavaPairRDD<Integer, String> sortRDD = count2WordRDD.sortByKey(false);

        JavaPairRDD<String, Integer> resultRDD = sortRDD.mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(Tuple2<Integer, String> tuple) throws Exception {
                return new Tuple2<>(tuple._2, tuple._1);
            }
        });

        resultRDD.saveAsTextFile("E:\\result7");

    }
}

（3）使用java jdk8進(jìn)行編寫

lambda表達(dá)式

SparkWordCountWithJava8.java

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.util.Arrays;

public class SparkWordCountWithJava8 {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        conf.setAppName("WortCount");
        conf.setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);

        JavaRDD<String> fileRDD = sc.textFile("E:\\hello.txt");
        JavaRDD<String> wordRdd = fileRDD.flatMap(line -> Arrays.asList(line.split(",")).iterator());
        JavaPairRDD<String, Integer> wordOneRDD = wordRdd.mapToPair(word -> new Tuple2<>(word, 1));
        JavaPairRDD<String, Integer> wordCountRDD = wordOneRDD.reduceByKey((x, y) -> x + y);
        JavaPairRDD<Integer, String> count2WordRDD = wordCountRDD.mapToPair(tuple -> new Tuple2<>(tuple._2, tuple._1));
        JavaPairRDD<Integer, String> sortRDD = count2WordRDD.sortByKey(false);
        JavaPairRDD<String, Integer> resultRDD = sortRDD.mapToPair(tuple -> new Tuple2<>(tuple._2, tuple._1));
        resultRDD.saveAsTextFile("E:\\result8");

    }

3.4　WordCount執(zhí)行過程圖

image

四、RDD的寬依賴和窄依賴

4.1　RDD依賴關(guān)系的本質(zhì)內(nèi)幕

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

image

從圖中可知：

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

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

需要特別說明的是對(duì)join操作有兩種情況：

（1）圖中左半部分join：如果兩個(gè)RDD在進(jìn)行join操作時(shí)，一個(gè)RDD的partition僅僅和另一個(gè)RDD中已知個(gè)數(shù)的Partition進(jìn)行join，那么這種類型的join操作就是窄依賴，例如圖1中左半部分的join操作(join with inputs co-partitioned)；

（2）圖中右半部分join：其它情況的join操作就是寬依賴,例如圖1中右半部分的join操作(join with inputs not co-partitioned)，由于是需要父RDD的所有partition進(jìn)行join的轉(zhuǎn)換，這就涉及到了shuffle，因此這種類型的join操作也是寬依賴。

總結(jié)：

在這里我們是從父RDD的partition被使用的個(gè)數(shù)來定義窄依賴和寬依賴，因此可以用一句話概括下：如果父RDD的一個(gè)Partition被子RDD的一個(gè)Partition所使用就是窄依賴，否則的話就是寬依賴。因?yàn)槭谴_定的partition數(shù)量的依賴關(guān)系，所以RDD之間的依賴關(guān)系就是窄依賴；由此我們可以得出一個(gè)推論：即窄依賴不僅包含一對(duì)一的窄依賴，還包含一對(duì)固定個(gè)數(shù)的窄依賴。

一對(duì)固定個(gè)數(shù)的窄依賴的理解：即子RDD的partition對(duì)父RDD依賴的Partition的數(shù)量不會(huì)隨著RDD數(shù)據(jù)規(guī)模的改變而改變；換句話說，無論是有100T的數(shù)據(jù)量還是1P的數(shù)據(jù)量，在窄依賴中，子RDD所依賴的父RDD的partition的個(gè)數(shù)是確定的，而寬依賴是shuffle級(jí)別的，數(shù)據(jù)量越大，那么子RDD所依賴的父RDD的個(gè)數(shù)就越多，從而子RDD所依賴的父RDD的partition的個(gè)數(shù)也會(huì)變得越來越多。

4.2　依賴關(guān)系下的數(shù)據(jù)流視圖

image

在spark中，會(huì)根據(jù)RDD之間的依賴關(guān)系將DAG圖（有向無環(huán)圖）劃分為不同的階段，對(duì)于窄依賴，由于partition依賴關(guān)系的確定性，partition的轉(zhuǎn)換處理就可以在同一個(gè)線程里完成，窄依賴就被spark劃分到同一個(gè)stage中，而對(duì)于寬依賴，只能等父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；

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

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

原博客:https://www.cnblogs.com/qingyunzong/p/8899715.html#_label0_0