大數(shù)據(jù)技術之Spark（一）——Spark概述

前言

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

一、Spark基礎

1.1 Spark是什么

Spark 是一種由 Scala 語言開發(fā)的快速、通用、可擴展的大數(shù)據(jù)分析引擎。

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

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

Spark Streaming：流式計算分解成一系列小的批處理作業(yè)利用spark輕量級低時延的框架來支持流數(shù)據(jù)處理，目前已經(jīng)支持Kafka，F(xiàn)lume等；

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

1.2 Spark VS Hadoop

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

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

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

1.3 Spark優(yōu)勢及特點

1.3.1 優(yōu)秀的數(shù)據(jù)模型和豐富計算抽象

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

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

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

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

1.3.2 spark的特點

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

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

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

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

1.4 Spark 運行環(huán)境

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

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

② standalone 獨立集群模式

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

③ standalone-HA 高可用模式

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

④ on yarn 集群模式

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

⑤ on mesos 集群模式

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

⑥ on cloud 集群模式

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

1.5 Spark運行架構

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

1、一個行動算子對應一個job
2、job中分為多個stage，通過是否shuffle來判定。shuffle一次一個stage（DAG有向無環(huán)圖）
3、最后task的數(shù)量由rdd算子的分區(qū)數(shù)量來決定。rdd算子定義了幾個分區(qū)，就有幾個task

1.5.1 Driver

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

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

1.5.2 Executor

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

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

1.5.3 Master & Worker

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

1.5.4 ApplicationMaster

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

1.6 核心概念

1.6.1 Executor與Core

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

1.6.2 并行度（Parallelism）

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

1.6.3 有向無環(huán)圖（DAG）

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

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

工作原理

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

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

寬依賴與窄依賴

二、安裝方式

Linux中Spark的安裝方式

輸入spark-shell運行，顯示Spark context(sc)和Spark session(spark)。
SparkContext和SparkSession的區(qū)別

WordCount方法實現(xiàn)

// 1. 讀取文件，獲取一行一行的數(shù)據(jù)
// hello world
val lines = sc.textFile("datas")

// 2. 將一行數(shù)據(jù)進行拆分，形成一個一個的單詞（分詞）
// "hello world" => hello, world, hello, world
val words = lines.flatMap(_.split(" "))

// 3. 將數(shù)據(jù)根據(jù)單詞進行分組，便于統(tǒng)計
// (hello, hello, hello), (world, world)
val wordGroup = words.groupBy(word => word)

// 4. 對分組后的數(shù)據(jù)進行轉換
//  (hello, hello, hello), (world, world)
// (hello, 3), (world, 3)
val wordToCount = wordGroup.map {
	case(word, list) => {
		(word, list.size)
	}
}

// 5. 將轉換結果采集到控制臺打印出來
wordToCount.collect.foreach(println)

Java中SparkContext的創(chuàng)建與配置

object WordCount {
  def main(args: Array[String]): Unit = {
// Spark配置
    val conf = new SparkConf().setMaster("local[*]").setAppName("wordcount")
// 創(chuàng)建SparkContext
    val sc = SparkContext.getOrCreate(conf)

// 從hdfs中讀取文件
    val rdd = sc.textFile("hdfs://192.168.153.139:9000/tmp/words.txt")
    val result = rdd.flatMap(x=>x.split("\t")).map((x=>(x,1))).reduceByKey(_+_)
// 保存文件并輸出
    result.saveAsTextFile("hdfs://192.168.153.139:9000/tmp/wordsResult.txt")
  }

}

編譯jar包

程序編寫完成后，將生成的jar包放到linux中。這里我放在了/opt目錄下，嘗試編譯代碼。由于輸入輸出固定，我們可以直接在9870端口中查看是否成功輸出名為wordResult.txt的文件

[root@hadoop02 opt]# spark-submit --class org.example.WordCount --master local[*] ./sparkstu-1.0-SNAPSHOT.jar 

這樣就成功啦~

三、Spark核心編程

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

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

3.1 RDD

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

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

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

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

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

3.1.1 RDD創(chuàng)建：parallelize和makeRDD

在spark中創(chuàng)建RDD的創(chuàng)建方式可以分為四種：

1）從集合（內存）中創(chuàng)建RDD

從集合中創(chuàng)建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：
從底層代碼實現(xiàn)來講，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）從外部存儲（文件）創(chuàng)建RDD

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

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

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

3）從其他RDD創(chuàng)建

通過一個RDD運算完后，再產(chǎn)生新的RDD。

4）直接創(chuàng)建RDD（new）

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

3.1.2 RDD并行度與分區(qū)

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

// RDD的并行度 & 分區(qū)
// makeRDD方法可以傳遞第二個參數(shù)，用于表示分區(qū)的數(shù)量
// makeRDD第二個參數(shù)可以不傳遞，會使用默認值
val rdd = sc.makeRDD( List(1,2,3,4), 2 )

// 將處理的數(shù)據(jù)保存成分區(qū)文件
rdd.saveAsTextFile("hdfs://hadoop02:9000/tmp/rdd2_output")

3.1.3 RDD算子

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

3.1.4 RDD序列化

RDD為什么需要序列化，實現(xiàn)序列化的方法

3.1.5 RDD依賴關系

RDD依賴關系：寬依賴、窄依賴

3.1.6 RDD持久化

RDD持久化：cache/persisit緩存，checkpoint檢查點

3.1.7 RDD分區(qū)器Partitioner

