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

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

3、r1個 executor 里面運行的 tasks 都屬于同1個applicationSpark on YarnYarn Client 和 Yarn Cluster區(qū)別Spark 示例Spark安裝交互式開發(fā)./spark-shell -master local2./spark-shell -master yarn-clientIDE開發(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提交程序，會讀取配置文件conf/spark-defaults.conf命令行中參數(shù)優(yōu)先級高于配置文件程序中SparkConf對象的設置優(yōu)先級最高Spark 關鍵術語術語含義Application包含了一個driver p

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

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

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

8、tion操作對final RDD 進行 action 操作，每個 partition 計算 后產(chǎn)生結果 result將 result 回送到 driver 端，進行最后的 f(listresult)計算如count() 實際包含了action() 和 sum() 兩步典型RDD依賴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通過 shuffle將 Key 相同的records 聚

9、合在一起groupByKey 沒有在map 端進行 combine典型RDD依賴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相當于MR中的reduce， 數(shù)據(jù)流一樣reduceyByKey在 map 端開啟 combine典型RDD依賴joinRDD依賴總結Each box is an RDD, with partitions shown as shaded r

10、ectanglesRDD依賴總結窄依賴（narrow dependencies）子RDD的每個分區(qū)依賴于常數(shù)個父分區(qū)（即 與數(shù)據(jù)規(guī)模無關）允許父分區(qū)以流水線的方式找到子分區(qū)map產(chǎn)生窄依賴寬依賴（wide dependencies）子RDD的每個分區(qū)依賴于所有父RDD分區(qū)類似于MR的shuffle過程Join產(chǎn)生寬依賴（除非父RDD被哈希分區(qū)）Spark物理執(zhí)行計劃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í)行計劃Spark物理執(zhí)行計劃R

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

13、 stage 的 stage，先等著所有 parent stages 中 final RDD 中數(shù)據(jù)計算好，然后shuffle實際計算時數(shù)據(jù)從前到后流動，計算出第一個 record流動到不能再流動后，再計算下一個 record并不是當前 RDD 的 partition 中所有 records 計算后再 整體向后流動Spark物理執(zhí)行計劃執(zhí)行count，先在每個 partition上執(zhí)行 count， 然后執(zhí)行結果被發(fā)送到 driver，最后在 driver 端進行 sum執(zhí)行take類似Spark應用調(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