廈門大學(xué)-林子雨-大數(shù)據(jù)技術(shù)原理與應(yīng)用-進階學(xué)習(xí)自學(xué)教程-Spark快速入門指南-–-Spark的安裝與基礎(chǔ)使用

智建領(lǐng)航者智建領(lǐng)航者廈門大學(xué)林子雨編著《大數(shù)據(jù)技術(shù)原理與應(yīng)用》進階學(xué)習(xí)自學(xué)教程Spark快速入門指南–Spark安裝與基礎(chǔ)使用主講教師：林子雨廈門大學(xué)數(shù)據(jù)庫實驗室二零一六年一月智建領(lǐng)航者智建領(lǐng)航者

目錄1 前言 12 準備工作 13 安裝Spark 14 運行Spark示例 25 通過SparkShell進行交互分析 35.1 基礎(chǔ)操作 45.2 RDD的更多操作 55.3 緩存 66 SparkSQL和DataFrames 67 SparkStreaming 88 獨立應(yīng)用程序（Self-ContainedApplication） 98.1 應(yīng)用程序代碼 98.2 安裝sbt 108.3 使用sbt打包Scala程序 128.4 通過spark-submit運行程序 139 進階學(xué)習(xí) 13附錄1:任課教師介紹 13附錄2：課程教材介紹 14附錄3：中國高校大數(shù)據(jù)課程公共服務(wù)平臺介紹 15智建領(lǐng)航者智建領(lǐng)航者

