云計算與大數(shù)據(jù)--配套習題(徐小龍) 第16章 分布式內存計算平臺Spark-習題答案

1、第16章 分布式內存計算平臺Spark習題選擇題1、Spark是Hadoop生態(tài)（B ）組件的替代方案。HadoopB. MapReduceC. YamD. HDFS2、以下（D ）不是Spark的主要組件。DriverB. SparkContextC. CIusterManagerD. ResourceManager3、Spark 中的 Execulor 是（ A ）。A.執(zhí)行器 B.主節(jié)點C.從節(jié)點D.上下文4、下面（D ）不是Spark的四大組件之一。Spark StreamingB.Spark MLlibSpark GraphXD.Spark R5、Scala屬于哪種編程語言（C ）.

2、A.匯編語言B.機器語言C.函數(shù)式編程語言D.多范式編程語言6、Spark組件中，SparkContext是應用的（C ）,控制應用的生命周期。A.主節(jié)點B.從節(jié)點C.上下文D.執(zhí)行器7、以下（D ）不是Spark的主要組件。A.DAGSchedulerB.TaskSchedulerSparkContextD.MultiScheduler8、Spark 組件中，CIusterManager 是（BA.從節(jié)點B.主節(jié)點C.執(zhí)行器D.上下文9、關于Spark中的RDD說法不正確的選項是（B ）。A.是彈性分布式數(shù)據(jù)集B.是可讀可寫分區(qū)的集合C.存在容錯機制D.是Spark中最基本的數(shù)據(jù)抽象10、G

3、raphX的BSP計算模型中，一個超步中的內容不包括（C ）。A.計算B.消息傳遞C.緩存D.整體同步點填空題1、內存計算主要用于處理（數(shù)據(jù)密集型）的計算任務，尤其是數(shù)據(jù)量極大且需要實時分析處理的應用。2、Ignite是一個可擴展的、(容錯性好的)分布式內存計算平臺。3、RDD通過一種名為( 血統(tǒng) )的容錯機制進行錯誤的時的數(shù)據(jù)恢更。4、數(shù)據(jù)分析棧 BDAS 包括( Spark SQL )、( Spark Streaming )、 (Spark GraphX )、( MLlib )四個局部。5、Spark Streaming是建立在Spark上的( 實時計算 )框架，提供了豐富的API、基 于

4、內存的高速執(zhí)行引擎，用戶可以結合流式、批處理進行交互式查詢應用16.3簡答題|、在硬件、軟件、應用與體系等方面，內存計算有哪些主要特性？ 答：在硬件方面，需要大容量的內存，以便盡量將待處理的數(shù)據(jù)全部存放在內存中，內 存可以是單機內存或分布式內存，且內存要足夠大。在軟件方面，需要有良好的編程模型和編程接口。在應用方面，主要面向數(shù)據(jù)密集型應用，數(shù)據(jù)規(guī)模大、對實時處理性能要求高。在體系方面，需要支持并行處理數(shù)據(jù)。2、請與MapReduce相比，Spark的優(yōu)勢有哪些？ 答：中間結果可輸出。基于M卯Reduce的計算模型會將中間結果序列化到磁盤上，而 Spark將執(zhí)行模型抽象為通用的有向無環(huán)圖，可以將

5、中間結果緩存在內存中。數(shù)據(jù)格式和內存布局.Spark抽象出分布式內存存儲結構RDD,用于進行數(shù)據(jù)存儲。 Spark能夠控制數(shù)據(jù)在不同節(jié)點上的分區(qū)，用戶可以自定義分區(qū)策略。執(zhí)行策略。M叩Reduce在數(shù)據(jù)Shuffle之前總是花費大量時間來排序，Spark支持 基于Hash的分布式聚合，Spark默認Shuffle已經(jīng)改為基于排序的方式。任務調度的開銷。當MapReduce上不同的作業(yè)在同一個節(jié)點運行時，會各自啟動 一個Java虛擬機(Java Virtual Machine, JVM)； Spark同一節(jié)點的所有任務都可 以在一個JVM上運行。編程模型。MapReduce僅僅提供了 M叩和Re

