大數據技術之Spark（一）——Spark概述

大數據技術之Spark（一）——Spark概述

---

文章目錄

• 前言
• 一、Spark基礎
• • 1.1 Spark是什么
  • 1.2 Spark VS Hadoop
  • 1.3 Spark優勢及特點
  • • 1.3.1 優秀的數據模型和豐富計算抽象
    • 1.3.3 spark的特點
  • 1.4 Spark 運行環境
  • 1.5 Spark運行架構
  • • 1.5.1 Driver
    • 1.5.2 Executor
    • 1.5.3 Master & Worker
    • 1.5.4 ApplicationMaster
  • 1.6 核心概念
  • • 1.6.1 Executor與Core
    • 1.6.2 并行度（Parallelism）
    • 1.6.3 有向無環圖（DAG）
    • • 工作原理
      • 寬依賴與窄依賴
• 二、安裝方式
• • WordCount方法實現
• 三、Spark核心編程
• • 3.1 RDD
  • • 3.1.1 RDD創建：parallelize和makeRDD
    • • 1）從集合（內存）中創建RDD
      • 2）從外部存儲（文件）創建RDD
      • 3）從其他RDD創建
      • 4）直接創建RDD（new）
    • 3.1.2 RDD并行度與分區
    • 3.1.3 RDD算子
  • 3.2 累加器
  • 3.3 廣播變量

---

前言

Apache Spark是一個開源的、強大的分布式查詢和處理引擎，它提供MapReduce的靈活性和可擴展性，但速度明顯要快上很多；拿數據存儲在內存中的時候來說，它比Apache Hadoop 快100倍，訪問磁盤時也要快上10倍。

一、Spark基礎

1.1 Spark是什么

Spark 是一種由 Scala 語言開發的快速、通用、可擴展的大數據分析引擎。

Spark Core：Spark Core包含Spark的基本功能，如內存計算、任務調度、部署模式、故障恢復、存儲管理等。Spark建立在統一的抽象RDD之上，使其可以以基本一致的方式應對不同的大數據處理場景；通常所說的Apache Spark，就是指Spark Core；

Spark SQL：兼容HIVE數據，提供比Hive更快的查詢速度（10~100x）的分布式SQL引擎，開發者可以輕松地使用SQL命令進行查詢，并進行更復雜的數據分析；

Spark Streaming：流式計算分解成一系列小的批處理作業利用spark輕量級低時延的框架來支持流數據處理，目前已經支持Kafka，Flume等；

MLilb：提供基于Spark的機器學習算法庫，包括聚類、分類、回歸、協同過濾等，降低了機器學習的門檻，開發人員只要具備一定的理論知識就能進行機器學習的工作；
GraphX：提供圖形計算框架，與Pregel/GraphLab兼容。

1.2 Spark VS Hadoop

盡管 Spark 相對于 Hadoop 而言具有較大優勢，但 Spark 并不能完全替代 Hadoop，Spark 主要用于替代Hadoop中的 MapReduce 計算模型。存儲依然可以使用 HDFS，但是中間結果可以存放在內存中；調度可以使用 Spark 內置的，也可以使用更成熟的調度系統 YARN 等。

HadopSpark

類型	分布式基礎平臺, 包含計算, 存儲, 調度	分布式計算工具
場景	大規模數據集上的批處理	迭代計算, 交互式計算, 流計算
價格	對機器要求低, 便宜	對內存有要求, 相對較貴
編程范式	Map+Reduce, API 較為底層, 算法適應性差	RDD 組成 DAG 有向無環圖, API 較為頂層, 方便使用
數據存儲結構	MapReduce 中間計算結果存在 HDFS 磁盤上, 延遲大	RDD 中間運算結果存在內存中 , 延遲小
運行方式	Task 以進程方式維護, 任務啟動慢	Task 以線程方式維護, 任務啟動快

Spark 和 Hadoop 的根本差異是多個作業之間的數據通信問題 ；Spark 多個作業之間數據通信是基于內存，而 Hadoop 是基于磁盤。
實際上，Spark 已經很好地融入了 Hadoop 生態圈，并成為其中的重要一員，它可以借助于 YARN 實現資源調度管理，借助于 HDFS 實現分布式存儲。

