第7章 內(nèi)存計算框架Spark

第7章內(nèi)存計算框架Spark7.1.1Spark簡介嵌入式系統(tǒng)基本概念

Spark最初由UCBerkeleyAMPlab(加州大學伯克利分校的AMP實驗室)于2009年開發(fā)，可用于構(gòu)建大型的、低延遲的數(shù)據(jù)分析應用程序，是基于內(nèi)存計算的大數(shù)據(jù)并行計算框架。7.1.1Spark簡介嵌入式系統(tǒng)基本概念

Spark具有如下幾個主要特點：1.運行速度快與MapReduce相比，Spark可以支持包括Map和Reduce在內(nèi)的更多操作，這些操作相互連接形成一個有向無環(huán)圖（DirectedAcyclicGraph，DAG），各個操作的中間結(jié)果被保存在內(nèi)存中，因此，Spark的處理速度比MapReduce快得多。2.容易使用Spark可以使用Scala、Java、Python、R和SOL進行編程，還可以通過SparkShell進行交互式編程。7.1.1Spark簡介嵌入式系統(tǒng)基本概念

3.通用性Spark提供了完整而強大的技術棧，包括SQL查詢、流式計算、機器學習和圖算法等組件，開發(fā)者可以在同一個應用程序中無縫組合地使用這些組件。4.運行模式多樣Spark支持多種運行模式：本地運行模式、分布式運行模式。Spark集群的底層資源可以借助于外部的框架進行管理，比如,Spark可以使用Hadoop的YARN和AppacheMesos作為它的資源管理和調(diào)度器,并且可以訪問HDFS、Cassandra、HBase、Hive等多種數(shù)據(jù)源。7.1.2Spark生態(tài)系統(tǒng)Spark的設計遵循“一個軟件棧滿足不同應用場景”的理念，逐漸形成了一套完整的生態(tài)系統(tǒng)，Spark生態(tài)系統(tǒng)如圖7-1所示。

Spark的設計遵循“一個軟件棧滿足不同應用場景”的理念，逐漸形成了一套完整的生態(tài)系統(tǒng)，Spark生態(tài)系統(tǒng)如圖7-1所示。圖7-1Spark生態(tài)系統(tǒng)7.1.3Spark運行架構(gòu)嵌入式系統(tǒng)基本概念

圖7-2Spark運行架構(gòu)7.2Scala基礎嵌入式系統(tǒng)基本概念由于Spark是由Scala語言編寫的，Scala語言是Spark編程的首選語言，為了后續(xù)更好地學習Spark，需要首先學習Scala語言。7.2.1Scala簡介嵌入式系統(tǒng)基本概念

Scala是一種將面向?qū)ο蠛秃瘮?shù)式編程結(jié)合在一起的高級語言，旨在以簡潔、優(yōu)雅和類型安全的方式表達通用編程模式。Spark支持使用Scala、Java、Python和R語言進行編程。Spark編程首選語言是Scala，因為Spark這個軟件本身就是使用Scala語言開發(fā)的，Scala具有強大的并發(fā)性，支持函數(shù)式編程，可以更好地支持分布式系統(tǒng)，并且支持高效的交互式編程。采用Scala語言編寫Spark應用程序，可以獲得很好的性能。7.2.1Scala簡介嵌入式系統(tǒng)基本概念

Scala是一種基于JVM的跨平臺編程語言，Scala編譯器可以將Scala源碼編譯成符合JVM虛擬機規(guī)范的中間字節(jié)碼文件，在JVM平臺上解釋和運行；Scala語言的API可無縫兼容Java的API。7.2.2Scala編程嵌入式系統(tǒng)基本概念

Scala中使用關鍵字val和var聲明變量。val類似Java中的final變量，也就是常量，一旦初始化將不能修改；var類似Java中的非final變量，可以被多次賦值，多次修改。Scala語言使用靈活，支持自動類型推測，非常簡潔高效。如定義一個初值為5的整型常量，下面兩種寫法是等效的：vala=5vala:Int=57.2.2Scala編程嵌入式系統(tǒng)基本概念【例7-1】從控制臺讀入數(shù)據(jù)：使用readInt、readDouble、readLine輸入整數(shù)、實數(shù)、字符串，使用print、println輸出已讀入的數(shù)據(jù)。scala>importio.StdIn._scala>varx=readInt()scala>vary=readDouble()scala>varstr=readLine("pleaseinputyourname:")scala>print(x)scala>println(y)scala>println(str)7.2.2Scala編程嵌入式系統(tǒng)基本概念

