Spark大數(shù)據(jù)技術(shù)平臺(tái)詳解

1、技術(shù)創(chuàng)新，變革未來(lái)Spark大數(shù)據(jù)技術(shù)平臺(tái)詳解內(nèi)容Spark介紹Spark體系結(jié)構(gòu) Spark示例Spark邏輯執(zhí)行計(jì)劃Spark物理執(zhí)行計(jì)劃大數(shù)據(jù)處理常見(jiàn)場(chǎng)景批量處理（Batch Processing）側(cè)重于處理海量數(shù)據(jù)，處理速度可忍受，時(shí) 間可能在數(shù)十分鐘到數(shù)小時(shí)MR歷史數(shù)據(jù)交互式查詢(xún)（Interactive Query）時(shí)間在數(shù)秒到數(shù)十分鐘之間Presto實(shí)時(shí)流數(shù)據(jù)處理（Streaming Processing）通常在數(shù)十毫秒到數(shù)秒之間StormSpark 是什么大數(shù)據(jù)處理一站式解決平臺(tái)??？基于內(nèi)存分布式并行計(jì)算框架加州大學(xué)伯克利分校AMP實(shí)驗(yàn)室開(kāi)發(fā)Spark VS Hadoop最大的誤

2、區(qū)：Spark替代HadoopSpark借鑒MR發(fā)展而來(lái)，繼承分布式并行 計(jì)算優(yōu)點(diǎn)并改進(jìn)了MR缺陷MR計(jì)算結(jié)果需落地磁盤(pán)，Spark中間數(shù)據(jù) 放在內(nèi)存中MR只提供Map和Reduce兩種操作， Spark提供了Transformations和Actions兩 大類(lèi)操作Spark VS MRSpark VS MRSpark VS MRSpark在hadoop生態(tài)體系中的位置Spark 生態(tài)系統(tǒng)Spark 支持語(yǔ)言ScalaJavaPythonRSpark 體系結(jié)構(gòu)Spark 體系結(jié)構(gòu)Client 可以在集群內(nèi)也可以在集群外1個(gè) application 包含1個(gè) driver 和多個(gè)executo

