摘要：數據科學任務主要是數據分析領域，數據科學家要負責分析數據并建模，具備統計預測建模機器學習等方面的經驗，以及一定的使用或語言進行編程的能力。監控運行時性能指標信息。

官網：http://spark.apache.org

Spark是一種快速、通用、可擴展的大數據分析引擎，2009年誕生于加州大學伯克利分校AMPLab，2010年開源，2013年6月成為Apache孵化項目，2014年2月成為Apache頂級項目。目前，Spark生態系統已經發展成為一個包含多個子項目的集合，其中包含SparkSQL、Spark Streaming、GraphX、MLlib等子項目，Spark是基于內存計算的大數據并行計算框架。Spark基于內存計算，提高了在大數據環境下數據處理的實時性，同時保證了高容錯性和高可伸縮性，允許用戶將Spark部署在大量廉價硬件之上，形成集群。

Spark是一個計算框架,而Hadoop中包含計算框架MapReduce和分布式文件系統HDFS,Hadoop更廣泛地說還包括在其生態系統上的其他系統.

Hadoop的MapReduce計算模型存在問題:
Hadoop的MapReduce的核心是Shuffle(洗牌).在整個Shuffle的過程中,至少產生6次I/O流.基于MapReduce計算引擎通常會將結果輸出到次盤上,進行存儲和容錯.另外,當一些查詢(如:hive)翻譯到MapReduce任務是,往往會產生多個Stage,而這些Stage有依賴底層文件系統來存儲每一個Stage的輸出結果,而I/O的效率往往較低,從而影響MapReduce的運行速度.

快

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

易用

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

通用

Spark提供了統一的解決方案。Spark可以用于批處理、交互式查詢（Spark SQL）、實時流處理（Spark Streaming）、機器學習（Spark MLlib）和圖計算（GraphX）。這些不同類型的處理都可以在同一個應用中無縫使用。Spark統一的解決方案非常具有吸引力，畢竟任何公司都想用統一的平臺去處理遇到的問題，減少開發和維護的人力成本和部署平臺的物力成本。

兼容性

Spark 可以非常方便地與其他的開源產品進行融合。比如，Spark 可以使用Hadoop 的 YARN 和 Apache Mesos 作為它的資源管理和調度器.并且可以處理所有 Hadoop 支持的數據，包括 HDFS、HBase 和 Cassandra 等。這對于已經部署Hadoop 集群的用戶特別重要，因為不需要做任何數據遷移就可以使用 Spark 的強大處理能力。Spark 也可以不依賴于第三方的資源管理和調度器，它實現了Standalone 作為其內置的資源管理和調度框架，這樣進一步降低了 Spark 的使用門檻，使得所有人都可以非常容易地部署和使用 Spark。此外，Spark 還提供了在EC2 上部Standalone 的 Spark 集群的工具。

Spark Core：

實現了 Spark 的基本功能，包含任務調度、內存管理、錯誤恢復、與存儲系統 交互等模塊。Spark Core 中還包含了對彈性分布式數據集(resilient distributed dataset，簡稱RDD)的 API 定義。

Spark Streaming:

Spark Streaming基于微批量方式的計算和處理,可以用于處理實時的流數據.它使用DStream,簡單來說是一個彈性分布式數據集(RDD)系列,處理實時數據.數據可以從Kafka,Flume,Kinesis或TCP套接字等眾多來源獲取,并且可以使用由高級函數（如 map，reduce，join 和 window）開發的復雜算法進行流數據處理。最后，處理后的數據可以被推送到文件系統，數據庫和實時儀表板。

Spark SQL

SPark SQL可以通過JDBC API將Spark數據集暴露出去,而且還可以用傳統的BI和可視化工具在Spark數據上執行類似SQL的查詢,用戶哈可以用Spark SQL對不同格式的數據(如Json, Parque以及數據庫等)執行ETl,將其轉化,然后暴露特定的查詢.

Spark MLlib

MLlib是一個可擴展的Spark機器學習庫，由通用的學習算法和工具組成，包括二元分類、線性回歸、聚類、協同過濾、梯度下降以及底層優化原語。

Spark Graphx:

GraphX是用于圖計算和并行圖計算的新的（alpha）Spark API。通過引入彈性分布式屬性圖（Resilient Distributed Property Graph），一種頂點和邊都帶有屬性的有向多重圖，擴展了Spark RDD。為了支持圖計算，GraphX暴露了一個基礎操作符集合（如subgraph，joinVertices和aggregateMessages）和一個經過優化的Pregel API變體。此外，GraphX還包括一個持續增長的用于簡化圖分析任務的圖算法和構建器集合。

集群管理器：

Spark 設計為可以高效地在一個計算節點到數千個計算節點之間伸縮計 算。為了實現這樣的要求，同時獲得最大靈活性，Spark 支持在各種集群管理器(cluster manager)上運行，包括 Hadoop YARN、Apache Mesos，以及 Spark 自帶的一個簡易調度 器，叫作獨立調度器。

Spark得到了眾多大數據公司的支持，這些公司包括Hortonworks、IBM、Intel、Cloudera、MapR、Pivotal、百度、阿里、騰訊、京東、攜程、優酷土豆。當前百度的Spark已應用于鳳巢、大搜索、直達號、百度大數據等業務；阿里利用GraphX構建了大規模的圖計算和圖挖掘系統，實現了很多生產系統的推薦算法；騰訊Spark集群達到8000臺的規模，是當前已知的世界上最大的Spark集群。

我們大致把Spark的用例分為兩類：數據科學應用和數據處理應用。也就對應的有兩種人群：數據科學家和工程師。

數據科學任務

主要是數據分析領域，數據科學家要負責分析數據并建模，具備 SQL、統計、預測建模(機器學習)等方面的經驗，以及一定的使用 Python、 Matlab 或 R 語言進行編程的能力。