【例7-2】創(chuàng)建一個列表list1，包含10個元素，這10個元素是亂序的，然后這些元素進行以下操作并生成新的列表，list1保持不變：遍歷，排序，反轉(zhuǎn)，求和（多種方法，包括用循環(huán)結(jié)構(gòu)），對所有元素乘2，對所有元素加1，過濾出只包含偶數(shù)的列表，拆分成大小為3的子列表。7.2.2Scala編程嵌入式系統(tǒng)基本概念

//創(chuàng)建一個列表list1scala>vallist1=List(3,9,2,5,7,6,1,8,0,4)//遍歷scala>list1.foreach(println)//排序scala>vallist2=list1.sorted//反轉(zhuǎn)scala>vallist3=list1.reverse7.2.2Scala編程嵌入式系統(tǒng)基本概念

//求和vallist4=list1.sum//求和scala>vallist5=list1.reduce(_+_)//用循環(huán)結(jié)構(gòu)求和scala>varh=0for(i<-0tolist1.length-1)h=h+list1(i)scala>println("h="+h)7.2.2Scala編程嵌入式系統(tǒng)基本概念//每個元素乘以2scala>vallist5=list1.map(_*2)//每個元素加1scala>vallist6=list1.map(_+1)//過濾出只包含偶數(shù)的列表scala>vallist7=list1.filter(x=>x%2==0)7.2.2Scala編程嵌入式系統(tǒng)基本概念

//拆分成大小為3的子列表scala>valged=list1.grouped(3)scala>ged.nextscala>ged.nextscala>ged.nextscala>ged.next7.4RDD編程基礎嵌入式系統(tǒng)基本概念

RDD(ResilientDistributedDataset)，彈性分布式數(shù)據(jù)集，是分布式內(nèi)存的一個抽象概念，RDD提供了一種高度受限的共享內(nèi)存模型，即RDD是只讀的記錄分區(qū)的集合。RDD是Spark的核心概念。7.4.1SparkShell的啟動和退出嵌入式系統(tǒng)基本概念

圖7-17SparkShell成功啟動后的界面7.4.2RDD創(chuàng)建嵌入式系統(tǒng)基本概念

1.從對象集合創(chuàng)建RDDSpark可以調(diào)用SparkContext的parallelize()，以一個已經(jīng)存在的集合（列表）為數(shù)據(jù)源，創(chuàng)建RDD。例如，通過一個List列表創(chuàng)建RDD，命令如圖7-18所示。圖7-18parallelize創(chuàng)建RDD7.4.2RDD創(chuàng)建嵌入式系統(tǒng)基本概念

2.從外部存儲創(chuàng)建RDDSpark的textFile()方法可以從文件系統(tǒng)中加載數(shù)據(jù)創(chuàng)建RDD，下面演示從HDFS和Linux本地加載數(shù)據(jù)創(chuàng)建RDD。（1）從HDFS中加載數(shù)據(jù)例如，在HDFS上有一個文件“/test2/f1.txt”，讀取該文件創(chuàng)建一個RDD，命令如圖7-19所示。圖7-19從HDFS中加載數(shù)據(jù)7.4.2RDD創(chuàng)建嵌入式系統(tǒng)基本概念

（2）從Linux本地加載數(shù)據(jù)例如，在Linux本地有一個文件“/user/data/f2.txt”，讀取該文件創(chuàng)建一個RDD，命令如圖7-20所示。圖7-20從Linux本地加載數(shù)據(jù)7.4.3常用的RDD操作嵌入式系統(tǒng)基本概念RDD的操作分為行動操作（Action）和轉(zhuǎn)換操作（Transformation）。1.行動操作Spark程序執(zhí)行到行動操作時，才會從文件中加載數(shù)據(jù)，完成一次又一次轉(zhuǎn)換操作，最后完成行動操作。由于轉(zhuǎn)換操作需要行動操作才能查看命令執(zhí)行的結(jié)果，所以先介紹行動操作。7.4.3常用的RDD操作嵌入式系統(tǒng)基本概念

（1）count()返回RDD中的元素個數(shù)（2）collect()以數(shù)組的形式返回數(shù)據(jù)集中的所有元素（3）first()返回數(shù)據(jù)集中的第一個元素（4）take(n)返回一個數(shù)組,數(shù)組元素由數(shù)據(jù)集的前n個元素組成（5）reduce(func)通過函數(shù)func聚合數(shù)據(jù)集中的元素（6）foreach(func)對RDD中的每一個元素運行給定的函數(shù)func7.4.3常用的RDD操作嵌入式系統(tǒng)基本概念例如，統(tǒng)計rdd1中的元素個數(shù)，以數(shù)組的形式返回rdd1的所有元素，取rdd1的第一個元素，取rdd1的前3個元素，求rdd1中所有元素之和，輸出rdd1的所有元素，代碼及執(zhí)行結(jié)果如圖7-21所示。RDD的操作如果沒有參數(shù)，后面的括號是可以省略的。7.4.3常用的RDD操作嵌入式系統(tǒng)基本概念