3、r1個(gè) executor 里面運(yùn)行的 tasks 都屬于同1個(gè)applicationSpark on YarnYarn Client 和 Yarn Cluster區(qū)別Spark 示例Spark安裝交互式開(kāi)發(fā)./spark-shell -master local2./spark-shell -master yarn-clientIDE開(kāi)發(fā)WordCountSpark 示例val textFile = sc.textFile(/spark/README.md)textFile.count textFile.firstval linesWithSpark = textFile.filter(line

4、 = line.contains(spark)linesWithSpark.countSpark 示例./bin/spark-submit-name -class -master yarn-cluster-num-executors 2-executor-memory 4G-executor-cores 2-driver-memory 1G*.jar spark-submit提交程序，會(huì)讀取配置文件conf/spark-defaults.conf命令行中參數(shù)優(yōu)先級(jí)高于配置文件程序中SparkConf對(duì)象的設(shè)置優(yōu)先級(jí)最高Spark 關(guān)鍵術(shù)語(yǔ)術(shù)語(yǔ)含義Application包含了一個(gè)driver p

5、rogram 和 集群中多個(gè)的executorDriver Program運(yùn)行在Driver進(jìn)程，執(zhí)行Application的main()函數(shù)并且創(chuàng)建SparkContext，Driver進(jìn)程占用一定數(shù)量的內(nèi)存和CPU coreCluster Manager如Standalone、Mesos、YarnWorker Node集群中任何運(yùn)行Executor進(jìn)程的節(jié)點(diǎn)Executor某Application運(yùn)行在worker node上的一個(gè)進(jìn)程，該進(jìn)程負(fù)責(zé) 運(yùn)行Task，Application之間不共用，每個(gè)Executor進(jìn)程都占有 一定數(shù)量的內(nèi)存和CPU coreJob包含多個(gè)Task組成的并

6、行計(jì)算，往往由Spark Action生成Stage每個(gè)Job會(huì)被拆分很多組task，每組任務(wù)被稱(chēng)為Stage，也可稱(chēng)TaskSet，由Shuffle和Action生成Task最小的計(jì)算單元，執(zhí)行一模一樣的計(jì)算邏輯（我們寫(xiě)的某個(gè)代 碼段），只是每個(gè)task處理的數(shù)據(jù)不同，在某個(gè)executor執(zhí)行Spark執(zhí)行計(jì)劃MR中，用戶(hù)直接面對(duì) task，mapper 和 reducer 的職責(zé)分明：一個(gè)進(jìn)行分塊處理， 一個(gè)進(jìn)行 aggregateMR中數(shù)據(jù)依賴(lài)和task執(zhí)行流程是一致且固定 的，只需實(shí)現(xiàn) map 和 reduce函數(shù)即可Spark中數(shù)據(jù)依賴(lài)可以非常靈活Spark將數(shù)據(jù)依賴(lài)和具體 tas

7、k的執(zhí)行流程分 開(kāi)，通過(guò)將邏輯執(zhí)行計(jì)劃轉(zhuǎn)換成物理執(zhí)行計(jì) 劃Spark執(zhí)行計(jì)劃首先建立邏輯執(zhí)行計(jì)劃（RDD依賴(lài)關(guān)系）把邏輯執(zhí)行計(jì)劃轉(zhuǎn)換為物理執(zhí)行計(jì)劃（劃分Job，形成Stage和Task）執(zhí)行TaskSpark邏輯執(zhí)行計(jì)劃邏輯執(zhí)行計(jì)劃描述了數(shù)據(jù)流：經(jīng)過(guò)哪些 transformation，中間生成哪些 RDD 及 RDD 之間的依賴(lài)關(guān)系邏輯執(zhí)行計(jì)劃表示數(shù)據(jù)上的依賴(lài)關(guān)系，不是task的執(zhí)行流程Spark邏輯執(zhí)行計(jì)劃Spark邏輯執(zhí)行計(jì)劃從數(shù)據(jù)源讀取數(shù)據(jù)創(chuàng)建最初的 RDD，如本地 file、 內(nèi)存數(shù)據(jù)結(jié)構(gòu)（parallelize）、 HDFS、HBase等對(duì) RDD 進(jìn)行一系列的 transforma

8、tion操作對(duì)final RDD 進(jìn)行 action 操作，每個(gè) partition 計(jì)算 后產(chǎn)生結(jié)果 result將 result 回送到 driver 端，進(jìn)行最后的 f(listresult)計(jì)算如count() 實(shí)際包含了action() 和 sum() 兩步典型RDD依賴(lài)groupByKeyval rdd = sc.parallelize(List(1, a),(2,b),(3,c),(4, d),(5,e),(3,f),(2, g),(1,h), 3)val rdd2 = rdd.groupByKey(2)groupByKey通過(guò) shuffle將 Key 相同的records 聚

9、合在一起groupByKey 沒(méi)有在map 端進(jìn)行 combine典型RDD依賴(lài)reduceByKeyval rdd = sc.parallelize(List( A,1),(B,1),(C,1),( B,1),(C,1),(D,1), (C,1),(A,1), 3)val rdd2 = rdd.reduceByKey(_+_, 2)reduceyByKey相當(dāng)于MR中的reduce， 數(shù)據(jù)流一樣reduceyByKey在 map 端開(kāi)啟 combine典型RDD依賴(lài)joinRDD依賴(lài)總結(jié)Each box is an RDD, with partitions shown as shaded r

10、ectanglesRDD依賴(lài)總結(jié)窄依賴(lài)（narrow dependencies）子RDD的每個(gè)分區(qū)依賴(lài)于常數(shù)個(gè)父分區(qū)（即 與數(shù)據(jù)規(guī)模無(wú)關(guān)）允許父分區(qū)以流水線的方式找到子分區(qū)map產(chǎn)生窄依賴(lài)寬依賴(lài)（wide dependencies）子RDD的每個(gè)分區(qū)依賴(lài)于所有父RDD分區(qū)類(lèi)似于MR的shuffle過(guò)程Join產(chǎn)生寬依賴(lài)（除非父RDD被哈希分區(qū)）Spark物理執(zhí)行計(jì)劃import org.apache.spark.HashPartitionerval rddA = sc.parallelize(List(5,e),(3,a),(2,b),(1,c),3) val rddAHash = rddA.

11、partitionBy(new HashPartitioner(3)val rddB = sc.parallelize(List(1,A),(2,B),(3,C),(4,D),2) val rddBMap = rddB.map(x = (x._1, x._2 + 1)val rddC = sc.parallelize(List(1,X),(2,Y),2) val rddBUnionC = rddBMap.union(rddC)val result = rddAHash.join(rddBUnionC ,3) result.toDebugStringSpark物理執(zhí)行計(jì)劃Spark物理執(zhí)行計(jì)劃R

12、DD中每個(gè)Partition是獨(dú)立的Spark物理執(zhí)行計(jì)劃從后往前計(jì)算，根據(jù)窄依賴(lài)和寬依賴(lài)劃分stagestage 里task 數(shù)目取決于最后一個(gè) RDD 中的 partition 個(gè)數(shù)Spark物理執(zhí)行計(jì)劃1個(gè)Application 可以包含多個(gè) job1個(gè) job可以包含多個(gè) stage1個(gè) stage可以包含多個(gè) taskSpark物理執(zhí)行計(jì)劃從后往前推，需要哪個(gè) partition 就計(jì)算哪個(gè) partition， 如果 partition 里面沒(méi)有數(shù)據(jù)，就繼續(xù)向前推對(duì)于沒(méi)有 parent stage 的 stage，該 stage 最左邊的RDD 是可以立即計(jì)算的對(duì)于有 parent

13、 stage 的 stage，先等著所有 parent stages 中 final RDD 中數(shù)據(jù)計(jì)算好，然后shuffle實(shí)際計(jì)算時(shí)數(shù)據(jù)從前到后流動(dòng)，計(jì)算出第一個(gè) record流動(dòng)到不能再流動(dòng)后，再計(jì)算下一個(gè) record并不是當(dāng)前 RDD 的 partition 中所有 records 計(jì)算后再 整體向后流動(dòng)Spark物理執(zhí)行計(jì)劃執(zhí)行count，先在每個(gè) partition上執(zhí)行 count， 然后執(zhí)行結(jié)果被發(fā)送到 driver，最后在 driver 端進(jìn)行 sum執(zhí)行take類(lèi)似Spark應(yīng)用調(diào)度RDD Objectsrdd1.join(rdd2).groupBy().filter()build operator DAGDAGSchedulersplit graph into stages of tasks submit each stage as readyDAGTaskSchedulerTaskSetlaunch tasks via cluster manager retry failed or straggling tasksCluster