數據處理應用

工程師定義為使用 Spark 開發 生產環境中的數據處理應用的軟件開發者，通過對接Spark的API實現對處理的處理和轉換等任務。

Driver:運行Application的main() 函數并創建SparkContext

Worker:從節點,負責控制計算節點,啟動Ex而粗投入或Driver

SparkContext: 整個應用的上下文,監控應用的生命周期

SparkConf：負責存儲配置信息。

Executor: 執行器,在worker node上執行任務組件,用于啟動線程執行任務.每個Application擁有獨立的一組Executors

ClusterManager:在standlone模式中即為Master(主節點),控制整個集群.監控Worker.在Yarn模式中為資源管理器.

RDD:彈性分布式集合,spark的基本計算單元，一組RDD可形成執行的有向無環圖RDD Graph

DAG Scheduler: 根據作業(Job)構建基于Stage的DAG,并交給Stage給TaskScheduler

TaskScheduler：將任務（Task）分發給Executor執行

SparkEnv：線程級別的上下文，存儲運行時的重要組件的引用。SparkEnv內創建并包含如下一些重要組件的引用。

MapOutPutTracker：負責Shuffle元信息的存儲。

BroadcastManager：負責廣播變量的控制與元信息的存儲。

BlockManager：負責存儲管理、創建和查找塊。

MetricsSystem：監控運行時性能指標信息。

Spark的整體流程:client提交應用,Master找到一個Worker啟動Driver,Driver向Master或者向資源管理器申請資源,之后將應用轉化為RDD Graph，再由DAGScheduler將RDD Graph轉化為Stage的有向無環圖提交給TaskScheduler，由TaskScheduler提交任務給Executor執行。在任務執行的過程中，其他組件協同工作，確保整個應用順利執行。

Spark的部署模式有Local、Local-Cluster、Standalone、Yarn、Mesos，我們選擇最具代表性的Standalone集群部署模式。安裝java環境,Spark自動會把scala SDK打包到Spark中無需安裝scala環境

linux: CentOS-7.5_x64
hadoop: hadoop-3.2.0
spark: spark-2.3.3
zookeeper: zookeeper-3.4.10

# 下載對應的Spark安裝包
$ wget http://mirrors.hust.edu.cn/apache/spark/spark-2.3.3/spark-2.3.3-bin-hadoop2.7.tgz

# 解壓縮
$ tar -zxvf spark-2.3.3-bin-hadoop2.7.tgz

# 進入spark解壓目錄
$ cd $SPARK_HOME

# 修改Spark的環境配置文件
$ cp conf/spark-env.sh.template spark-env.sh
$ vim conf/spark-env.sh

# 添加如下配置
export JAVA_HOME=/usr/java/jdk1.8.0_191

# 修改slave的配置
$ cp $SPARK_HOME/conf/slaves.template slaves
$ vi slaves

# 在該文件中添加子節點所在的位置（Worker節點）
node-2
node-3
node-4

# 將配置好的spark 復制到其他機器上(node-2,node-3,node-4)
$ scp -r spark-2.3.2-bin-hadoop2.7 root@node-2:/xxx/xxx

# 啟動spark集群
$ sbin/start-master.sh
$ sbin/start-slaves.sh

# 也可以是用這個腳本啟動所有機器
$ sbin/start-all.sh

啟動后執行jps命令，主節點上有Master進程，其他子節點上有Work進行，登錄Spark管理界面查看集群狀態（主節點）：http://node-1:8080/

1.安裝配置zk集群，并啟動zk集群zookeeper安裝

2.修改spark的配置文件添加如下配置

export SPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=ZOOKEEPER -Dspark.deploy.zookeeper.url=node-2:2181,node-3:2181,node-4:2181 -Dspark.deploy.zookeeper.dir=/spark"

3.修改所有節點的slaves文件改為（node-3,node-4）節點

4.在node1上執行 sbin/start-all.sh，然后在 node-2 上啟動第二個 Master（sbin/start-master.sh ）

$SPARK_HOME/bin/spark-submit --class org.apache.spark.examples.SparkPi --master spark://master-ip:7077 --executor-memory 1G --total-executor-cores 2 $SPARK_HOME/examples/jars/spark-examples_2.11-2.3.3.jar 100

spark-shell是Spark自帶的交互式Shell程序，方便用戶進行交互式編程，用戶可以在該命令行下用scala編寫spark程序。

$SPARK_HOME/bin/spark-shell --master spark://node-1:7077 --executor-memory 2g --total-executor-cores 2

參數說明：

# 指定Master的地址
--master spark://node-1:7077

# 指定每個worker可用內存為2G
--executor-memory 2g

# 指定整個集群使用的cup核數為2個
--total-executor-cores 2

注意

如果啟動spark shell時沒有指定master地址，但是也可以正常啟動spark shell和執行spark
shell中的程序，其實是啟動了spark的local模式，該模式僅在本機啟動一個進程，沒有與集群建立聯系。
Spark Shell中已經默認將SparkContext類初始化為對象sc。用戶代碼如果需要用到，則直接應用sc即可

在spark shell中用scala語言編寫spark程序

# sc是SparkContext對象，該對象時提交spark程序的入口
sc.textFile("file:///root/data/words.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).saveAsTextFile("file:///root/data/output1")

# 從本地文件系統中讀取數據
textFile("file:///root/data/words.txt")

# 讀取每一行數據并切分
flatMap(_.split(" "))

# 將數據切分映射將單詞和1構成元組
map((_,1))

# 按照key進行reduce，并將value累加
reduceByKey(_+_)

# 將結果寫入到指定位置
saveAsTextFile("file:///root/data/output1")