圖7-21RDD的行動操作7.4.3常用的RDD操作嵌入式系統(tǒng)基本概念2.轉(zhuǎn)換操作對于RDD而言，每一次轉(zhuǎn)換操作都會產(chǎn)生新的RDD，供給下一個“轉(zhuǎn)換”使用。轉(zhuǎn)換操作是惰性操作，整個轉(zhuǎn)換過程只是記錄了轉(zhuǎn)換的軌跡，并不會發(fā)生真正的計算，只有遇到行動操作時，才會觸發(fā)“從頭到尾”的真正的計算。7.4.3常用的RDD操作嵌入式系統(tǒng)基本概念

（1）filter(func)例如，過濾出rdd1中大于3的元素，并輸出結(jié)果，可用兩種方法實現(xiàn)，代碼及執(zhí)行結(jié)果如圖7-22所示。7.4.3常用的RDD操作嵌入式系統(tǒng)基本概念圖7-22過濾出rdd1中大于3的元素7.4.3常用的RDD操作嵌入式系統(tǒng)基本概念

（2）map(func)例如，將rdd1中每一個元素加5，并輸出結(jié)果，代碼及執(zhí)行結(jié)果如圖7-23所示。圖7-23將rdd1中每一個元素加57.4.3常用的RDD操作嵌入式系統(tǒng)基本概念

（3）flatMap(func)flatMap(func)操作是將RDD中的每一個元素傳遞到函數(shù)func中，將返回的迭代器（數(shù)組、列表）中的所有元素構(gòu)成新的RDD。例如，用map(func)和flatMap(func)分割字符串，代碼及執(zhí)行結(jié)果如圖7-24所示。7.4.3常用的RDD操作嵌入式系統(tǒng)基本概念

圖7-24用map(func)和flatMap(func)分割字符串7.4.3常用的RDD操作嵌入式系統(tǒng)基本概念（4）groupByKey()groupByKey()應用于(key,value)鍵值對的RDD，可以將相同key的元素聚集到一起，返回一個新的(key,Iterable)形式的RDD。例如，有兩位員工zhangsan和lisi，zhangsan的工資和獎金分別為5000元、1500元，lisi的工資和獎金分別為6000元、2000元，統(tǒng)計zhangsan和lisi的總收入，代碼及執(zhí)行結(jié)果如圖7-25所示。7.4.3常用的RDD操作嵌入式系統(tǒng)基本概念

圖7-25groupByKey()示例7.4.3常用的RDD操作嵌入式系統(tǒng)基本概念（5）reduceByKey(func)reduceByKey(func)應用于(key,value)鍵值對的RDD時，返回一個新的(key,value)形式的RDD，其中的每個值是將每個key傳遞到函數(shù)func中進行聚合后得到的結(jié)果。用reduceByKey(func)重做上例，代碼及執(zhí)行結(jié)果如圖7-26所示。7.4.3常用的RDD操作嵌入式系統(tǒng)基本概念圖7-26reduceByKey(func)示例7.4.3常用的RDD操作嵌入式系統(tǒng)基本概念

（6）sortBy()sortBy()將RDD中的元素按照某一字段進行排序，其中第一個參數(shù)為排序字段，第二個參數(shù)是一個布爾值，指定升序（默認）或降序。若第二個參數(shù)是false則為降序排列。例如，列表中存放了一些(key,value)鍵值對，用該列表創(chuàng)建RDD，然后對該RDD中的元素按照第二個字段降序排列，代碼及執(zhí)行結(jié)果如圖7-27所示。7.4.3常用的RDD操作嵌入式系統(tǒng)基本概念圖7-27sortBy()示例7.4.3常用的RDD操作嵌入式系統(tǒng)基本概念

（7）union()union()的參數(shù)是RDD，它將兩個RDD合并為一個新的RDD，兩個RDD中的數(shù)據(jù)類型要保持一致。例如，創(chuàng)建兩個RDD，并將這兩個RDD合并成一個新的RDD，代碼及執(zhí)行結(jié)果如圖7-28所示。7.4.3常用的RDD操作嵌入式系統(tǒng)基本概念

