本文目錄
第1章 Spark概述
1.1 什麼是Spark
1.2 Spark内置子產品
1.3 Spark特點
第2章 Spark運作模式
2.1 Spark安裝位址
2.2 重要角色
2.2.1 Driver（驅動器）
2.2.2 Executor（執行器）
2.3 Local模式
2.3.1 概述
2.3.2 安裝使用
2.4 Standalone模式
2.4.1 概述
2.4.2 安裝使用
2.4.3 JobHistoryServer配置
2.4.4 HA配置
2.5 Yarn模式（重點）
2.5.1 概述
2.5.2 安裝使用
2.5.3 日志檢視
2.6 Mesos模式(了解)
2.7 幾種模式對比
第3章 案例實操
3.1 編寫WordCount程式
3.2 本地調試
第1章 Spark概述
1.1什麼是Spark
1、定義
Spark是一種基于記憶體的快速、通用、可擴充的大資料分析引擎。
2、曆史
2009年誕生于加州大學伯克利分校AMPLab，項目采用Scala編寫。
2010年開源；
2013年6月成為Apache孵化項目2014年2月成為Apache頂級項目。
1.2Spark内置子產品
Spark Core：實作了Spark的基本功能，包含任務排程、記憶體管理、錯誤恢複、與存儲系統互動等子產品。Spark Core中還包含了對彈性分布式資料集(Resilient Distributed DataSet，簡稱RDD)的API定義。
Spark SQL：是Spark用來操作結構化資料的程式包。通過Spark SQL，我們可以使用 SQL或者Apache Hive版本的SQL方言(HQL)來查詢資料。Spark SQL支援多種資料源，比如Hive表、Parquet以及JSON等。
Spark Streaming：是Spark提供的對實時資料進行流式計算的元件。提供了用來操作資料流的API，并且與Spark Core中的 RDD API高度對應。
Spark MLlib：提供常見的機器學習(ML)功能的程式庫。包括分類、回歸、聚類、協同過濾等，還提供了模型評估、資料 導入等額外的支援功能。
叢集管理器：Spark 設計為可以高效地在一個計算節點到數千個計算節點之間伸縮計 算。為了實作這樣的要求，同時獲得最大靈活性，Spark支援在各種叢集管理器(Cluster Manager)上運作，包括Hadoop YARN、Apache Mesos，以及Spark自帶的一個簡易排程 器，叫作獨立排程器。
Spark得到了衆多大資料公司的支援，這些公司包括Hortonworks、IBM、Intel、Cloudera、MapR、Pivotal、百度、阿裡、騰訊、京東、攜程、優酷洋芋。目前百度的Spark已應用于大搜尋、直達号、百度大資料等業務；阿裡利用GraphX建構了大規模的圖計算和圖挖掘系統，實作了很多生産系統的推薦算法；騰訊Spark叢集達到8000台的規模，是目前已知的世界上最大的Spark叢集。
1.3 Spark特點
1）快：與Hadoop的MapReduce相比，Spark基于記憶體的運算要快100倍以上，基于硬碟的運算也要快10倍以上。Spak實作了高效的DAG執行引擎，可以通過基于記憶體來高效處理資料流。計算的中間結果是存在于記憶體中的。
2）易用：Spark支援Java、Python和Scala的API，還支援超過80種進階算法，使使用者可以快速建構不同的應用。而且Spak支援互動式的Python和Scala的Shell，可以非常友善地在這些Shell中使用Spark叢集來驗證解決問題的方法。
3）通用：Spark提供了統一的解決方案。Spark可以用于批處理、互動式查詢（SparkSQL）、實時流處理（Spark Streaming）、機器學習（SparkMLlib）和圖計算（Graphx）。這些不同類型的處理都可以在同一個應用中無縫使用。減少了開發和維護的人力成本和部署平台的物力成本。
4）相容性：Spark可以非常友善地與其他的開源産品進行融合。比如，Spark可以使用Hadoop的YARN和Apache Mesos作為它的資源管理和排程器，并且可以處理所有Hadoop支援的資料，包括HDFS、HBase等。這對于已經部署Hadoop叢集的使用者特别重要，因為不需要做任何資料遷移就可以使用Spark的強大處理能力。
第2章 Spark運作模式
2.1 Spark安裝位址
1．官網位址
http://spark.apache.org/
2．文檔檢視位址
https://spark.apache.org/docs/2.1.1/
3．下載下傳位址
https://spark.apache.org/downloads.html
2.2 重要角色
2.2.1 Driver（驅動器）
Spark的驅動器是執行開發程式中的main方法的程序。它負責開發人員編寫的用來建立SparkContext、建立RDD，以及進行RDD的轉化操作和行動操作代碼的執行。如果你是用spark shell，那麼當你啟動Spark shell的時候，系統背景自啟了一個Spark驅動器程式，就是在Spark shell中預加載的一個叫作 sc的SparkContext對象。如果驅動器程式終止，那麼Spark應用也就結束了。主要負責：
1）把使用者程式轉為作業（JOB）
2）跟蹤Executor的運作狀況
3）為執行器節點排程任務
4）UI展示應用運作狀況
2.2.2 Executor（執行器）
Spark Executor是一個工作程序，負責在 Spark 作業中運作任務，任務間互相獨立。Spark 應用啟動時，Executor節點被同時啟動，并且始終伴随着整個 Spark 應用的生命周期而存在。如果有Executor節點發生了故障或崩潰，Spark 應用也可以繼續執行，會将出錯節點上的任務排程到其他Executor節點上繼續運作。主要負責：
1）負責運作組成 Spark 應用的任務，并将結果傳回給驅動器程序；
2）通過自身的塊管理器（Block Manager）為使用者程式中要求緩存的RDD提供記憶體式存儲。RDD是直接緩存在Executor程序内的，是以任務可以在運作時充分利用緩存資料加速運算。
2.3 Local模式
2.3.1 概述
Local模式就是運作在一台計算機上的模式，通常就是用于在本機上練手和測試。
它可以通過以下集中方式設定Master。
local：所有計算都運作在一個線程當中，沒有任何并行計算，通常我們在本機執行一些測試代碼，或者練手，就用這種模式；local[K]：指定使用幾個線程來運作計算，比如local[4]就是運作4個Worker線程。通常我們的Cpu有幾個Core，就指定幾個線程，最大化利用Cpu的計算能力；
local[*]：這種模式直接幫你按照Cpu最多Cores來設定線程數了。
2.3.2 安裝使用
1）上傳并解壓spark安裝包
[[email protected] sorfware]$ tar -zxvf spark-2.1.1-bin-hadoop2.7.tgz -C /opt/module/
[[email protected] module]$ mv spark-2.1.1-bin-hadoop2.7 spark
2）官方求PI案例
[[email protected] spark]$ bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--executor-memory 1G \
--total-executor-cores 2 \
./examples/jars/spark-examples_2.11-2.1.1.jar \
100
（1）基本文法
bin/spark-submit \
--class <main-class>
--master <master-url> \
--deploy-mode <deploy-mode> \
--conf <key>=<value> \
... # other options
<application-jar> \
[application-arguments]
（2）參數說明：
--master 指定Master的位址，預設為Local
--class: 你的應用的啟動類 (如 org.apache.spark.examples.SparkPi)
--deploy-mode: 是否釋出你的驅動到worker節點(cluster) 或者作為一個本地用戶端 (client) (default: client)*
--conf: 任意的Spark配置屬性， 格式key=value. 如果值包含空格，可以加引号“key=value”
application-jar: 打包好的應用jar,包含依賴. 這個URL在叢集中全局可見。 比如hdfs:// 共享存儲系統， 如果是 file:// path， 那麼所有的節點的path都包含同樣的jar
application-arguments: 傳給main()方法的參數
--executor-memory 1G 指定每個executor可用記憶體為1G
--total-executor-cores 2 指定每個executor使用的cup核數為2個
3）結果展示
該算法是利用蒙特·卡羅算法求PI
4）準備檔案
[[email protected] spark]$ mkdir input
在input下建立3個檔案1.txt和2.txt，并輸入以下内容
hello atguigu
hello spark
5）啟動spark-shell
[[email protected] spark]$ bin/spark-shell
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
19/09/14 22:26:49 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
19/09/14 22:26:58 WARN ObjectStore: Version information not found in metastore. hive.metastore.schema.verification is not enabled so recording the schema version 1.2.0
19/09/14 22:26:58 WARN ObjectStore: Failed to get database default, returning NoSuchObjectException
19/09/14 22:27:00 WARN ObjectStore: Failed to get database global_temp, returning NoSuchObjectException
Spark context Web UI available at http://192.168.1.102:4040
Spark context available as 'sc' (master = local[*], app id = local-1568471211466).
Spark session available as 'spark'.
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/___/ .__/\_,_/_/ /_/\_\ version 2.1.1
/_/
Using Scala version 2.11.8 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_144)
Type in expressions to have them evaluated.
Type :help for more information.
scala>
開啟另一個CRD視窗
[[email protected] spark]$ jps
3627 SparkSubmit
4047 Jps
可登入hadoop102:4040檢視程式運作
6)運作WordCount程式
先在spark安裝目錄下建立input檔案夾,添加多個文本檔案
scala>sc.textFile("input").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect
res0: Array[(String, Int)] = Array((hadoop,6), (oozie,3), (spark,3), (hive,3), (atguigu,3), (hbase,6))
scala>
可登入hadoop102:4040檢視程式運作
7）WordCount程式分析
送出任務分析：
資料流分析：
textFile("input")：讀取本地檔案input檔案夾資料；
flatMap(_.split(" "))：壓平操作，按照空格分割符将一行資料映射成一個個單詞；
map((_,1))：對每一個元素操作，将單詞映射為元組；
reduceByKey(_+_)：按照key将值進行聚合，相加；
collect：将資料收集到Driver端展示。
2.4 Standalone模式
2.4.1 概述
建構一個由Master+Slave構成的Spark叢集，Spark運作在叢集中。
2.4.2 安裝使用
1）進入spark安裝目錄下的conf檔案夾
[[email protected] module]$ cd spark/conf/
2）修改配置檔案名稱
[[email protected] conf]$ mv slaves.template slaves
[[email protected] conf]$ mv spark-env.sh.template spark-env.sh
3）修改slave檔案，添加work節點：
[[email protected] conf]$ vim slaves
hadoop102
hadoop103
hadoop104
4）修改spark-env.sh檔案，添加如下配置：
[[email protected] conf]$ vim spark-env.sh
SPARK_MASTER_HOST=hadoop102
SPARK_MASTER_PORT=7077
注意:需要注釋掉yarn
5）分發spark包
[[email protected] module]$ xsync spark/
6）啟動
[[email protected] spark]$ sbin/start-all.sh
[[email protected] spark]$ util.sh
[email protected]================
3330 Jps
3238 Worker
3163 Master
[email protected]================
2966 Jps
2908 Worker
[email protected]================
2978 Worker
3036 Jps
網頁檢視：hadoop102:8080
注意：如果遇到 “JAVA_HOME not set” 異常，可以在sbin目錄下的spark-config.sh(也可以在conf/spark-env.sh中配置) 檔案中加入如下配置：
export JAVA_HOME=XXXX(java安裝路徑)
7）官方求PI案例
[[email protected] spark]$ bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://hadoop102:7077 \
--executor-memory 1G \
--total-executor-cores 2 \
./examples/jars/spark-examples_2.11-2.1.1.jar \
100
8）啟動spark shell
/opt/module/spark/bin/spark-shell \
--master spark://hadoop102:7077 \
--executor-memory 1g \
--total-executor-cores 2
參數：--master spark://hadoop102:7077指定要連接配接的叢集的master
執行WordCount程式
scala>sc.textFile("input").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect
res0: Array[(String, Int)] = Array((hadoop,6), (oozie,3), (spark,3), (hive,3), (atguigu,3), (hbase,6))
scala>
2.4.3 JobHistoryServer配置
1）修改spark-default.conf.template名稱
[[email protected] conf]$ mv spark-defaults.conf.template spark-defaults.conf
2）修改spark-default.conf檔案，開啟Log：
[[email protected] conf]$ vi spark-defaults.conf
spark.eventLog.enabled true
spark.eventLog.dir hdfs://hadoop102:9000/directory
注意：HDFS上的目錄需要提前存在。
[[email protected] hadoop]$ hadoop fs –mkdir /directory
3）修改spark-env.sh檔案，添加如下配置：
[[email protected] conf]$ vi spark-env.sh
export SPARK_HISTORY_OPTS="-Dspark.history.ui.port=18080
-Dspark.history.retainedApplications=30
-Dspark.history.fs.logDirectory=hdfs://hadoop102:9000/directory"
參數描述：
spark.eventLog.dir：Application在運作過程中所有的資訊均記錄在該屬性指定的路徑下；
spark.history.ui.port=18080 WEBUI通路的端口号為18080
spark.history.fs.logDirectory=hdfs://hadoop102:9000/directory 配置了該屬性後，在start-history-server.sh時就無需再顯式的指定路徑，Spark History Server頁面隻展示該指定路徑下的資訊
spark.history.retainedApplications=30指定儲存Application曆史記錄的個數，如果超過這個值，舊的應用程式資訊将被删除，這個是記憶體中的應用數，而不是頁面上顯示的應用數。
4）分發配置檔案
[[email protected] conf]$ xsync spark-defaults.conf
[[email protected] conf]$ xsync spark-env.sh
5）啟動曆史服務
[[email protected] spark]$ sbin/start-history-server.sh
6）再次執行任務
[[email protected] spark]$ bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://hadoop102:7077 \
--executor-memory 1G \
--total-executor-cores 2 \
./examples/jars/spark-examples_2.11-2.1.1.jar \
100
7）檢視曆史服務
hadoop102:18080
2.4.4 HA配置
圖1 HA架構圖
1）zookeeper正常安裝并啟動
2）修改spark-env.sh檔案添加如下配置：
[[email protected] conf]$ vi spark-env.sh
注釋掉如下内容：
#SPARK_MASTER_HOST=hadoop102
#SPARK_MASTER_PORT=7077
添加上如下内容：
export SPARK_DAEMON_JAVA_OPTS="
-Dspark.deploy.recoveryMode=ZOOKEEPER
-Dspark.deploy.zookeeper.url=hadoop102,hadoop103,hadoop104
-Dspark.deploy.zookeeper.dir=/spark"
3）分發配置檔案
[[email protected] conf]$ xsync spark-env.sh
4）在hadoop102上啟動全部節點
[[email protected] spark]$ sbin/start-all.sh
5）在hadoop103上單獨啟動master節點
[atguigu@hadoop103 spark]$ sbin/start-master.sh
6）spark HA叢集通路
/opt/module/spark/bin/spark-shell \
--master spark://hadoop102:7077,hadoop103:7077 \
--executor-memory 2g \
--total-executor-cores 2
2.5 Yarn模式（重點）
2.5.1 概述
Spark用戶端直接連接配接Yarn，不需要額外建構Spark叢集。有yarn-client和yarn-cluster兩種模式，主要差別在于：Driver程式的運作節點。
yarn-client：Driver程式運作在用戶端，适用于互動、調試，希望立即看到app的輸出
yarn-cluster：Driver程式運作在由RM（ResourceManager）啟動的AP（APPMaster）适用于生産環境。
2.5.2 安裝使用
1）修改hadoop配置檔案yarn-site.xml,添加如下内容：
[[email protected] hadoop]$ vi yarn-site.xml
<!--是否啟動一個線程檢查每個任務正使用的實體記憶體量，如果任務超出配置設定值，則直接将其殺掉，預設是true -->
<property>
<name>yarn.nodemanager.pmem-check-enabled</name>
<value>false</value>
</property>
<!--是否啟動一個線程檢查每個任務正使用的虛拟記憶體量，如果任務超出配置設定值，則直接将其殺掉，預設是true -->
<property>
<name>yarn.nodemanager.vmem-check-enabled</name>
<value>false</value>
</property>
2）修改spark-env.sh，添加如下配置：
[[email protected] conf]$ vi spark-env.sh
YARN_CONF_DIR=/opt/module/hadoop-2.7.2/etc/hadoop
3）分發配置檔案
[[email protected] conf]$ xsync /opt/module/hadoop-2.7.2/etc/hadoop/yarn-site.xml
[[email protected] conf]$ xsync spark-env.sh
4）執行一個程式
[[email protected] spark]$ bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master yarn \
--deploy-mode client \
./examples/jars/spark-examples_2.11-2.1.1.jar \
100
注意：在送出任務之前需啟動(第一次要重新開機叢集)HDFS以及YARN叢集。
2.5.3 日志檢視
1）修改配置檔案spark-defaults.conf
添加如下内容：
spark.yarn.historyServer.address=hadoop102:18080
spark.history.ui.port=18080
2）重新開機spark曆史服務
[[email protected] spark]$ sbin/stop-history-server.sh
stopping org.apache.spark.deploy.history.HistoryServer
[[email protected] spark]$ sbin/start-history-server.sh
starting org.apache.spark.deploy.history.HistoryServer, logging to /opt/module/spark/logs/spark-atguigu-org.apache.spark.deploy.history.HistoryServer-1-hadoop102.out
3）送出任務到Yarn執行
[[email protected] spark]$ bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master yarn \
--deploy-mode client \
./examples/jars/spark-examples_2.11-2.1.1.jar \
100
4）Web頁面檢視日志
2.6 Mesos模式(了解)
Spark用戶端直接連接配接Mesos；不需要額外建構Spark叢集。國内應用比較少，更多的是運用yarn排程。
2.7 幾種模式對比
| 模式 | Spark安裝機器數 | 需啟動的程序 | 所屬者 |
| Local | 1 | 無 | Spark |
| Standalone | 3 | Master及Worker | Spark |
| Yarn | 1 | Yarn及HDFS | Hadoop |
第3章 案例實操
Spark Shell僅在測試和驗證我們的程式時使用的較多，在生産環境中，通常會在IDE中編制程式，然後打成jar包，然後送出到叢集，最常用的是建立一個Maven項目，利用Maven來管理jar包的依賴。
3.1 編寫WordCount程式
1）建立一個Maven項目WordCount并導入依賴
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.11</artifactId>
<version>2.1.1</version>
</dependency>
</dependencies>
<build>
<finalName>WordCount</finalName>
<plugins>
<plugin>
<groupId>net.alchim31.maven</groupId>
<artifactId>scala-maven-plugin</artifactId>
<version>3.2.2</version>
<executions>
<execution>
<goals>
<goal>compile</goal>
<goal>testCompile</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
2）編寫代碼
import org.apache.spark.{SparkConf, SparkContext}
object WordCount{
def main(args: Array[String]): Unit = {
//1.建立SparkConf并設定App名稱
val conf = new SparkConf().setAppName("WC")
//2.建立SparkContext，該對象是送出Spark App的入口
val sc = new SparkContext(conf)
//3.使用sc建立RDD并執行相應的transformation和action
sc.textFile(args(0)).flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_, 1).sortBy(_._2, false).saveAsTextFile(args(1))
//4.關閉連接配接
sc.stop()
}
}
3）打包插件
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
<version>3.0.0</version>
<configuration>
<archive>
<manifest>
<mainClass>WordCount</mainClass>
</manifest>
</archive>
<descriptorRefs>
<descriptorRef>jar-with-dependencies</descriptorRef>
</descriptorRefs>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
4）打包到叢集測試
bin/spark-submit \
--class WordCount \
--master spark://hadoop102:7077 \
WordCount.jar \
/word.txt \
/out
代碼筆記:
測試環境.
package wordcount
import org.apache.spark.{SparkConf, SparkContext}
/**
* @author cherry
* @create 2019-09-15-12:14
*/
object WordCount {
def main(args: Array[String]): Unit = {
/** 使用開發工具完成spark wordcount開發
* local模式
* 建立sparkConf
* 設定spark計算架構的運作(部署)環境
* appid
*/
val config = new SparkConf().setMaster("local[*]").setAppName("WordCount")
//建立spark上下文對象
val sc = new SparkContext(config)
println(sc)
}
}
控制台列印顯示
Wordcount程式
//讀取檔案,将檔案内容一行一行地讀取出來
val lines = sc.textFile("Spark\\sparkWordCount20190915\\src\\main\\scala\\files\\word.txt")
//逐行分解單詞
val words = lines.flatMap(_.split(" "))
//為了統計友善,将單詞資料進行結構的轉換
val wordToOne = words.map((_, 1))
//對轉換結構後的資料進行分組聚合
val wordToSum = wordToOne.reduceByKey(_ + _)
//将采集後的結果列印
val result: Array[(String, Int)] = wordToSum.collect()
//println(result)//直接列印結果顯示為[Lscala.Tuple2;@32f0c7f8
result.foreach(println)
列印結果
3.2 本地調試
本地Spark程式調試需要使用local送出模式，即将本機當做運作環境，Master和Worker都為本機。運作時直接加斷點調試即可。如下：
建立SparkConf的時候設定額外屬性，表明本地執行：
val conf = new SparkConf().setAppName("WC").setMaster("local[*]")
如果本機作業系統是windows，如果在程式中使用了hadoop相關的東西，比如寫入檔案到HDFS，則會遇到如下異常：
出現這個問題的原因，并不是程式的錯誤，而是用到了hadoop相關的服務，解決辦法是将附加裡面的hadoop-common-bin-2.7.3-x64.zip解壓到任意目錄。
在IDEA中配置Run Configuration，添加HADOOP_HOME變量