此外，Hadoop 可以使用廉價的、異構的機器來做分布式存儲與計算，但是，Spark 對硬件的要求稍高一些，對內存與 CPU 有一定的要求。

1.3 Spark優勢及特點

1.3.1 優秀的數據模型和豐富計算抽象

首先看看MapReduce，它提供了對數據訪問和計算的抽象，但是對于數據的復用就是簡單的將中間數據寫到一個穩定的文件系統中(例如 HDFS)，所以會產生數據的復制備份，磁盤的I/O以及數據的序列化，所以在遇到需要在多個計算之間復用中間結果的操作時效率就會非常的低。而這類操作是非常常見的，例如迭代式計算，交互式數據挖掘，圖計算等。

因此 AMPLab 提出了一個新的模型，叫做 RDD。

• RDD 是一個可以容錯且并行的數據結構（其實可以理解成分布式的集合，操作起來和操作本地集合一樣簡單)，它可以讓用戶顯式的將中間結果數據集保存在 內存 中，并且通過控制數據集的分區來達到數據存放處理最優化。同時 RDD 也提供了豐富的 API (map、reduce、filter、foreach、redeceByKey...)來操作數據集。

后來 RDD 被 AMPLab 在一個叫做 Spark 的框架中提供并開源。

1.3.3 spark的特點

快：與 Hadoop 的 MapReduce 相比，Spark 基于內存的運算要快 100 倍以上，基于硬盤的運算也要快 10 倍以上。Spark 實現了高效的 DAG 執行引擎，可以通過基于內存來高效處理數據流。

易用：Spark 支持 Java、Python、R 和 Scala 的 API，還支持超過 80 種高級算法，使用戶可以快速構建不同的應用。而且 Spark 支持交互式的 Python 和 Scala 的 shell，可以非常方便地在這些 shell 中使用 Spark 集群來驗證解決問題的方法。

通用：Spark 提供了統一的解決方案。Spark 可以用于批處理、交互式查詢(Spark SQL)、實時流處理(Spark Streaming)、機器學習(Spark MLlib)和圖計算(GraphX)，這些不同類型的處理都可以在同一個應用中無縫使用。

兼容性：Spark 可以非常方便地與其他的開源產品進行融合。比如，Spark 可以使用 Hadoop 的 YARN 和 Apache Mesos 作為它的資源管理和調度器，并且可以處理所有 Hadoop 支持的數據，包括 HDFS、HBase 和 Cassandra 等。這對于已經部署 Hadoop 集群的用戶特別重要，因為不需要做任何數據遷移就可以使用 Spark 的強大處理能力。

1.4 Spark 運行環境

① local 本地模式(單機) - 不需要其他任何節點資源就可以在本地執行Spark代碼的環境

• 學習測試使用。
• 分為 local 單線程和 local-cluster 多線程。

② standalone 獨立集群模式

• 學習測試使用。
• 典型的 Mater/slave 模式。

③ standalone-HA 高可用模式

• 生產環境使用
• 基于 standalone 模式，使用 zk 搭建高可用，避免 Master 是有單點故障的。

④ on yarn 集群模式

• 生產環境使用
• 運行在 yarn 集群之上，由 yarn 負責資源管理，Spark 負責任務調度和計算。
• 好處：計算資源按需伸縮，集群利用率高，共享底層存儲，避免數據跨集群遷移。

⑤ on mesos 集群模式

• 國內使用較少
• 運行在 mesos 資源管理器框架之上，由 mesos 負責資源管理，Spark 負責任務調度和計算。

⑥ on cloud 集群模式

• 中小公司未來會更多的使用云服務
• 比如 AWS 的 EC2，使用這個模式能很方便的訪問 Amazon 的 S3。

1.5 Spark運行架構

Spark 框架的核心是一個計算引擎，整體來說，它采用了標準 master-slave 的結構。
如下圖所示，它展示了一個 Spark 執行時的基本結構。圖形中的 Driver 表示 master，負責管理整個集群中的作業任務調度。圖形中的 Executor 則是 slave，負責實際執行任務。

1.5.1 Driver