《大數(shù)據(jù)技術(shù)原理與應(yīng)用》Spark快速入門指南–Spark安裝與基礎(chǔ)使用主講教師：林子雨E-mail:ziyulin@個人主頁：/linziyu前言ApacheSpark是一個新興的大數(shù)據(jù)處理通用引擎，提供了分布式的內(nèi)存抽象。Spark正如其名，最大的特點就是快（Lightning-fast），可比HadoopMapReduce的處理速度快100倍。此外，Spark提供了簡單易用的API，幾行代碼就能實現(xiàn)WordCount。本教程主要參考官網(wǎng)快速入門教程，介紹了Spark的安裝，Sparkshell、RDD、SparkSQL、SparkStreaming等的基本使用。本教程的具體運行環(huán)境如下：CentOS6.4Spark1.6Hadoop2.6.0JavaJDK1.7Scala2.10.5準備工作運行Spark需要JavaJDK1.7，CentOS6.x系統(tǒng)默認只安裝了JavaJRE，還需要安裝JavaJDK，并配置好JAVA_HOME變量。此外，Spark會用到HDFS與YARN，因此請先安裝Hadoop，具體請瀏覽Hadoop安裝教程，在此就不再復(fù)述。安裝Spark待Hadoop安裝好之后，我們再開始安裝Spark。官網(wǎng)下載地址：/downloads.html本教程選擇的是Spark1.6.0版本，選擇packagetype為“Pre-buildwithuser-providedHadoop[canusewithmostHadoopdistributions]”，再點擊給出的下載連接/dyn/closer.lua/spark/spark-1.6.0/spark-1.6.0-bin-without-hadoop.tgz就可以下載了，如下圖所示：PackagetypeSourcecode:Spark源碼，需要編譯才能使用，另外Scala2.11需要使用源碼編譯才可使用Pre-buildwithuser-providedHadoop:“Hadoopfree”版，可應(yīng)用到任意Hadoop版本Pre-buildforHadoop2.6andlater:基于Hadoop2.6的預(yù)先編譯版，需要與本機安裝的Hadoop版本對應(yīng)?？蛇x的還有Hadoop2.4andlater、Hadoop2.3、Hadoop1.x，以及CDH4。為方便，本教程選擇的是Pre-buildwithuser-providedHadoop，簡單配置后可應(yīng)用到任意Hadoop版本。下載后，執(zhí)行如下命令進行安裝：sudotar-zxf~/下載/spark-1.6.0-bin-without-hadoop.tgz-C/usr/local/cd/usr/localsudomv./spark-1.6.0-bin-without-hadoop/./sparksudochown-Rhadoop:hadoop./spark#此處的hadoop為你的用戶名安裝后，需要在./conf/spark-env.sh中修改Spark的Classpath，執(zhí)行如下命令拷貝一個配置文件：cd/usr/local/sparkcp./conf/spark-env.sh.template./conf/spark-env.sh編輯./conf/spark-env.sh（vim./conf/spark-env.sh），在最后面加上如下一行：exportSPARK_DIST_CLASSPATH=$(/usr/local/hadoop/bin/hadoopclasspath)保存后，Spark就可以啟動、運行了。運行Spark示例注意，必須安裝Hadoop才能使用Spark，但如果使用Spark過程中沒用到HDFS，不啟動Hadoop也是可以的。此外，接下來教程中出現(xiàn)的命令、目錄，若無說明，則一般以Spark的安裝目錄（/usr/local/spark）為當前路徑，請注意區(qū)分。在./examples/src/main目錄下有一些Spark的示例程序，有Scala、Java、Python、R等語言的版本。我們可以先運行一個示例程序SparkPi（即計算π的近似值），執(zhí)行如下命令：cd/usr/local/spark./bin/run-exampleSparkPi執(zhí)行時會輸出非常多的運行信息，輸出結(jié)果不容易找到，可以通過grep命令進行過濾（命令中的2>&1可以將所有的信息都輸出到stdout中，否則由于輸出日志的性質(zhì)，還是會輸出到屏幕中）：./bin/run-exampleSparkPi2>&1|grep"Piisroughly"過濾后的運行結(jié)果如下圖所示，可以得到π的5位小數(shù)近似值：Python版本的SparkPi則需要通過spark-submit運行：./bin/spark-submitexamples/src/main/python/pi.py通過SparkShell進行交互分析Sparkshell提供了簡單的方式來學(xué)習(xí)API，也提供了交互的方式來分析數(shù)據(jù)。SparkShell支持Scala和Python，本教程選擇使用Scala來進行介紹。Scala是一門現(xiàn)代的多范式編程語言，志在以簡練、優(yōu)雅及類型安全的方式來表達常用編程模式。它平滑地集成了面向?qū)ο蠛秃瘮?shù)語言的特性。Scala運行于Java平臺（JVM，Java虛擬機），并兼容現(xiàn)有的Java程序。Scala是Spark的主要編程語言，如果僅僅是寫Spark應(yīng)用，并非一定要用Scala，用Java、Python都是可以的。使用Scala的優(yōu)勢是開發(fā)效率更高，代碼更精簡，并且可以通過SparkShell進行交互式實時查詢，方便排查問題。執(zhí)行如下命令啟動SparkShell：./bin/spark-shell啟動成功后如圖所示，會有“scala>”的命令提示符?；A(chǔ)操作Spark的主要抽象是分布式的元素集合（distributedcollectionofitems），稱為RDD（ResilientDistributedDataset，彈性分布式數(shù)據(jù)集），它可被分發(fā)到集群各個節(jié)點上，進行并行操作。RDDs可以通過HadoopInputFormats創(chuàng)建（如HDFS），或者從其他RDDs轉(zhuǎn)化而來。我們從./README文件新建一個RDD，代碼如下（代碼中//后的內(nèi)容為注釋，本教程以////開頭的內(nèi)容表示交互式輸出結(jié)果）：valtextFile=sc.textFile("file:///usr/local/spark/README.md")////textFile:org.apache.spark.rdd.RDD[String]=MapPartitionsRDD[1]attextFileat<console>:27代碼中通過“file://”前綴指定讀取本地文件。Sparkshell默認是讀取HDFS中的文件，需要先上傳文件到HDFS中，否則會有“org.apache.hadoop.mapred.InvalidInputException:Inputpathdoesnotexist:hdfs://localhost:9000/user/hadoop/README.md”的錯誤。上述命令的輸出結(jié)果如下圖所示：RDDs支持兩種類型的操作：actions:在數(shù)據(jù)集上運行計算后返回值transformations:轉(zhuǎn)換,從現(xiàn)有數(shù)據(jù)集創(chuàng)建一個新的數(shù)據(jù)集下面我們就來演示count()和first()操作：textFile.count()//RDD中的item數(shù)量，對于文本文件，就是總行數(shù)////res0:Long=95textFile.first()//RDD中的第一個item，對于文本文件，就是第一行內(nèi)容////res1:String=#ApacheSpark接著演示transformation，通過filtertransformation來返回一個新的RDD，代碼如下：vallinesWithSpark=textFile.filter(line=>line.contains("Spark"))//篩選出包含Spark的行l(wèi)inesWithSpark.count()//統(tǒng)計行數(shù)////res4:Long=17可以看到一共有17行內(nèi)容包含Spark，這與通過Linux命令cat./README.md|grep"Spark"-c得到的結(jié)果一致，說明是正確的。action和transformation可以用鏈式操作的方式結(jié)合使用，使代碼更為簡潔：textFile.filter(line=>line.contains("Spark")).count()//統(tǒng)計包含Spark的行數(shù)////res4:Long=17RDD的更多操作RDD的actions和transformations可用在更復(fù)雜的計算中，例如通過如下代碼可以找到包含單詞最多的那一行內(nèi)容共有幾個單詞：textFile.map(line=>line.split("").size).reduce((a,b)=>if(a>b)aelseb)////res5:Int=14代碼首先將每一行內(nèi)容map為一個整數(shù)，這將創(chuàng)建一個新的RDD，并在這個RDD中執(zhí)行reduce操作，找到最大的數(shù)。map()、reduce()中的參數(shù)是Scala的函數(shù)字面量（functionliterals，也稱為閉包closures），并且可以使用語言特征或Scala/Java的庫。例如，通過使用Math.max()函數(shù)（需要導(dǎo)入Java的Math庫），可以使上述代碼更容易理解：importjava.lang.MathtextFile.map(line=>line.split("").size).reduce((a,b)=>Math.max(a,b))////res6:Int=14HadoopMapReduce是常見的數(shù)據(jù)流模式，在Spark中同樣可以實現(xiàn)（下面這個例子也就是WordCount）：valwordCounts=textFile.flatMap(line=>line.split("")).map(word=>(word,1)).reduceByKey((a,b)=>a+b)//實現(xiàn)單詞統(tǒng)計////wordCounts:org.apache.spark.rdd.RDD[(String,Int)]=ShuffledRDD[4]atreduceByKeyat<console>:29wordCounts.collect()//輸出單詞統(tǒng)計結(jié)果////res7:Array[(String,Int)]=Array((package,1),(For,2),(Programs,1),(processing.,1),(Because,1),(The,1)...)緩存Spark支持在集群范圍內(nèi)將數(shù)據(jù)集緩存至每一個節(jié)點的內(nèi)存中，可避免數(shù)據(jù)傳輸，當數(shù)據(jù)需要重復(fù)訪問時這個特征非常有用，例如查詢體積小的“熱”數(shù)據(jù)集，或是運行如PageRank的迭代算法。調(diào)用cache()，就可以將數(shù)據(jù)集進行緩存：linesWithSpark.cache()SparkSQL和DataFramesSparkSQL是Spark內(nèi)嵌的模塊，用于結(jié)構(gòu)化數(shù)據(jù)。在Spark程序中可以使用SQL查詢語句或DataFrameAPI。DataFrames和SQL提供了通用的方式來連接多種數(shù)據(jù)源，支持Hive、Avro、Parquet、ORC、JSON、和JDBC，并且可以在多種數(shù)據(jù)源之間執(zhí)行join操作。下面仍在Sparkshell中演示一下SparkSQL的基本操作，該部分內(nèi)容主要參考了SparkSQL、DataFrames和Datasets指南。SparkSQL的功能是通過SQLContext類來使用的，而創(chuàng)建SQLContext是通過SparkContext創(chuàng)建的。在Sparkshell啟動時，輸出日志的最后有這么幾條信息：16/01/1613:25:41INFOrepl.SparkILoop:Createdsparkcontext..Sparkcontextavailableassc.16/01/1613:25:41INFOrepl.SparkILoop:Createdsqlcontext..SQLcontextavailableassqlContext.這些信息表明SparkContent和SQLContext都已經(jīng)初始化好了，可通過對應(yīng)的sc、sqlContext變量直接進行訪問。使用SQLContext可以從現(xiàn)有的RDD或數(shù)據(jù)源創(chuàng)建DataFrames。作為示例，我們通過Spark提供的JSON格式的數(shù)據(jù)源文件./examples/src/main/resources/people.json來進行演示，該數(shù)據(jù)源內(nèi)容如下：{"name":"Michael"}{"name":"Andy","age":30}{"name":"Justin","age":19}執(zhí)行如下命令導(dǎo)入數(shù)據(jù)源，并輸出內(nèi)容：valdf=sqlContext.read.json("file:///usr/local/spark/examples/src/main/resources/people.json")////df:org.apache.spark.sql.DataFrame=[age:bigint,name:string]df.show()//輸出數(shù)據(jù)源內(nèi)容////+++////|age|name|////+++////|null|Michael|////|30|Andy|////|19|Justin|////+++接著，我們來演示DataFrames處理結(jié)構(gòu)化數(shù)據(jù)的一些基本操作：df.select("name").show()//只顯示"name"列////++////|name|////++////|Michael|////|Andy|////|Justin|////++df.select(df("name"),df("age")+1).show()//將"age"加1////+++////|name|(age+1)|////+++////|Michael|null|////|Andy|31|////|Justin|20|////+++df.filter(df("age")>21).show()#條件語句////+++////|age|name|////+++////|30|Andy|////+++df.groupBy("age").count().show()//groupBy操作////+++////|age|count|////+++////|null|1|////|19|1|////|30|1|////+++當然，我們也可以使用SQL語句來進行操作：df.registerTempTable("people")//將DataFrame注冊為臨時表peoplevalresult=sqlContext.sql("SELECTname,ageFROMpeopleWHEREage>=13ANDage<=19")//執(zhí)行SQL查詢result.show()//輸出結(jié)果////+++////|name|age|////+++////|Justin|19|////+++更多的功能可以查看完整的DataFramesAPI，此外DataFrames也包含了豐富的DataFramesFunction可用于字符串處理、日期計算、數(shù)學(xué)計算等。SparkStreaming流計算除了使用Storm框架，使用SparkStreaming也是一個很好的選擇?；赟parkStreaming，可以方便地構(gòu)建可拓展、高容錯的流計算應(yīng)用程序。SparkStreaming使用SparkAPI進行流計算，這意味著在Spark上進行流處理與批處理的方式一樣。因此，你可以復(fù)用批處理的代碼，使用SparkStreaming構(gòu)建強大的交互式應(yīng)用程序，而不僅僅是用于分析數(shù)據(jù)。下面以一個簡單的SparkStreaming示例（基于流的單詞統(tǒng)計）來演示一下SparkStreaming：本地服務(wù)器通過TCP接收文本數(shù)據(jù)，實時輸出單詞統(tǒng)計結(jié)果。該部分內(nèi)容主要參考了SparkStreaming編程指南。運行該示例需要Netcat（在網(wǎng)絡(luò)上通過TCP或UDP讀寫數(shù)據(jù)），CentOS6.x系統(tǒng)中默認沒有安裝，經(jīng)過測試，如果通過yum直接安裝，運行時會有“nc:Protocolnotavailable”的錯誤，需要下載較低版本的nc才能正常使用。我們選擇Netcat0.6.1版本，在終端中運行如下命令進行安裝：wget/project/netcat/netcat/0.6.1/netcat-0.6.1-1.i386.rpm-O~/netcat-0.6.1-1.i386.rpm#下載sudorpm-iUv~/netcat-0.6.1-1.i386.rpm#安裝安裝好NetCat之后，使用如下命令建立本地數(shù)據(jù)服務(wù)，監(jiān)聽TCP端口9999：#記為終端1nc-l-p9999啟動后，該端口就被占用了，需要開啟另一個終端運行示例程序，執(zhí)行如下命令：#需要另外開啟一個終端，記為終端2，然后運行如下命令/usr/local/spark/bin/run-examplestreaming.NetworkWordCountlocalhost9999接著在終端1中輸入文本，在終端2中就可以實時看到單詞統(tǒng)計結(jié)果了。SparkStreaming的內(nèi)容較多，本教程就簡單介紹到這，更詳細的內(nèi)容可查看官網(wǎng)教程。最后需要關(guān)掉終端2，并按ctrl+c退出終端1的Netcat。獨立應(yīng)用程序（Self-ContainedApplication）接著我們通過一個簡單的應(yīng)用程序SimpleApp來演示如何通過SparkAPI編寫一個獨立應(yīng)用程序。使用Scala編寫的程序需要使用sbt進行編譯打包，相應(yīng)的，Java程序使用Maven編譯打包，而Python程序通過spark-submit直接提交。應(yīng)用程序代碼在終端中執(zhí)行如下命令創(chuàng)建一個文件夾sparkapp作為應(yīng)用程序根目錄：cd~#進入用戶主文件夾mkdir./sparkapp#創(chuàng)建應(yīng)用程序根目錄mkdir-p./sparkapp/src/main/scala#創(chuàng)建所需的文件夾結(jié)構(gòu)在./sparkapp/src/main/scala下建立一個名為SimpleApp.scala的文件（vim./sparkapp/src/main/scala/SimpleApp.scala），添加代碼如下：/*SimpleApp.scala*/importorg.apache.spark.SparkContextimportorg.apache.spark.SparkContext._importorg.apache.spark.SparkConfobjectSimpleApp{defmain(args:Array[String]){vallogFile="file:///usr/local/spark/README.md"http://Shouldbesomefileonyoursystemvalconf=newSparkConf().setAppName("SimpleApplication")valsc=newSparkContext(conf)vallogData=sc.textFile(logFile,2).cache()valnumAs=logData.filter(line=>line.contains("a")).count()valnumBs=logData.filter(line=>line.contains("b")).count()println("Lineswitha:%s,Lineswithb:%s".format(numAs,numBs))}}該程序計算/usr/local/spark/README文件中包含“a”的行數(shù)和包含“b”的行數(shù)。代碼第8行的/usr/local/spark為Spark的安裝目錄，如果不是該目錄請自行修改。不同于Sparkshell，獨立應(yīng)用程序需要通過valsc=newSparkContext(conf)初始化SparkContext，SparkContext的參數(shù)SparkConf包含了應(yīng)用程序的信息。該程序依賴SparkAPI，因此我們需要通過sbt進行編譯打包。在./sparkapp中新建文件simple.sbt（vim./sparkapp/simple.sbt），添加內(nèi)容如下，聲明該獨立應(yīng)用程序的信息以及與Spark的依賴關(guān)系：name:="SimpleProject"version:="1.0"scalaVersion:="2.10.5"libraryDependencies+="org.apache.spark"%%"spark-core"%"1.6.0"文件simple.sbt需要指明Spark和Scala的版本。啟動Sparkshell的過程中，當輸出到Spark的符號圖形時，可以看到相關(guān)的版本信息。安裝sbtSpark中沒有自帶sbt，需要手動安裝sbt，我們選擇安裝在/usr/local/sbt中：sudomkdir/usr/local/sbtsudochown-Rhadoop/usr/local/sbt#此處的hadoop為你的用戶名cd/usr/local/sbt經(jīng)筆者測試，按官網(wǎng)教程安裝sbt0.13.9后，使用時可能存在網(wǎng)絡(luò)問題，無法下載依賴包，導(dǎo)致sbt無法正常使用，需要進行一定的修改。為方便，請使用筆者修改后的版本，下載地址：/s/1eRyFddw。下載后，執(zhí)行如下命令拷貝至/usr/local/sbt中：cp~/下載/sbt-launch.jar.接著在/usr/local/sbt中創(chuàng)建sbt腳本（vim./sbt），添加如下內(nèi)容：#!/bin/bashSBT_OPTS="-Xms512M-Xmx1536M-Xss1M-XX:+CMSClassUnloadingEnabled-XX:MaxPermSize=256M"java$SBT_OPTS-jar`dirname$0`/sbt-launch.jar"$@"保存后，為./sbt腳本增加可執(zhí)行權(quán)限：chmodu+x./sbt最后檢驗sbt是否可用（首次運行會處于“Gettingorg.scala-sbtsbt0.13.9…”的下載狀態(tài)，請耐心等待。筆者等待了7分鐘才出現(xiàn)第一條下載提示）：./sbtsbt-version下載過程中可能會類似“ServeraccessError:java.security.ProviderException:java.security.KeyExceptionurl=/org/scala-sbt/precompiled-2_9_3/0.13.9/precompiled-2_9_3-0.13.9.jar”的錯誤，可以忽略?？稍賵?zhí)行一次./sbtsbt-version，只要能得到如下圖的版本信息就沒問題：如果由于網(wǎng)絡(luò)問題無法下載依賴，導(dǎo)致sbt無法正確運行的話，可以下載筆者提供的離線依賴包sbt-0.13.9-repo.tar.gz到本地中（依賴包的本地位置為~/.sbt和~/.ivy2，檢查依賴關(guān)系時，首先檢查本地，本地未找到，再從網(wǎng)絡(luò)中下載），下載地址：/s/1sjTQ8yD。下載后，執(zhí)行如下命令解壓依賴包：tar-zxf~/下載/sbt-0.13.9-local-repo.tar.gz~通過這個方式，一般可以解決依賴包缺失的問題（讀者提供的依賴包僅適合于Spark1.6版本，不同版本依賴關(guān)系不一樣）。使用sbt打包Scala程序為保證sbt能正常運行，先執(zhí)行如下命令檢查整個應(yīng)用程序的文件結(jié)構(gòu)：cd~/sparkappfind.文件結(jié)構(gòu)應(yīng)如下圖所示：接著，我們就可以通過如下代碼將整個應(yīng)用程序打包成JAR（首次運行同樣需要下載依賴包，如果這邊遇到網(wǎng)絡(luò)問題無法成功，也請下載上述安裝sbt提到的離線依賴包sbt-0.13.9-repo.tar.gz）：/usr/local/sbt/sbtpackage打包成功的話，會輸出如下圖內(nèi)容：生成的jar包的位置為~/sparkapp/target/scala-2.10/simple-project_2.10-1.0.jar。通過spark-submit運行程序最后，我們就可以將生成的jar包通過spark-submit提交到Spark中運行了，命令如下：/usr/local/spark/bin/spark-submit\--class"SimpleApp"~/sparkapp/target/scala-2.10/simple-project_2.10-1.0.jar#輸出信息太多，可以通過如下命令過濾直接查看結(jié)果/usr/local/spark/bin/spark-submit\--class"SimpleApp"~/sparkapp/target/scala-2.10/simple-project_2.10-1.0.jar2>&1\|grep"Lineswitha:"最終得到的結(jié)果如下：Lineswith