Spark2.1.0怎么用

本篇內(nèi)容介紹了“Spark2.1.0怎么用”的有關(guān)知識(shí)，在實(shí)際案例的操作過(guò)程中，不少人都會(huì)遇到這樣的困境，接下來(lái)就讓小編帶領(lǐng)大家學(xué)習(xí)一下如何處理這些情況吧！希望大家仔細(xì)閱讀，能夠?qū)W有所成！

運(yùn)行spark-shell

    在《Spark2.1.0之運(yùn)行環(huán)境準(zhǔn)備》一文曾經(jīng)簡(jiǎn)單運(yùn)行了spark-shell，并用下圖進(jìn)行了展示（此處再次展示此圖）。

圖1    執(zhí)行spark-shell進(jìn)入Scala命令行

圖1中顯示了很多信息，這里進(jìn)行一些說(shuō)明：

• 在安裝完Spark 2.1.0后，如果沒(méi)有明確指定log4j的配置，那么Spark會(huì)使用core模塊的org/apache/spark/目錄下的log4j-defaults.properties作為log4j的默認(rèn)配置。log4j-defaults.properties指定的Spark日志級(jí)別為WARN。用戶可以到Spark安裝目錄的conf文件夾下從log4j.properties.template復(fù)制一份log4j.properties文件，并在其中增加自己想要的配置。

• 除了指定log4j.properties文件外，還可以在spark-shell命令行中通過(guò)sc.setLogLevel(newLevel)語(yǔ)句指定日志級(jí)別。

• SparkContext的Web UI的地址是：http://192.168.0.106:4040。192.168.0.106是筆者安裝Spark的機(jī)器的ip地址，4040是SparkContext的Web UI的默認(rèn)監(jiān)聽(tīng)端口。

• 指定的部署模式（即master）為local[*]。當(dāng)前應(yīng)用（Application）的ID為local-1497084620457。

• 可以在spark-shell命令行通過(guò)sc使用SparkContext，通過(guò)spark使用SparkSession。sc和spark實(shí)際分別是SparkContext和SparkSession在Spark REPL中的變量名，具體細(xì)節(jié)已在《Spark2.1.0之剖析spark-shell》一文有過(guò)分析。

    由于Spark core的默認(rèn)日志級(jí)別是WARN，所以看到的信息不是很多?，F(xiàn)在我們將Spark安裝目錄的conf文件夾下的log4j.properties.template以如下命令復(fù)制出一份：

1. cp log4j.properties.template log4j.properties  

cp log4j.properties.template log4j.properties

并將log4j.properties中的log4j.logger.org.apache.spark.repl.Main=WARN修改為log4j.logger.org.apache.spark.repl.Main=INFO，然后我們?cè)俅芜\(yùn)行spark-shell，將打印出更豐富的信息，如圖2所示。

圖2  Spark啟動(dòng)過(guò)程打印的部分信息

從圖2展示的啟動(dòng)日志中我們可以看到SecurityManager、SparkEnv、BlockManagerMasterEndpoint、DiskBlockManager、MemoryStore、SparkUI、Executor、NettyBlockTransferService、BlockManager、BlockManagerMaster等信息。它們是做什么的？剛剛接觸Spark的讀者只需要知道這些信息即可，具體內(nèi)容將在后邊的博文給出。

執(zhí)行word count

      這一節(jié)，我們通過(guò)word count這個(gè)耳熟能詳?shù)睦觼?lái)感受下Spark任務(wù)的執(zhí)行過(guò)程。啟動(dòng)spark-shell后，會(huì)打開(kāi)Scala命令行，然后按照以下步驟輸入腳本：

步驟1    

      輸入val lines =sc.textFile("../README.md", 2)，以Spark安裝目錄下的README.md文件的內(nèi)容作為word count例子的數(shù)據(jù)源，執(zhí)行結(jié)果如圖3所示。

圖3   步驟1執(zhí)行結(jié)果

圖3告訴我們lines的實(shí)際類(lèi)型是MapPartitionsRDD。

步驟2

       textFile方法對(duì)文本文件是逐行讀取的，我們需要輸入val words =lines.flatMap(line => line.split(" "))，將每行文本按照空格分隔以得到每個(gè)單詞，執(zhí)行結(jié)果如圖4所示。

圖4   步驟2執(zhí)行結(jié)果

圖4告訴我們lines在經(jīng)過(guò)flatMap方法的轉(zhuǎn)換后得到的words的實(shí)際類(lèi)型也是MapPartitionsRDD。

步驟3

     對(duì)于得到的每個(gè)單詞，通過(guò)輸入val ones = words.map(w => (w,1))，將每個(gè)單詞的計(jì)數(shù)初始化為1，執(zhí)行結(jié)果如圖5所示。

圖5   步驟3執(zhí)行結(jié)果

圖5告訴我們words在經(jīng)過(guò)map方法的轉(zhuǎn)換后得到的ones的實(shí)際類(lèi)型也是MapPartitionsRDD。

步驟4

    輸入val counts = ones.reduceByKey(_ + _)，對(duì)單詞進(jìn)行計(jì)數(shù)值的聚合，執(zhí)行結(jié)果如圖6所示。

圖6   步驟4執(zhí)行結(jié)果

圖6告訴我們ones在經(jīng)過(guò)reduceByKey方法的轉(zhuǎn)換后得到的counts的實(shí)際類(lèi)型是ShuffledRDD。

步驟5

       輸入counts.foreach(println)，將每個(gè)單詞的計(jì)數(shù)值打印出來(lái)，作業(yè)的執(zhí)行過(guò)程如圖7和圖8所示。作業(yè)的輸出結(jié)果如圖9所示。

圖7   步驟5執(zhí)行過(guò)程第一部分

圖8  步驟5執(zhí)行過(guò)程第二部分

圖7和圖8展示了很多作業(yè)提交、執(zhí)行的信息，這里挑選關(guān)鍵的內(nèi)容進(jìn)行介紹：

• SparkContext為提交的Job生成的ID是0。

• 一共有四個(gè)RDD，被劃分為ResultStage和ShuffleMapStage。ShuffleMapStage的ID為0，嘗試號(hào)為0。ResultStage的ID為1，嘗試號(hào)也為0。在Spark中，如果Stage沒(méi)有執(zhí)行完成，就會(huì)進(jìn)行多次重試。Stage無(wú)論是首次執(zhí)行還是重試都被視為是一次Stage嘗試（Stage Attempt），每次Attempt都有一個(gè)唯一的嘗試號(hào)（AttemptNumber）。

• 由于Job有兩個(gè)分區(qū)，所以ShuffleMapStage和ResultStage都有兩個(gè)Task被提交。每個(gè)Task也會(huì)有多次嘗試，因而也有屬于Task的嘗試號(hào)。從圖中看出ShuffleMapStage中的兩個(gè)Task和ResultStage中的兩個(gè)Task的嘗試號(hào)也都是0。

• HadoopRDD則用于讀取文件內(nèi)容。

圖9  步驟5輸出結(jié)果

圖9展示了單詞計(jì)數(shù)的輸出結(jié)果和最后打印的任務(wù)結(jié)束的日志信息。

“Spark2.1.0怎么用”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識(shí)可以關(guān)注億速云網(wǎng)站，小編將為大家輸出更多高質(zhì)量的實(shí)用文章！