（一）概述

Spark計算框架封裝了三種主要的資料結構：RDD（彈性分布式資料集）、累加器（分布式共享只寫變數）、廣播變數（分布式共享支只讀變數）

（二）RDD

RDD：彈性分布式資料集是Spark中十分重要的一種資料結構，RDD 是可以并行操作的元素的集合，RDD具有幾大屬性：

1. RDD是由一系列partition組成
2. 函式是作用在每個partition(split)上的
3. RDD之間有一系列的依賴關系
4. RDD提供了一系列最佳的計算位置

RDD的運行程序可以通過下面這張表來表示，當Spark從資料源讀取資料之后，會在RDD中組成多個partition，這些partition可以并行進行操作計算，

（三）RDD的使用

RDD的創建主要有兩種方式，第一種是通過spark提供的parallelize方法來創建：

List<Integer> data = Arrays.asList(1, 2, 3, 4, 5);
JavaRDD<Integer> distData = sc.parallelize(data);

第二種方法是通過外部資料源直接生成：

JavaRDD<String> lines = sc.textFile("data/*");

接下來以scala和Java的案例分別介紹一下RDD的使用，案例依舊是最開始的WordCount例子，

首先是Java版本：

public class JavaWordCount {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("wordCount").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        //讀取檔案轉成RDD
        JavaRDD<String> lines = sc.textFile("data/*");
        //將每一行的單詞根據空格拆分
        JavaRDD<String> words = lines.flatMap((FlatMapFunction<String, String>) s -> Arrays.asList(s.split(" ")).iterator());
        //將Hello轉化為（Hello,1）這種格式
        JavaPairRDD<String, Integer> wordToOne = words.mapToPair((PairFunction<String, String, Integer>) s -> new Tuple2<String,Integer>(s,1));
        //根據key進行統計
        JavaPairRDD<String, Integer> wordToCount = wordToOne.reduceByKey((x, y) -> x + y);
        //輸出結果
        wordToCount.foreach((VoidFunction<Tuple2<String, Integer>>) stringIntegerTuple2 -> System.out.println(stringIntegerTuple2._1+stringIntegerTuple2._2));
        sc.close();
    }
}

接下來是scala版本：

object WordCount {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount");
    val sparkContext = new SparkContext(sparkConf);
    val lines: RDD[String] = sparkContext.textFile(path = "data/*");
    val words: RDD[String] = lines.flatMap(_.split(" "))
    val wordToOne: RDD[(String, Int)] = words.map(word => (word, 1))
    val wordToCount = wordToOne.reduceByKey((x, y) => x + y).foreach(println)
    sparkContext.stop();
  }
}

其實從這里也能看出來，scala版本會比Java版本看起來更加簡潔，
通過上面兩段代碼可以看出Spark的基本代碼流程：

1. 創建SparkConf物件，
2. 創建SparkContext物件，
3. 基于SparkContext物件創建RDD，對RDD進行處理，
4. 觸發Transformation類算子執行，
5. 關閉Spark背景關系物件，

（四）算子

算子從功能上可以分為Transformations轉換算子和Action行動算子，轉換算子用來做資料的轉換操作，比如map、flatMap、reduceByKey等都是轉換算子，這類算子通過懶加載執行，行動算子的作用是觸發執行，比如foreach、collect、count等都是行動算子，只有程式運行到行動算子時，轉換算子才會去執行，

一個應用程式中有幾個Action行動算子執行，就會有幾個Job運行，

4.1 常見的轉換算子

filter：過濾記錄，true保留，false過濾，

map：將一個RDD中的每個資料項，通過函式映射為一個新的元素，特點：輸入一條輸出一條，

flatMap：在map執行過后執行flat，和map類似，但是一個輸入項可以對應0到多個輸出項，

sample：隨機抽樣算子，根據傳進去的小數按照比例進行有放回或者無放回的抽樣，

reduceByKey：將相同的Key按照邏輯處理，

sortByKey：作用在Key-Value格式的RDD上，對Key進行升序或者降序排序，

4.2 常見的行動算子

count：回傳資料集中的元素數量，會在結果計算完成后回收到Driver端，

take(n)：回傳包含資料集前n個元素的集合，