Spark驅動器節點，用于執行Spark任務中的main方法，負責實際代碼的執行工作。Driver在Spark作業執行時主要負責：

• 將用戶程序轉化為作業（job）
• 在Executor之間調度任務(task)
• 跟蹤Executor的執行情況
• 通過UI展示查詢運行情況

1.5.2 Executor

Spark Executor是集群中工作節點（Worker）中的一個JVM進程，負責在Spark 作業中運行具體任務（Task），任務彼此之間相互獨時啟動，并且始終伴隨著整個Spark 應用的生命周期而存在。如果有Executor節點發生了故障或崩潰，Spark 應用也可以繼續執行，會將出錯節點上的任務調度到其他Executor節點上繼續運行。
Executor有兩個核心功能：

• 負責運行組成Spark應用的任務，并將結果返回給驅動器進程
• 它們通過自身的塊管理器（Block Manager）為用戶程序中要求緩存的RDD 提供內存式存儲。RDD 是直接緩存在Executor進程內的，因此任務可以在運行時充分利用緩存數據加速運算。

1.5.3 Master & Worker

Spark集群的獨立部署環境中，不需要依賴其他的資源調度框架，自身就實現了資源調度的功能，所以環境中還有其他兩個核心組件：Master和Worker，這里的Master是一個進程，主要負責資源的調度和分配，并進行集群的監控等職責，類似于Yarn環境中的RM, 而Worker呢，也是進程，一個Worker運行在集群中的一臺服務器上，由Master分配資源對數據進行并行的處理和計算，類似于Yarn環境中NM。

1.5.4 ApplicationMaster

Hadoop用戶向YARN集群提交應用程序時,提交程序中應該包含ApplicationMaster，用于向資源調度器申請執行任務的資源容器Container，運行用戶自己的程序任務job，監控整個任務的執行，跟蹤整個任務的狀態，處理任務失敗等異常情況。
說的簡單點就是，ResourceManager（資源）和Driver（計算）之間的解耦合靠的就是ApplicationMaster。

1.6 核心概念

1.6.1 Executor與Core

Spark Executor是集群中運行在工作節點（Worker）中的一個JVM進程，是整個集群中的專門用于計算的節點。在提交應用中，可以提供參數指定計算節點的個數，以及對應的資源。這里的資源一般指的是工作節點Executor的內存大小和使用的虛擬CPU核（Core）數量。
應用程序相關啟動參數如下：

名稱說明

–num-executors	配置Executor的數量
–executor-memory	配置每個Executor的內存大小
–executor-cores	配置每個Executor的虛擬CPU core數量

1.6.2 并行度（Parallelism）

在分布式計算框架中一般都是多個任務同時執行，由于任務分布在不同的計算節點進行計算，所以能夠真正地實現多任務并行執行，記住，這里是并行，而不是并發。這里我們將整個集群并行執行任務的數量稱之為并行度。那么一個作業到底并行度是多少呢？這個取決于框架的默認配置。應用程序也可以在運行過程中動態修改。

1.6.3 有向無環圖（DAG）

這里所謂的有向無環圖，并不是真正意義的圖形，而是由Spark程序直接映射成的數據流的高級抽象模型。簡單理解就是將整個程序計算的執行過程用圖形表示出來,這樣更直觀，更便于理解，可以用于表示程序的拓撲結構。
DAG（Directed Acyclic Graph）有向無環圖是由點和線組成的拓撲圖形，該圖形具有方向，不會閉環。

sc.parallelize(1 to 10).map((_,1)).reduceByKey(_+_)

工作原理

• 根據RDD之間的依賴關系，形成一個DAG DAG
• Scheduler將DAG劃分為多個Stage

劃分依據：是否發生寬依賴（shuffle）
劃分規則：從后往前，遇到寬依賴且各位新的stage
每個stage由一組并行的Task組成

寬依賴與窄依賴

二、安裝方式

Spark安裝方式

WordCount方法實現

