學習一個工具的最好途徑,就是使用它。在IT領域,在深入瞭解一個系統的原理、實現細節之前,應當先準備好它的運行環境或者源碼閱讀環境。如果能在實際環境下安裝和運行Spark,顯然能夠提升讀者對於Spark的一些感受,對系統能有個大體的印象,有經驗的工程師甚至能夠猜出一些Spark在實現過程中採用的設計模... ...
學習一個工具的最好途徑,就是使用它。這就好比《極品飛車》玩得好的同學,未必真的會開車,要學習車的駕駛技能,就必須用手觸摸方向盤、用腳感受剎車與油門的力道。在IT領域,在深入瞭解一個系統的原理、實現細節之前,應當先準備好它的運行環境或者源碼閱讀環境。如果能在實際環境下安裝和運行Spark,顯然能夠提升讀者對於Spark的一些感受,對系統能有個大體的印象,有經驗的工程師甚至能夠猜出一些Spark在實現過程中採用的設計模式、編程模型。
考慮到大部分公司在開發和生產環境都採用Linux操作系統,所以筆者選用了64位的Linux。在正式安裝Spark之前,先要找台好機器。為什麼?因為筆者在安裝、編譯、調試的過程中發現Spark非常耗費記憶體,如果機器配置太低,恐怕會跑不起來。Spark的開發語言是Scala,而Scala需要運行在JVM之上,因而搭建Spark的運行環境應該包括JDK和Scala。
本文只介紹最基本的與Spark相關的準備工作,至於Spark在實際生產環境下的配置,則需要結合具體的應用場景進行準備。
安裝JDK
自Spark2.0.0版本開始,Spark已經準備放棄對Java 7的支持,所以我們需要選擇Java 8。我們還需要使用命令getconf LONG_BIT查看linux機器是32位還是64位,然後下載相應版本的JDK並安裝。
下載地址:
http://www.oracle.com/technetwork/java/javase/downloads/index.html。
配置環境:
cd ~ vim .bash_profile
添加如下配置:
exportJAVA_HOME=/opt/java exportPATH=$PATH:$JAVA_HOME/bin exportCLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
輸入以下命令使環境變數快速生效:
source .bash_profile
安裝完畢後,使用java –version命令查看,確認安裝正常,如圖1所示。
圖1 查看java安裝是否正常
安裝Scala
由於從Spark 2.0.0開始,Spark預設使用Scala 2.11來編譯、打包,不再是以前的Scala 2.10,所以我們需要下載Scala 2.11。
下載地址:
http://www.scala-lang.org/download/
選擇Scala 2.11的版本進行下載,下載方法如下:
wget https://downloads.lightbend.com/scala/2.11.8/scala-2.11.8.tgz
移動到選好的安裝目錄,例如:
mv scala-2.11.8.tgz~/install/
進入安裝目錄,執行以下命令:
chmod 755scala-2.11.8.tgz tar -xzvfscala-2.11.8.tgz
配置環境:
cd ~ vim .bash_profile
添加如下配置:
export SCALA_HOME=$HOME/install/scala-2.11.8 export PATH=$SCALA_HOME/bin:$PATH
輸入以下命令使環境變數快速生效:
source .bash_profile
安裝完畢後鍵入scala,進入scala命令行以確認安裝正常,如圖2所示。
圖2 進入Scala命令行
安裝Spark
Spark進入2.0時代之後,目前一共有兩個大的版本:一個是2.0.0,一個是2.1.0。本書選擇2.1.0。
下載地址:
http://spark.apache.org/downloads.html
下載方法如下:
wget http://d3kbcqa49mib13.cloudfront.net/spark-2.1.0-bin-hadoop2.6.tgz
移動到選好的安裝目錄,如:
mv spark-2.1.0-bin-hadoop2.6.tgz~/install/
進入安裝目錄,執行以下命令:
chmod 755 spark-2.1.0-bin-hadoop2.6.tgz tar -xzvf spark-2.1.0-bin-hadoop2.6.tgz
配置環境:
cd ~ vim .bash_profile
添加如下配置:
export SPARK_HOME=$HOME/install/spark-2.1.0-bin-hadoop2.6 export PATH=$SPARK_HOME/bin:$PATH
輸入以下命令使環境變數快速生效:
source .bash_profile
安裝完畢後鍵入spark-shell,進入scala命令行以確認安裝正常,如圖3所示。
圖3 執行spark-shell進入Scala命令行
既然已經介紹瞭如何準備好基本的Spark運行環境,現在是時候實踐一下,以便於在使用過程中提升讀者對於Spark最直接的感觸!本文通過Spark的基本使用,讓讀者對Spark能有初步的認識,便於引導讀者逐步深入學習。
運行spark-shell
在《Spark2.1.0——運行環境準備》一文曾經簡單運行了spark-shell,並用下圖進行了展示(此處再次展示此圖)。
圖4 執行spark-shell進入Scala命令行
圖4中顯示了很多信息,這裡進行一些說明:
- 在安裝完Spark 2.1.0後,如果沒有明確指定log4j的配置,那麼Spark會使用core模塊的org/apache/spark/目錄下的log4j-defaults.properties作為log4j的預設配置。log4j-defaults.properties指定的Spark日誌級別為WARN。用戶可以到Spark安裝目錄的conf文件夾下從log4j.properties.template複製一份log4j.properties文件,併在其中增加自己想要的配置。
- 除了指定log4j.properties文件外,還可以在spark-shell命令行中通過sc.setLogLevel(newLevel)語句指定日誌級別。
- SparkContext的Web UI的地址是:http://192.168.0.106:4040。192.168.0.106是筆者安裝Spark的機器的ip地址,4040是SparkContext的Web UI的預設監聽埠。
- 指定的部署模式(即master)為local[*]。當前應用(Application)的ID為local-1497084620457。
- 可以在spark-shell命令行通過sc使用SparkContext,通過spark使用SparkSession。sc和spark實際分別是SparkContext和SparkSession在Spark REPL中的變數名,具體細節已在《Spark2.1.0——剖析spark-shell》一文有過分析。
由於Spark core的預設日誌級別是WARN,所以看到的信息不是很多。現在我們將Spark安裝目錄的conf文件夾下的log4j.properties.template以如下命令複製出一份:
cp log4j.properties.template log4j.properties
並將log4j.properties中的log4j.logger.org.apache.spark.repl.Main=WARN修改為log4j.logger.org.apache.spark.repl.Main=INFO,然後我們再次運行spark-shell,將列印出更豐富的信息,如圖5所示。
圖5 Spark啟動過程列印的部分信息
從圖5展示的啟動日誌中我們可以看到SecurityManager、SparkEnv、BlockManagerMasterEndpoint、DiskBlockManager、MemoryStore、SparkUI、Executor、NettyBlockTransferService、BlockManager、BlockManagerMaster等信息。它們是做什麼的?剛剛接觸Spark的讀者只需要知道這些信息即可,具體內容將在後邊的博文給出。
執行word count
這一節,我們通過word count這個耳熟能詳的例子來感受下Spark任務的執行過程。啟動spark-shell後,會打開Scala命令行,然後按照以下步驟輸入腳本:
步驟1
輸入val lines =sc.textFile("../README.md", 2),以Spark安裝目錄下的README.md文件的內容作為word count例子的數據源,執行結果如圖6所示。
圖6 步驟1執行結果
圖6告訴我們lines的實際類型是MapPartitionsRDD。
步驟2
textFile方法對文本文件是逐行讀取的,我們需要輸入val words =lines.flatMap(line => line.split(" ")),將每行文本按照空格分隔以得到每個單詞,執行結果如圖7所示。圖7 步驟2執行結果
圖7告訴我們lines在經過flatMap方法的轉換後得到的words的實際類型也是MapPartitionsRDD。
步驟3
對於得到的每個單詞,通過輸入val ones = words.map(w => (w,1)),將每個單詞的計數初始化為1,執行結果如圖8所示。圖8 步驟3執行結果
圖8告訴我們words在經過map方法的轉換後得到的ones的實際類型也是MapPartitionsRDD。
步驟4
輸入val counts = ones.reduceByKey(_ + _),對單詞進行計數值的聚合,執行結果如圖9所示。
圖9 步驟4執行結果
圖9告訴我們ones在經過reduceByKey方法的轉換後得到的counts的實際類型是ShuffledRDD。
步驟5
輸入counts.foreach(println),將每個單詞的計數值列印出來,作業的執行過程如圖10和圖11所示。作業的輸出結果如圖12所示。
圖10 步驟5執行過程第一部分
圖11 步驟5執行過程第二部分
圖10和圖11展示了很多作業提交、執行的信息,這裡挑選關鍵的內容進行介紹:
- SparkContext為提交的Job生成的ID是0。
- 一共有四個RDD,被劃分為ResultStage和ShuffleMapStage。ShuffleMapStage的ID為0,嘗試號為0。ResultStage的ID為1,嘗試號也為0。在Spark中,如果Stage沒有執行完成,就會進行多次重試。Stage無論是首次執行還是重試都被視為是一次Stage嘗試(Stage Attempt),每次Attempt都有一個唯一的嘗試號(AttemptNumber)。
- 由於Job有兩個分區,所以ShuffleMapStage和ResultStage都有兩個Task被提交。每個Task也會有多次嘗試,因而也有屬於Task的嘗試號。從圖中看出ShuffleMapStage中的兩個Task和ResultStage中的兩個Task的嘗試號也都是0。
- HadoopRDD則用於讀取文件內容。
圖12 步驟5輸出結果
圖12展示了單詞計數的輸出結果和最後列印的任務結束的日誌信息。
本文介紹的word count例子是以SparkContext的API來實現的,讀者朋友們也可以選擇在spark-shell中通過運用SparkSession的API來實現。
有了對Spark的初次體驗,下麵可以來分析下spark-shell的實現原理了,請看——《Spark2.1.0——剖析spark-shell》
想要對Spark源碼進行閱讀的同學,可以看看《Spark2.1.0——代碼結構及載入Ecplise方法》