Spark基礎理論及安裝
一、初始Spark
1、產生背景
由於mapreduce只適用於離線資料處理(批處理),而不能支援對迭代式,互動式,流式資料的處理,因此,spark就誕生了
hive的產生就是為了解決mapreduce的程式設計複雜的問題
spark的產生就是為了解決mapreduce的計算緩慢的問題
Spark和MapReduce的對比之後的優勢:
1、減少磁碟IO
MapReduce:基於磁碟 Spark:基於記憶體 (儘量把臨時資料快取在記憶體中)
2、增加並行度
MapReduce: MapTask ReduceTask JVM 一個程序一個Task 預設情況下
Spark : ShuffleMapTask ResultTask 使用的是一個執行緒執行
3、避免重複計算
可以把資料強制的持久化到記憶體中,以供其他的task使用
4、可選的shuffle和排序
可選的shuffle : MapReduce:提供的是一種通用的shuffle combiner partitioner sorter
Spark:提供四種shuffle策略,分別有對應不同的需求場景 HashShuffle SortShuffle
可選的排序 : mapreducer如果有Reduce階段的話,那麼sorter的資料排序一定會執行
5、靈活的記憶體管理策略
要多少給多少。可以合理的分配到底哪個階段,哪個元件,使用多少
2、Spark的概念 (一棧式的通用計算框架)
Spark是一快速、通用、可擴充套件的大資料分析引擎
2009年誕生於加州大學伯克利分校
3、Hadoop生態和Spark生態的區別
Hadoop生態:
hdfs+zookeeper+mapreduce/hive+hbase+storm+mahout+其他工具
Spark生態體系:
hdfs+zookeeper+spark+hbase+storm+mahout+其他工具
spark的產生就是為了替代Mapreduce
4、特點
1)Speed:快速高效
2)Ease of Use:簡單易用
3)Generality:全棧式資料處理
- 支援批處理(Spark Core):Spark Core 是 Spark 的核心功能實現,包括:SparkContext 的初始化(DriverApplication 通過 SparkContext 提交)、部署模式、儲存體系、任務提交與執行、計算引擎等
- 支援互動式查詢(Spark SQL)。Spark SQL 是 Spark 來操作結構化資料的程式包,可以讓我們使用 SQL 語句的方式來查詢資料,Spark 支援多種資料來源,包含 Hive 表,parquet 以及 JSON等內容
- 支援流式計算(Spark Streaming)。與 MapReduce 只能處理離線資料相比,Spark 還支援實時的流計算。Spark 依賴 Spark Streaming 對資料進行實時的處理。
- 支援機器學習(Spark MLlib)。提供機器學習相關的統計、分類、迴歸等領域的多種演算法實現。其一致的 API 介面大大降低了使用者的學習成本
- 支援圖計算(Spark GraghX)。提供圖計算處理能力,支援分散式, Pregel 提供的 API 可以解決圖計算中的常見
4)Run EveryWhere:相容
二、Spark叢集的安裝
1、叢集規劃
server | Master | Worker |
qyl01 | √ | √ |
qyl02 | √ | √ |
qyl03 | √ |
2、具體步驟
1、上傳下載好的 Spark 到叢集中的一個節點,比如是 qyl01
put c:/spark-2.3.0-bin-hadoop2.7.tgz
2、使用之前安裝 hadoop 叢集相同的 hadoop 使用者安裝 spark 叢集,現在規劃安裝目錄/home/qyl/apps/,解壓縮排行安裝
tar -zxvf spark-2.3.0-bin-hadoop2.7.tgz -apps /home/qyl/apps/
3、修改配置檔案 spark-env.sh
進入 SPARK_HOME 的 conf 目錄中,進行如下更改:
cd /home/qyl/apps/spark-2.3.0-bin-hadoop2.7/conf
mv spark-env.sh.template spark-env.sh
然後修改 spark-env.sh:
export JAVA_HOME=/usr/local/java/jdk1.8.0_73
export SPARK_MASTER_HOST=qyl01
export SPARK_MASTER_PORT=7077
4、修改配置檔案 slave
修改配置檔案 slave
進入 SPARK_HOME 的 conf 目錄中,進行如下更改:
cd /home/qyl/apps/spark-2.3.0-bin-hadoop2.7/conf
mv slaves.template slaves
在 slaves 的最後新增所有 worker 節點的主機名
qyl01
qyl02
qyl03
5、將spark安裝包copy到所有安裝節點
scp -r spark-2.3.0-bin-hadoop2.7 qyl02:/home/qyl/apps/
scp -r spark-2.3.0-bin-hadoop2.7 qyl03:/home/qyl/apps/
6、修改配置檔案
vim /etc/profile
export SPARK_HOME=/home/qyl/apps/spark-2.3.0-bin-hadoop2.7
export PATH=$PATH:$SPARK_HOME/bin:$SPARK_HOME/sbin
source /etc/profile
7、配置zookeeper叢集,並且啟動zookeeper叢集
8、修改SPARK_HOME/conf目錄中的spark-env.sh配置檔案
刪掉:
export SPARK_MASTER_HOST=qyk01
增加一行:
export SPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=ZOOKEEPER -
Dspark.deploy.zookeeper.url=qyl01,qyl02,qyl03 -
Dspark.deploy.zookeeper.dir=/spark"
9、把core-site.xml和hdfs-site.xml放置在$SPARK_HOME/conf目錄下,然後所有節點要同步
10、同步配置檔案
scp -r spark-env.sh qyl02:$PWD
scp -r spark-env.sh qyl03:$PWD
11、啟動叢集
[[email protected] spark-2.3.0-bin-hadoop2.7]$ sbin/start-all.sh
三、核心概念
1、Master
就如同名字一樣,是spark叢集的管理者,或者管理節點
在Spark基於Yarn的方式執行過程中,master就是resourcemanager,基於自身的standalone模式執行就是叢集中檢視到的master節點
2、Work
是spark叢集的工作節點,在spark基於Yarn的方式執行過程中,work就是nodemanager,基於自身的standalone模式執行就是叢集中檢視到的work節點
作用:調動本臺機器上面的資源
3、Application
提交到spark叢集中取執行的完整應用就是一個Application,當然一個Application由多個job組成
4、Job
每提交到比較會提交一個Spark的job作業,或者說,每觸發一次action操作,就會提交一次spark作業job,從這個角度而言,一個application又多個Job組成,這裡的job,就可以理解MapReduce中的Job
5、SparkContext
整個一個Application的應用上下文物件
一個spark應用的入口就是sparkcontext,是spark作業最要的概念
6、Driver
驅動,簡單理解,就是用來提交程式碼,提交job作業
再換句話說,jobs是在叢集中執行的,而這些dirver是在本地提交這些job作業的
7、RDD:彈性分散式資料集(resilient distributed dataset)
RDD就能夠被平行計算的,被分割槽儲存在叢集中各個節點的資料集
RDD其實不儲存真是的資料,只儲存資料的獲取的方法,以及分割槽的方法,還有就是資料的型別。
分散式:資料儲存在rdd對應的各個partition分割槽中
彈性:spark的資料在記憶體中儲存或者計算,當記憶體儲存不了的時候,對應的資料外溢到磁碟進行儲存和計算
RDD的建立方式:可以hdfs檔案建立,還可以通過並行對scala的集合建立,在編碼過程中還可以通過轉化,將一個rdd轉化為其他的rdd
四、SparkRDD執行架構圖
以wordcount為例:
在spark-shell中執行
sc.textFile("hdfs://myha01/spark/word.txt")flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).foreach(println)
五、建立Spark作業
1、maven專案
2、匯入依賴 在pom.xml中新增
<properties>
<scala.version>2.11.8<scala.version>
<spark.version>2.3.2<spark.version>
<hadoop.version>2.7.6<hadoop.version>
<properties>
<dependency>
<groupid>org.apache.spark</groupid>
<artifactId>spark-core_2.11</artifactId>
<version>${spark.version}</version>
</dependency>
<dependency>
<groupid>org.apache.hadoop</groupid>
<artifactId>hadoop-client</artifactId>
<version>${hadoop.version}</version>
</dependency>