// 1. 讀取文件，獲取一行一行的數據 // hello world val lines = sc.textFile("datas")// 2. 將一行數據進行拆分，形成一個一個的單詞（分詞） // "hello world" => hello, world, hello, world val words = lines.flatMap(_.split(" "))// 3. 將數據根據單詞進行分組，便于統計 // (hello, hello, hello), (world, world) val wordGroup = words.groupBy(word => word)// 4. 對分組后的數據進行轉換 // (hello, hello, hello), (world, world) // (hello, 3), (world, 3) val wordToCount = wordGroup.map {case(word, list) => {(word, list.size)} }// 5. 將轉換結果采集到控制臺打印出來 wordToCount.collect.foreach(println)

三、Spark核心編程

Spark計算框架為了能夠進行 高并發 和 高吞吐 的數據處理，封裝了三大數據結構，用于處理不同的應用場景。三大數據結構分別是：

• RDD : 彈性分布式數據集
• 累加器：分布式共享 只寫 變量
• 廣播變量：分布式共享 只讀 變量

3.1 RDD

RDD（Resilient Distributed Dataset）叫做彈性分布式數據集，是Spark中最基本的 數據處理模型 。代碼中是一個抽象類，它代表一個彈性的、不可變、可分區、里面的元素可并行計算的集合。

Resilient ：它是彈性的，RDD 里面的中的數據可以保存在 內存 中或者 磁盤 里面。

存儲的彈性：內存與磁盤的自動切換
容錯的彈性：數據丟失可以自動回復
計算的彈性：計算出錯重試機制
分片的彈性：可根據需要重新分片

Distributed ： 它里面的元素是分布式存儲的，可以用于分布式計算。數據存儲在大數據集群不同節點上。

Dataset: 它是一個集合，封裝了計算邏輯，并不保存數據。

3.1.1 RDD創建：parallelize和makeRDD

在spark中創建RDD的創建方式可以分為四種：

1）從集合（內存）中創建RDD

從集合中創建RDD，Spark主要提供了兩個方法：parallelize和makeRDD

parallelize：

scala> var rdd = sc.parallelize(1 to 4) rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[168] at parallelize at <console>:24 scala> rdd.collect res107: Array[Int] = Array(1, 2, 3, 4)

makeRDD：
從底層代碼實現來講，makeRDD方法其實就是parallelize方法

scala> var rdd = sc.parallelize(1 to 4) rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[168] at parallelize at <console>:24 scala> rdd.collect res107: Array[Int] = Array(1, 2, 3, 4) ```

2）從外部存儲（文件）創建RDD

由外部存儲系統的數據集創建RDD包括：本地的文件系統，所有Hadoop支持的數據集，比如HDFS、HBase等。

// 從文件中創建RDD，將文件中的數據作為處理的數據源 // 1. path路徑默認以當前環境的根路徑為基準?？梢詫懡^對路徑，也可以寫相對路徑。 // 2. path路徑可以是文件的具體路徑，也可以是目錄名稱 // 3. path路徑還可以使用通配符 * // 4. path路徑可以是分布式存儲系統路徑HDFS scala> val rdd = sc.textFile("hdfs://hadoop02:9000/tmp/wordcount.txt") scala> val rdd = sc.textFile("file:///opt/stufile/words.txt")

textFile：以行為單位來讀取數據。讀取的數據都是字符串
wholeTextFiles：以文件為單位讀取數據。讀取的結果表示為元組。第一個元素表示文件路徑，第二個元素表示文件內容

3）從其他RDD創建

通過一個RDD運算完后，再產生新的RDD。

4）直接創建RDD（new）

使用new的方式直接構造RDD，一般由Spark框架自身使用。

3.1.2 RDD并行度與分區

默認情況下，Spark可以將一個作業切分多個任務后，發送給Executor節點并行計算，而能夠并行計算的任務數量我們稱之為并行度。
需要注意的是，并行執行的任務數量 ≠ 切分任務的數量。

// RDD的并行度 & 分區 // makeRDD方法可以傳遞第二個參數，用于表示分區的數量 // makeRDD第二個參數可以不傳遞，會使用默認值 val rdd = sc.makeRDD( List(1,2,3,4), 2 )// 將處理的數據保存成分區文件 rdd.saveAsTextFile("hdfs://hadoop02:9000/tmp/rdd2_output")

3.1.3 RDD算子

我們放在下一篇里面詳細介紹=>RDD算子

3.2 累加器

3.3 廣播變量