圖7-28union()示例7.4.3常用的RDD操作嵌入式系統(tǒng)基本概念

（8）mapValues(func)mapValues(func)對鍵值對RDD中的每個value都應用函數(shù)func，但是key不會發(fā)生變化。例如，將鍵值對RDD中的每個value加2，代碼及執(zhí)行結(jié)果如圖7-29所示。7.4.3常用的RDD操作嵌入式系統(tǒng)基本概念

圖7-29mapValues(func)示例7.4.3常用的RDD操作嵌入式系統(tǒng)基本概念

（9）keys鍵值對RDD中的每個元素都是(key,value)形式，keys操作只會把鍵值對RDD中的key返回，形成一個新的RDD。（10）values鍵值對RDD中的每個元素都是(key,value)形式，values操作只會把鍵值對RDD中的value返回，形成一個新的RDD。例如，分別輸出鍵值對RDD中的每個元素的鍵和值，代碼及執(zhí)行結(jié)果如圖7-30所示。7.4.3常用的RDD操作嵌入式系統(tǒng)基本概念

圖7-30keys和values示例7.4.4RDD編程實例嵌入式系統(tǒng)基本概念

【例7-3】假設在HDFS的/test2目錄下，有多個文本文件，每個文件存儲多個英文單詞，單詞之間用空格分隔。請統(tǒng)計每個單詞出現(xiàn)的次數(shù)。7.4.4RDD編程實例嵌入式系統(tǒng)基本概念

scala>vallines=sc.textFile("/test2")scala>valrddfalt=lines.flatMap(x=>x.split(""))scala>valwordCount=rddfalt.map(x=>(x,1)).reduceByKey((a,b)=>a+b)scala>wordCount.collect()scala>wordCount.foreach(println)7.4.4RDD編程實例嵌入式系統(tǒng)基本概念

【例7-4】hadoop.txt和spark.txt分別存儲了同一個班的《Hadoop基礎》和《Spark編程基礎》成績，其中第1列表示學號，第2列表示成績。請在SparkShell中通過編程來完成以下操作：7.4.4RDD編程實例嵌入式系統(tǒng)基本概念

1.從本地文件系統(tǒng)中加載hadoop.txtscala>valhadoop1=sc.textFile("file:///home/shxx/data/hadoop.txt")2.從本地文件系統(tǒng)中加載spark.txtscala>valspark1=sc.textFile("file:///home/shxx/data/spark.txt")7.4.4RDD編程實例嵌入式系統(tǒng)基本概念

3.查詢學生spark成績中的前3名使用map轉(zhuǎn)換數(shù)據(jù)，每條數(shù)據(jù)被分割成2列，表示學號和成績，分隔符為“,”，存儲為二元組格式，成績要求轉(zhuǎn)化為Int類型。scala>valm_spark=spark1.map{x=>valline=x.split(",");(line(0),line(1).trim.toInt)}通過sortBy對元組中的成績列降序排序scala>valsort_spark=m_spark.sortBy(x=>x._2,false)通過take()操作取出每個RDD的前3個值scala>sort_spark.take(3).foreach(println)7.4.4RDD編程實例嵌入式系統(tǒng)基本概念4.查詢spark成績90分以上的學生scala>valfilter_spark=m_spark.filter(x=>x._2>90)scala>filter_spark.foreach(println)5.查詢spark成績的最高分和最低分scala>valmax_spark=m_spark.values.maxscala>valmin_spark=m_spark.values.min7.4.4RDD編程實例嵌入式系統(tǒng)基本概念6.求spark課程的平均成績scala>valp1_spark=m_spark.values.map(x=>(x,1.0)).reduce((x,y)=>(x._1+y._1,x._2+y._2))scala>valp2_spark=p1_spark._1/p1_spark._27.4.4RDD編程實例嵌入式系統(tǒng)基本概念7.輸出每位學生所有科目的總成績scala>valm_hadoop=hadoop1.map{x=>valline=x.split(",");(line(0),line(1).trim.toInt)}scala>valunion_hs=m_spark.union(m_hadoop)scala>valall_score=union_hs.reduceByKey((a,b)=>a+b)scala>all_score.foreach(println)7.4.4RDD編程實例嵌入式系統(tǒng)基本概念

