搭建Spark高可用集群

 

Spark簡介

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

Apache Spark™是用於大規模數據處理的統一分析引擎。

從右側最后一條新聞看，Spark也用於AI人工智能 spark是一個實現快速通用的集群計算平台。它是由加州大學伯克利分校AMP實驗室 開發的通用內存並行計算框架，用來構建大型的、低延遲的數據分析應用程序。它擴展了廣泛使用的MapReduce計算 模型。高效的支撐更多計算模式，包括交互式查詢和流處理。spark的一個主要特點是能夠在內存中進行計算，及時依賴磁盤進行復雜的運算，Spark依然比MapReduce更加高效。

為什么要學Spark

中間結果輸出：基於MapReduce的計算引擎通常會將中間結果輸出到磁盤上，進行存儲和容錯。出於任務管道承接的，考慮，當一些查詢翻譯到MapReduce任務時，往往會產生多個Stage，而這些串聯的Stage又依賴於底層文件系統（如HDFS）來存儲每一個Stage的輸出結果。 Spark是MapReduce的替代方案，而且兼容HDFS、Hive，可融入Hadoop的生態系統，以彌補MapReduce的不足。

Spark的四大特性

高效性

運行速度提高100倍。 Apache 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集群的工具。

Mesos：Spark可以運行在Mesos里面（Mesos 類似於yarn的一個資源調度框架） standalone：Spark自己可以給自己分配資源（master，worker） YARN：Spark可以運行在yarn上面 Kubernetes：Spark接收 Kubernetes的資源調度

Spark的組成

Spark組成(BDAS)：全稱伯克利數據分析棧，通過大規模集成算法、機器、人之間展現大數據應用的一個平台。也是處理大數據、雲計算、通信的技術解決方案。

它的主要組件有：

SparkCore：將分布式數據抽象為彈性分布式數據集（RDD），實現了應用任務調度、RPC、序列化和壓縮，並為運行在其上的上層組件提供API。

SparkSQL：Spark Sql 是Spark來操作結構化數據的程序包，可以讓我使用SQL語句的方式來查詢數據，Spark支持 多種數據源，包含Hive表，parquest以及JSON等內容。

SparkStreaming： 是Spark提供的實時數據進行流式計算的組件。

MLlib：提供常用機器學習算法的實現庫。

GraphX：提供一個分布式圖計算框架，能高效進行圖計算。

BlinkDB：用於在海量數據上進行交互式SQL的近似查詢引擎。

Tachyon：以內存為中心高容錯的的分布式文件系統。

應用場景

Yahoo將Spark用在Audience Expansion中的應用，進行點擊預測和即席查詢等 淘寶技術團隊使用了Spark來解決多次迭代的機器學習算法、高計算復雜度的算法等。應用於內容推薦、社區發現等 騰訊大數據精准推薦借助Spark快速迭代的優勢，實現了在“數據實時采集、算法實時訓練、系統實時預測”的全流程實時並行高維算法，最終成功應用於廣點通pCTR投放系統上。 優酷土豆將Spark應用於視頻推薦(圖計算)、廣告業務，主要實現機器學習、圖計算等迭代計算。

spark安裝

上傳spark-安裝包到Linux上

[root@hdp-01 ~]# wget https://d3kbcqa49mib13.cloudfront.net/spark-2.2.0-bin-hadoop2.7.tgz
[root@hdp-01 ~]# tar -zxvf spark-2.2.0-bin-hadoop2.7.tgz -C apps/
[root@hdp-01 apps]# mv spark-2.2.0-bin-hadoop2.7  spark

 

進入conf目錄並重命名並修改spark-env.sh.template文件

[root@hdp-01 conf]# mv spark-env.sh.template spark-env.sh
[root@hdp-01 conf]# vi spark-env.sh

 

在該配置文件中添加如下配置

export JAVA_HOME=/root/apps/java
export SPARK_MASTER_IP=hdp-01 export SPARK_MASTER_PORT=7077

 

重命名並修改slaves.template文件

[root@hdp-01 conf]# mv slaves.template slaves
[root@hdp-01 conf]# vi slaves

 

在該文件中添加子節點所在的位置（Worker節點）

hdp-02 hdp-03
hdp-04

 

將配置好的Spark拷貝到其他節點上

[root@hdp-01 apps]# scp -r spark/ hdp-02:$PWD
[root@hdp-01 apps]# scp -r spark/ hdp-03:$PWD
[root@hdp-01 apps]# scp -r spark/ hdp-04:$PWD

 

Spark集群配置完畢，目前是1個Master，3個Work，在hdp-01上啟動Spark集群

[root@hdp-01 spark]# sbin/start-all.sh

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

 

 