6、duce兩個計算原語，需要將數(shù)據(jù)處理 操作轉化為Map和Reduce操作，在一定程度增加了編程難度：Spark那么提供了豐 富的輸出處理算子，實現(xiàn)了分布式大數(shù)據(jù)處理的高層次抽象。統(tǒng)一數(shù)據(jù)處理。Spark框架為批處理(Spark Core)、交互式(Spark SQL)流式(Spark Streaming機器學習(MLlib)、圖計算(GraphX)等計算任務提供一個統(tǒng)一的數(shù) 據(jù)處理平臺，各組件間可以共享數(shù)據(jù)。3、請描述Pregel計算模型的缺點或局限。答：在圖的劃分上，采用的是簡單的Hash方式，這樣固然能夠滿足負載均衡，但Hash 方式并不能根據(jù)圖的連通特性進行劃分，導致超步之間的消息傳遞開

7、銷影響性能。簡單的Checkpoint機制只能將狀態(tài)恢復到當前超步的幾個超步之前，要到當前超 步還需要重復計算。BSP計算模型本身有其局限性，整體同步并行對于計算速度快 的Worker,長期等待的問題無法解決。由于Pregel目前的計算狀態(tài)都是常駐內存的，對于規(guī)模繼續(xù)增大的圖處理可能會 導致內存缺乏。4、請簡要描述函數(shù)式編程中尾遞歸的含義。答：尾遞歸是遞歸的一種優(yōu)化方法。遞歸的空間效率很低，當遞歸深度很深時，容易產(chǎn)生棧 溢出的情況。尾遞歸就是將遞歸語句寫在函數(shù)的最底部，這樣在每次調用尾遞歸時，就不需 要保存當前狀態(tài)值，可以直接把當前的狀態(tài)值傳遞給下次一次調用，然后清空當前的狀態(tài)。 占用的?？臻g

8、就是常量值，不會出現(xiàn)棧溢出的情況。16.4解答題1、根據(jù)用戶手機上網(wǎng)的行為記錄，基于Spark設計程序來分別統(tǒng)計不同設備的用戶使用的 上行總流量以及卜.行總流量。其中，數(shù)據(jù)記錄的字段描述如卜。序號字段字段類型描述0reportTimelong記錄報告時間戳1deviceldString手機號碼2upPackNumlong上行數(shù)據(jù)包數(shù)，單位：個3down Pack Numlong下行數(shù)據(jù)包總數(shù)，單位：個數(shù)據(jù)文件的具體內容（一局部）如下：77e3c9e 1811d4fb291 d()d9bbd456bb4b 7997611496(92ecfBe076d44b89f2d070fb Idf7l9795

9、291890923de7d6514fl d4ac790c630fa63d8d0be5702950228dd382d2a20464a74hbb7414e429ae452 2042893467bb2956150d6741 df875fbcca76ae9e7c5199457706答：Stcpl：將SparkConf封裝在一個類中。import org.apache.spark.SparkConf;import org.apache.spark.api.java.JavaSparkContext;public class ComniSparkContext public static JavaSpar

10、kContext getsc()SparkConfsparkConfnewSparkConf().sctAppNamc(CoinmSparkContcxt).sctMastcr(local); return new JavaSparkContext(sparkConf);)Step2：自定義數(shù)據(jù)類型Loginfo import java.io.Serializable;public class Loglnfb implements Serializable private long timeStamp;private long upTraffic;private long downTraffi

11、c;public long getTimeStampO return timeSiamp;public void setTimeStame(long timeStame) this.timcStamp = timeStame;)public long ge(UpTrafflc() return upTraffic;)public void sctUpTraffic(long upTraffic) Ihis.upTraffic = upTraffic;public long getDownTrafficO return downTraffic;)public void se(DownTraffl

12、c(long downTrafflc) this.downTraffic = downTraffic;Ipublic Loglnfo()public Loglnfo(long timeStame, long upTraffic, long downTraffic) this.timeStainp = timeStame;this.upTraffic = upTraffic; this.downTraffic = downTraffic;Step3 ：自定義key排序類LogSort import scala.Serializable;import scala.niath.Ordered;pub

13、lic class LogSort extends Loginfb implements Ordered , Serializable private long timeStamp;private long upTraffic;private long downTraffic;Overridepublic long ge(TiineStamp() return timeStamp;)public void selTimcStamp(long timcStamp) this.timeStamp = limeStamp;IOverridepublic long gctUpTrafficO retu

14、rn upTraffic;Overridepublic void setUpTraffic(long upTraffic) this.upTraffic = upTraffic;)Ovcrridcpublic long ge(DovvnTraffic() return downTraffic;I()verridcpublic void setDownTraffic(long downTraffic) this.downTraffic = downTraffic;)public LogSorl()public LogSort(long timeStamp. long upTraffic, lon

15、g downTraffic) this.timeStamp = timcStamp;this.upTraffic = upTraffic;this.downTraffic = downTraffic;public in( compare(LogSort (hat) int comp = Long.valueOf(this.getUpTraffic() pareTo(that.getUpTraffic();if (comp = 0)comp = Long.valueOf(this.getDownTraffic() parelb(tha(.getDownTraffic(); if (comp =

16、0)comp = Long.valueOf(this.getTinieStamp() pareTo(that.getTimeStanip();)return comp;)public boolean $lcss(LogSort that) return false;Ipublic boolean $greater(LogSort that) return false;)public boolean $less$eq(LogSort that) return false;)public boolean $grcatcr$cq(LogSort that) return false;)public

17、int compareTo(LogSort that) int comp = Long.valueOf(his.getUpTraffic() pareTo(that.getUpTraff,ic();if (comp = 0)comp = Long.valueOf(this.getDovnTraffic().coinpareTo(that.getDownTraffic(); )if (comp = 0)comp = Long.valueOf(this.getTimeSlamp().coinpareTo(that.getTimeS(amp();)return comp;)Stcp4：定義主類imp

18、ort com.kfk.spark inon inSparkContext;import org.apache.spark.api.java.JavaPairRDD;import org.apache.spark.api.java.JavaRDD;import org.apache.spark.api.java.JavaSparkContext;import org.apache.spark.api.java.function.Function2;import org.apache.spark.api.java.function.PairFunction;import scala.Tuple2

19、;import java.util.List;public class LogApp public static JavaPairRDD mapToPairValues(JavaRDD rdd)rdd .mapToPa ir(newrdd .mapToPa ir(newJavaPairRDD mapToPairRddPairFunction() public Tuple2 call(String line) throws Exception long timeSiamp = Long.parseLong(line.split(,t,)0);String diviceld = String. v

20、alueOf(line.spIit(t) 1 );long upTraffic = Long.parseLong(line.split(Mt)2);long downTraffic = Long.parseLong(line.split(t)3);Loginfo loglnfb = new LogInfo(timeStamp,upTraffic,downTraffic);return new Tuple2(diviceld,loginfo);)；return mapToPairRdd;publicstaticJavaPairRDDreduceByKeyValues(JavaPairRDD ma

21、pPairRdd)JavaPairRDD reduceByKcyRdd = mapPairRdd.reduceByKcy(newFunc(ion2() public Loginfo calKLoglnfb vl, Loglnfb v2) throws Exception long timeSiamp = Math.min(v 1 .getTimeStampO, v2.getTimeStamp();long upTraffic = vl .getUpTraffic() + v2.getUpTraffic();long downTraffic = v I .getDownTraffic() + v

22、2.getDownTraffic();Loginfo loginfo = new Loglnfb();loglnfo.setTimeStame(timeStamp);loglnfo.sctUpTraffic(upTraffic);loglnfo.setDownTraffic(downTrafnc);return loglnfb;0；return reduceByKeyRdd;publicpublicstaticJavaPairRDDmapToPairSortValucs(JavaPairRDD aggrcgatcByKcyRdd) JavaPairRDDmapToPairSortRdd=agg

23、regateByKeyRdd.mapToPair(new PairFunctionTuple2, LogSort, String() | public Tuple2 call(Tuple2 stringLoginfoTuple2) throws Exception (String diviceld = stringLogInfoTuplc2._l;long timeStamp = stringLogInfbTuple2._2.getTimeStamp();long upTraffic = stringLogInfoTuple2._2.getUpTraffic();long downTraffi

24、c = stringLogInfbTuple2._2.getDownTraffic();LogSort logSort = new LogSort(timcStamp,upTraffic,downTraffic);return new Tuplc2(logSort,diviceld);0；return mapToPairSortRdd;)public static void main(String args) JavaSparkContcxt sc = CommSparkContcxt.gctsc();JavaRDD rdd = sc.textFile( 文件路徑);/ rdd map() -

25、 JavaPairRDD niapToPairRdd = m叩ToPairValues(rdd); TOC o 1-5 h z /niapToPairRddrcduccByKeyO-JavaPairRDDreduceByKeyRdd=reduceByKeyValues(inapToPairRdd);/ reduceByKeyRdd map() - JavaPairRDDmapToPairSortRdd=mapToPairSortValucs(rcduccByKcyRdd);/ sortByKeyJavaPairRDDsortByKey Values=mapToPairSortRdd.sortByKey(false);/TopNListTuple2LogSort,String sortKeyList = sortByKeyValues.take( 10);for (Tuplc2 logSortStri