8.輸出每位學生的平均成績scala>valpj_score=all_score.mapValues(x=>x/2.0)scala>pj_score.foreach(println)7.5IDEA的安裝和使用嵌入式系統(tǒng)基本概念IDEA全稱IntelliJIDEA，是一個通用的集成開發(fā)環(huán)境，IntelliJIDEA在業(yè)界被公認為最好的scala和java開發(fā)工具之一。我們可以使用IDEA開發(fā)Spark獨立應用程序，7.5.2IDEA的使用嵌入式系統(tǒng)基本概念在$SPARK_HOME/examples/src/main/scala/org/apache/spark/examples目錄下，Spark自帶了名為“SparkPi”的例子程序，該例子采用分布式估算的方法求圓周率。計算原理為：假設有一個正方形，邊長是x，該正方形里面有一個內(nèi)切圓，可以算出正方形的面積S等于x2，內(nèi)切圓的面積C等于Pi×(x/2)2，于是就有Pi=4×C/S?？梢岳糜嬎銠C隨機產(chǎn)生大量位于正方形內(nèi)部的點，通過點的數(shù)量去近似表示面積。假設位于正方形中點的數(shù)量為Ps，落在圓內(nèi)的點的數(shù)量為Pc，則當隨機點的數(shù)量趨近于無窮時，4×Pc/Ps將逼近于Pi。7.5.2IDEA的使用嵌入式系統(tǒng)基本概念importscala.math.randomimportorg.apache.spark.sql.SparkSession/**Computesanapproximationtopi*/objectSparkPi{defmain(args:Array[String]):Unit={valspark=SparkSession.builder.appName("SparkPi")/**如果在IDEA本地上運行需要加上.master("local")，如果要提交到Spark集群一定不能加.master("local")*/.master("local").getOrCreate()7.5.2IDEA的使用嵌入式系統(tǒng)基本概念

valslices=if(args.length>0)args(0).toIntelse2valn=math.min(100000L*slices,Int.MaxValue).toInt//avoidoverflowvalcount=spark.sparkContext.parallelize(1untiln,slices).map{i=>valx=random*2-1valy=random*2-1if(x*x+y*y<=1)1else0}.reduce(_+_)println(s"Piisroughly${4.0*count/(n-1)}")spark.stop()}}1.4.3YARN嵌入式系統(tǒng)基本概念

YARN（YetAnotherResourceNegotiator，另一種資源協(xié)調(diào)者）是一種新的Hadoop資源管理器，它是一個通用資源管理系統(tǒng)，可為上層應用提供統(tǒng)一的資源管理和調(diào)度，它的引入為集群在資源利用率、資源統(tǒng)一管理和數(shù)據(jù)共享等方面帶來了很大好處。YARN采用主從架構(gòu)（Master/Slave），包括：ResourceManager、ApplicationMaster和NodeManager三個核心組件。1.5Hadoop生態(tài)系統(tǒng)嵌入式系統(tǒng)基本概念

圖1-1hadoop的生態(tài)系統(tǒng)1.5.1HBase嵌入式系統(tǒng)基本概念HBase（HadoopDatabase，hadoop數(shù)據(jù)庫）是一個針對非結(jié)構(gòu)化數(shù)據(jù)的可伸縮、高可靠、高性能、分布式和面向列的開源數(shù)據(jù)庫，一般采用HDFS作為底層數(shù)據(jù)存儲系統(tǒng)。HBase是針對GoogleBigtable的開源實現(xiàn)，二者采用相同數(shù)據(jù)模型，具有強大的非結(jié)構(gòu)化數(shù)據(jù)存儲能力。HBase使用ZooKeeper進行管理，它保障查詢速度的一個關鍵因素就是RowKey的設計是否合理。HBase中保存的數(shù)據(jù)可以使用MapReduce來處理。1.5.2Hive嵌入式系統(tǒng)基本概念

Hive是一個基于Hadoop的數(shù)據(jù)倉庫工具，最早由Facebook開發(fā)并使用。Hive可以將結(jié)構(gòu)化的數(shù)據(jù)文件映射為一張數(shù)據(jù)庫表，并提供SQL查詢功能，這套SQL稱為HiveQL（HiveQueryLanguage）。當采用MapReduce作為執(zhí)行引擎時，Hive能將HiveQL語句轉(zhuǎn)換成一系列MapReduce作業(yè)，并提交到Hadoop集群上運行。Hive大大降低了學習門檻，同時也提升了開發(fā)效率，非常適合數(shù)據(jù)倉庫的統(tǒng)計分析。1.5.3Pig嵌入式系統(tǒng)基本概念

Pig是一種數(shù)據(jù)流語言和運行環(huán)境，用于檢索大規(guī)模的數(shù)據(jù)集。Pig定義了一種數(shù)據(jù)流語言—PigLatin，將腳本轉(zhuǎn)換為MapReduce任務在Hadoop上執(zhí)行，適合于