到此為止，Spark集群安裝完畢，但是有一個很大的問題，那就是Master節點存在單點故障，要解決此問題，就要借助zookeeper，並且啟動至少兩個Master節點來實現高可靠

 配置Spark集群之前一定要把zookper集群啟動

Spark集群規划：hdp-01，hdp-02是Master；hdp-03，hdp-04是Worker 安裝配置zk集群，並啟動zk集群 停止spark所有服務

集群中Master地址是不固定的 所以必須把export SPARK_MASTER_IP=hdp-01 注釋掉

[root@hdp-01 sbin]# ./stop-all.sh
[root@hdp-01 conf]# vi spark-env.sh export JAVA_HOME=/root/apps/java #export SPARK_MASTER_IP=hdp-01 export SPARK_MASTER_PORT=7077 export SPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=ZOOKEEPER -Dspark.deploy.zookeeper.url=hdp-01:2181,hdp-02:2181,hdp-03:2181 -Dspark.deploy.zookeeper.dir=/spark"  [root@hdp-01 conf]#scp -r spark-env.sh/ hdp-02:$PWD [root@hdp-01 conf]#scp -r spark-env.sh/ hdp-03:$PWD [root@hdp-01 conf]#scp -r spark-env.sh/ hdp-04:$PWD [root@hdp-01 sbin]# ./start-all.sh
// 啟動hdp-02上的master，如果hdp-01宕機 hdp-02會有worker變為master
[root@hdp-02 spark]# sbin/start-master.sh 

測試集群部署

打開瀏覽器  輸入        http://hdp-01:8080/  http://hdp-02:8080/

 

 

 

 standby:備用狀態

 現在停掉hdp-01上的Master

[root@hdp-01 sbin]# jps
35473 Master 1682 NameNode 15109 NodeManager 35575 Jps 2200 SecondaryNameNode 14984 ResourceManager 14636 QuorumPeerMain 1839 DataNode [root@hdp-01 sbin]# kill -9 35473

 

 hdp-02接替hdp-01工作 從Master變為Worker  如果現在再啟動 hdp-01 hdp-01為從節點 不會變為主節點 

 如果hdp-02發生故障后 hdp-01就會自動變成Master

 

Spark Shell

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

啟動spark shell

[root@hdp-01 spark]# bin/spark-shell --master spark:/hdp-01:7077 --executor-memory 1g --total-executor-cores 1

 

 

 出現此logo證明spark shell 啟動成功

集群模式下啟動spark shell

[root@hdp-01 spark]# bin/spark-shell --master spark://hdp-01:7077,hdp-04:7077

 

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

Spark控制台有一個Running Applications 也就是剛剛啟動的spark shell

 

 執行Spark程序

Spark程序求圓周率

[root@hdp-01 spark]# bin/spark-submit --master spark://hdp-01:7077 --class org.apache.spark.examples.SparkPi  --executor-memory 1G --total-executor-cores 1 examples/jars/spark-examples_2.11-2.2.0.jar 100

[root@hdp-01 spark]# bin/spark-submit --class org.apache.spark.examples.SparkPi --master spark://hdp-01:7077 examples/jars/spark-examples_2.11-2.2.0.jar 100

參數說明： --master spark://hdp-01:7077 指定Master的地址 --executor-memory 1G 指定每個worker可用內存為1G --total-executor-cores 1 指定整個集群使用的cup核數為1

 

 集群模式執行spark程序

[root@hdp-01 spark]# bin/spark-submit --class org.apache.spark.examples.SparkPi --master spark://hdp-01:7077,hdp-04:7077 --executor-memory 1G --total-executor-cores 2 examples/jars/spark-examples_2.11-2.2.0.jar 100

 

spark shell中編寫WordCount程序

啟動hdfs 上傳單詞文件到hdfs中

[root@hdp-01 ~]# start-all.sh
[root@hdp-01 ~]# vi spark.txt helo java hello spark hello hdfs hello yarn yarn hdfs [root@hdp-01 ~]# hadoop fs -mkdir -p /spark [root@hdp-01 ~]# hadoop fs -put spark.txt /spark

 

spark shell 執行任務

scala> sc.textFile("hdfs://hdp-01:9000/spark").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect
res0: Array[(String, Int)] = Array(("",4), (hello,4), (java,1), (yarn,2), (spark,1), (hdfs,2))

這樣就統計出每個單詞的數量以key-value的形式在控制台顯示 以下將結果輸出到hdfs中

scala> sc.textFile("hdfs://hdp-01:9000/spark").flatMap(.split(" ")).map((,1)).reduceByKey(+).saveAsTextFile("hdfs://hdp-01:9000/outText")