?val rdd = sc.makeRDD(lIST(
    ("AAA","1231231234"),
    ("BBB","12341234123"),
    ("CCC","1212123"),
    ("DDD","111222333")
),3)

val partRDD = rdd.partitionBy(new MyPartitioner)
/**
  * 自定義分區(qū)器
  * 1. 繼承Partitioner
  * 2. 重寫方法
  */
    class MyPartitioner extends Partitioner {
        // 分區(qū)數(shù)量
        override def numPatririons: Int = 3

        // 根據(jù)數(shù)據(jù)的key值，返回數(shù)據(jù)所在的分區(qū)索引（從0開始）
        override def getPartition(key: Any): Int = {
            key match {
                case "AAA" => 0
                case "BBB" => 1
                case _ => 2
            }
        }
    }

3.1.8RDD文件讀取與保存

Spark的數(shù)據(jù)讀取及數(shù)據(jù)保存可以從兩個維度來作區(qū)分：文件格式 和 文件系統(tǒng)。

文件格式：text文件、csv文件、sequence文件以及Object文件
文件系統(tǒng)：本地文件系統(tǒng)、HDFS、HBASE以及數(shù)據(jù)庫

text文件：
讀取：sc.textFile("input/1.txt")
保存：rdd.saveAsTextFile("output1");

sequence文件：
讀?。簊c.sequenceFile[Int,Int]("output").collect().foreach(println)
保存：rdd.saveAsSequenceFile("output2");

object文件：
讀?。簊c.objectFile[Int]("output").collect().foreach(println)
保存：rdd.saveAsObjectFile("output3");

3.2 累加器

舉例說明：實現(xiàn)數(shù)字的累加運算

object AccDemo {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("checkPoint")
    val sc = new SparkContext(conf)

	val rdd = sc.makeRDD(List(1,2,3,4))
// 方法一：reduce 既有分區(qū)內計算，又有分區(qū)間計算
	val i = rdd.reduce(_+_)
	println(i)
// 方法二：使用foreach實現(xiàn)累加
	var sum = 0;
	rdd.foreach(
	  num => { sum += num }
	)
	println("sum = " + sum)

	sc.stop()
  }
}

出來的結果明顯與期望不符：

——為什么結果不是10？
foreach不是分布式的，但是spark是分布式的。
也就意味著每個分區(qū)都會執(zhí)行foreach，導致結果不確定。
——為什么結果是0？
1. sum在foreach之外，所以sum在Driver端進行初始化，sum=0
2. sum在foreach中執(zhí)行相加操作
3. 由于是分布式執(zhí)行，所以兩個Executor都會執(zhí)行sum+=num的操作
4. 由于存在閉包檢測，所以sum=0被傳到Executor端
5. 計算完畢之后，應該將sum計算的返回結果給Driver。但這里不存在返回操作。
6. 所以說，執(zhí)行計算的是Executor端的sum，對Driver端的sum不造成影響。

那當我們想要實現(xiàn)累加的功能，但閉包的計算結果無法返回到Driver端，此時需要一種特殊結構，可以將Driver端的數(shù)據(jù)傳給Executor，當計算結束之后，可以將結果返回給Driver，類似于如下圖：

總結：
累加器用來把 Executor 端變量信息聚合到 Driver 端。
在 Driver 程序中定義的變量，在Executor 端的每個 Task 都會得到這個變量的一份新的副本，每個 task 更新這些副本的值后，傳回 Driver 端進行 merge。

累加器的實現(xiàn)

object acc {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("checkPoint")
    val sc = new SparkContext(conf)

    val rdd = sc.makeRDD(List(1,2,3,4))
    
    // 獲取系統(tǒng)累加器
    // spark默認提供了簡單的數(shù)據(jù)聚合的累加器
    var sumAcc = sc.longAccumulator("sum")
	rdd.foreach(
	  // 使用累加器
      num => { sumAcc.add(num) }
    )
	// 獲取累加器的值
	println(sumAcc.value)
    

    sc.stop()
  }
}

注意：一般情況下，累加器會放置在行動算子中進行操作。

3.3 廣播變量

3.3.1 實現(xiàn)原理

廣播變量用來高效分發(fā)較大的對象。向所有工作節(jié)點發(fā)送一個較大的只讀值，以供一個或多個 Spark 操作使用。比如，如果你的應用需要向所有節(jié)點發(fā)送一個較大的只讀查詢表，廣播變量用起來都很順手。在多個并行操作中使用同一個變量，但是 Spark 會為每個任務分別發(fā)送。文章來源地址http://www.zghlxwxcb.cn/news/detail-622903.html

3.3.2 基礎編程

object BraodCastDemo {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("BroadCast").setMaster("local[*]")
    val sc = new SparkContext(conf)

// 聲明廣播變量
    val arr = Array("hello","hi","com on baby")
    val broadcastVar = sc.broadcast(arr)
    
    val rdd = sc.parallelize(Array((1,"leader"),(2,"teamLeader"),(3,"worker")))
    val rdd2 = rdd.mapValues(x=>{
      println("value is" + x)
      
      // 使用廣播變量
      broadcastVar.value(2) + ":" + x
    })
    rdd2.collect.foreach(println)

	sc.stop()
  }
}

到了這里，關于大數(shù)據(jù)技術之Spark（一）——Spark概述的文章就介紹完了。如果您還想了解更多內容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關文章，希望大家以后多多支持TOY模板網(wǎng)！