first：回傳資料集中的第一個元素，

foreach：回圈遍歷資料集中的每個元素，運行相應的邏輯，

collect：將計算結果回收到Driver端，

（五）RDD的持久化

將RDD持久化的算子主要有三種：cache、persist、checkpoint，其中cache和persist都是懶加載，當有一個action算子觸發時才會執行，而checkpoint算子不僅能將RDD持久化到磁盤，還能切斷RDD之間的依賴關系，

持久化寸的是RDD中的partition，如果沒有使用持久化，一個RDD執行了Action算子后再次執行就需要重新拿資料，使用持久化可以節省代碼運行時間，

5.1 cache

cache默認是將RDD的資料持久化到記憶體中，cache的使用很簡單，只需要在RDD之后執行一次即可：

val lines: RDD[String] = sparkContext.textFile(path = "data/*");
lines.cache();

cache在原始碼中的實作等于persist

因此cache和persist的關系可以表示為：

cache()=persist()=persist(StorageLevel.MEMORY_ONLY)

5.2 persist

persist可以指定持久化的級別，其中MEMORY_ONLY和MEMORY_AND_DISK是最常用的兩種持久化方式，

使用方式和cache一樣：

val lines: RDD[String] = sparkContext.textFile(path = "data/*");
lines.persist();

cache和persist的注意事項：

1. cache和persist都是懶執行，必須要有一個action類算子觸發執行，
2. cache和persist的回傳值可以賦值給一個變數，在其他job中使用這個變數就是使用持久化的資料，

5.3 checkpoint

checkpoint會將RDD持久化到磁盤，還可以切斷RDD之間的依賴關系，

checkpoint的使用也比較簡單，首先在SparkContext中設定checkpoint在磁盤中保存的位置，接著執行RDD.checkpoint()

val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount");
val sparkContext = new SparkContext(sparkConf);
sparkContext.setCheckpointDir("./checkpoint")
val lines: RDD[String] = sparkContext.textFile(path = "data/*");
lines.checkpoint();

checkpoint的執行原理：

1. 從finalRDD往前找，當找到某一個RDD呼叫了checkpoint方法，給這個RDD打上一個標簽，
2. 啟動一個新的job，重新計算這個RDD的資料，最后將資料持久化，
   使用技巧：
   在使用checkpoint時先對RDD執行cache，這樣新啟動的job只需要將記憶體中的資料持久化就可以，節省一次計算的時間，

（六）Spark中的廣播變數

當Spark的轉換算子在作業時，在函式方法中使用到的所有外部變數都是一個獨立的副本，這些變數會隨著任務的執行被復制到每臺機器上面，但是Spark提供了兩種共享變數的型別，分別是廣播變數和累加器，

廣播變數允許程式員在每臺機器上快取一個只讀變數，而不是隨任務一起發送它的副本，例如，它們可用于以有效的方式為每個節點提供大型輸入資料集的副本，廣播變數通過 SparkContext.broadcast(v) 方法創建，通過呼叫value方法獲取具體的值，

public class TestBroadcast {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("wordCount").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        Broadcast<int[]> broadcastVar = sc.broadcast(new int[] {1, 2, 3});
        broadcastVar.value();
    }
}

要釋放廣播變數復制到執行程式的資源，需要呼叫unpersist()方法，要永久釋放廣播變數使用的所有資源，需要呼叫destroy()方法，

broadcastVar.unpersist();
broadcastVar.destroy();

.

（七）Spark中的累加器

累加器也是共享變數中的一種，Spark的計算會被分配到各個作業節點中，因此如果用普通的i++的方式無法獲取到預期的累加效果，Spark提供了累加器資料模型，實作資料類加：

public class TestAdd {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("wordCount").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        List<Integer> data = Arrays.asList(1, 2, 3, 4, 5);
        JavaRDD<Integer> rdd = sc.parallelize(data);
        LongAccumulator accum = sc.sc().longAccumulator();
        rdd.foreach(x -> accum.add(1));
        System.out.println(accum.value());
    }
}

（八）總結

Spark的核心在于RDD，理解了RDD就相當于對Spark編程徹底入門了，我是魚仔，我們下期再見，