深入淺出spark.pptx

深入淺出Spark,2015年8月 研究院 王鳳,目錄,Spark簡介,核心模塊的實(shí)現(xiàn),Spark應(yīng)用,Spark與Hadoop的區(qū)別與聯(lián)系,Spark總體流程,Spark應(yīng)用庫,大數(shù)據(jù)的電花火石,什么是Spark,Apache Sparkis a fast and general engine for large-scale data processing,Spark是UC Berkeley AMPLab開發(fā)的是一種計(jì)算框架，分布式資源工作交由集群管理軟件（Mesos、YARN）,a MapReduce-like cluster computing framework designed for low-latency iterative jobs and interactive use from an interpreter,Spark的發(fā)展歷程,Spark代碼貢獻(xiàn)者每個(gè)月的增長曲線,2009：Spark誕生于伯克利大學(xué) AMPLab 2010：開源 2013.6：Apache孵化器項(xiàng)目 2014.2：Apache頂級(jí)項(xiàng)目 目前為止，發(fā)布的最新版本為Spark1.4.1 Spark在最近6年內(nèi)發(fā)展迅速，相較于其他大數(shù)據(jù)平臺(tái)或框架而言，Spark的代碼庫最為活躍。,/,截止2015年6月 Spark的Contributor比2014年漲了3倍，達(dá)到730人； 總代碼行數(shù)也比2014年漲了2倍多，達(dá)到40萬行 Spark應(yīng)用也越來越廣泛，最大的集群來自騰訊8000個(gè)節(jié)點(diǎn)，單個(gè)Job最大分別是阿里巴巴和Databricks1PB,Spark特點(diǎn),Logistic regression in Hadoop and Spark,高效,提供Cache機(jī)制來支持需要反復(fù)迭代的計(jì)算或者多次數(shù)據(jù)共享，減少數(shù)據(jù)讀取的IO開銷。 與Hadoop的MapReduce相比，Spark基于內(nèi)存的運(yùn)算比MR要快100倍；而基于硬盤的運(yùn)算也要快10倍！,易用,Spark提供廣泛的數(shù)據(jù)集操作類型（20+種），不像Hadoop只提供了Map和Reduce兩種操作。 Spark支持Java，Python和Scala API，支持交互式的Python和Scala的shell。,與Hadoop無縫銜接,Spark可以使用YARN作為它的集群管理器 讀取HDFS,HBase等一切Hadoop的數(shù)據(jù),先進(jìn)架構(gòu),Spark采用Scala語言編寫，底層采用了actor model的akka作為通訊框架，代碼十分簡潔高效。 基于DAG圖的執(zhí)行引擎，減少多次計(jì)算之間中間結(jié)果寫到Hdfs的開銷。 建立在統(tǒng)一抽象的RDD（分布式內(nèi)存抽象）之上,使得它可以以基本一致的方式應(yīng)對(duì)不同的大數(shù)據(jù)處理場景。,提供整體 解決方案,以其RDD模型的強(qiáng)大表現(xiàn)能力，逐漸形成了一套自己的生態(tài)圈，提供了full-stack的解決方案。 主要包括Spark內(nèi)存中批處理，Spark SQL交互式查詢，Spark Streaming流式計(jì)算， GraphX和MLlib提供的常用圖計(jì)算和機(jī)器學(xué)習(xí)算法。,One Stack to rule them all,Spark整體架構(gòu),Spark SQL,Spark Streaming,GraphX (graph),MLlib (machine learning),Mesos,YARN,S3,Spark,Standalone,存儲(chǔ)層,資源調(diào)度層,計(jì)算層,工具層,Cassandra,Hbase,Tachyon,HDFS,Spark提供了多種高級(jí)工具： Shark SQL應(yīng)用于即席查詢（Ad-hoc query）、Spark Streaming應(yīng)用于流式計(jì)算、 MLlib應(yīng)用于機(jī)器學(xué)習(xí)、GraphX應(yīng)用于圖處理。 Spark可以基于自帶的standalone集群管理器獨(dú)立運(yùn)行，也可以部署在Apache Mesos 和 Hadoop YARN 等集群管理器上運(yùn)行。 Spark可以訪問存儲(chǔ)在HDFS、 Hbase、Cassandra、Amazon S3、本地文件系統(tǒng)等等上的數(shù)據(jù)，Spark支持文本文件，序列文件，以及任何Hadoop的InputFormat。,目錄,Spark簡介,核心模塊的實(shí)現(xiàn),Spark應(yīng)用,Spark與Hadoop的區(qū)別與聯(lián)系,Spark總體流程,Spark應(yīng)用庫,Spark組件 Driver Program (驅(qū)動(dòng)程序) 是Spark 的核心組件 構(gòu)建SparkContext(Spark應(yīng)用的入口,它負(fù)責(zé)和整個(gè)集群的交互，創(chuàng)建需要的變量，還包含集群的配置信息等) 將用戶提交的job轉(zhuǎn)換為DAG圖(類似數(shù)據(jù)處理的流程圖) 根據(jù)策略將DAG圖劃分為多個(gè)stage，根據(jù)分區(qū)從而生成一系列tasks 根據(jù)tasks要求向資源管理器申請(qǐng)資源 提交任務(wù)并檢測任務(wù)狀態(tài) Executor 真正執(zhí)行task的單元，一個(gè)Worker Node上可以有多個(gè)Executor,Spark任務(wù) 應(yīng)用程序：由一個(gè)driver program和多個(gè)job構(gòu)成； job：由多個(gè)stage組成； stage：對(duì)應(yīng)一個(gè)taskset， taskset：對(duì)應(yīng)一組關(guān)聯(lián)的相互之間沒有shuffle依賴關(guān)系的task組成。 task：任務(wù)最小的工作單元,資源管理組件 YARN（通用） Master/Slave結(jié)構(gòu) Resource Manager(RM) 全局的資源管理器，負(fù)責(zé)系統(tǒng)的資源管理和分配 Node Manager(NM) 每個(gè)節(jié)點(diǎn)上的資源和任務(wù)管理器 ApplicationMaster(AM) 每個(gè)應(yīng)用程序都有一個(gè)，負(fù)責(zé)任務(wù)的調(diào)度和監(jiān)視，并與RM調(diào)度器協(xié)商為任務(wù)獲取資源。 Standalone（Spark自帶） Master/Slave結(jié)構(gòu) Master 類似于YARN的RM Worker 類似于YARN的NM,Spark核心概念,Spark任務(wù)提供多層分解的概念，Spark組件將用戶的應(yīng)用程序分解為內(nèi)部執(zhí)行任務(wù)并提供執(zhí)行容器，資源管理為spark組件提供資源管理和調(diào)度。,Spark端到端流程,整體流程 Spark的Driver Program (簡稱Driver)包含用戶的應(yīng)用程序 Driver完成task的解析和生成 Driver向Cluster Manager（集群資源管理器）申請(qǐng)運(yùn)行task需要的資源。 集群資源管理器為task分配滿足要求的節(jié)點(diǎn)，并在節(jié)點(diǎn)按照要求創(chuàng)建Executor 創(chuàng)建的Executor向Driver注冊(cè)。 Driver將spark應(yīng)用程序的代碼和文件傳送給分配的executor executor運(yùn)行task，運(yùn)行完之后將結(jié)果返回給Driver或者寫入HDFS或其他介質(zhì)。,Spark流程,DriveProgram,Driver運(yùn)行在Client,Driver運(yùn)行在Worker,執(zhí)行流程描述： 1.客戶端把作業(yè)發(fā)布到Master 2.Master讓一個(gè)Worker啟動(dòng)Driver，并將作業(yè)推送給Driver 3. Driver進(jìn)程生成一系列task 4.Driver向Master申請(qǐng)資源 5.Master讓調(diào)度的Worker啟動(dòng)Exeuctor 6. Exeuctor啟動(dòng)后向Driver注冊(cè)。 7.Driver將task調(diào)度到Exeuctor執(zhí)行。 8. Executor執(zhí)行結(jié)果寫入文件或返回Driver,執(zhí)行流程描述： 1. .客戶端啟動(dòng)后直接運(yùn)行用戶程序，啟動(dòng)Driver 2. Driver進(jìn)程生成一系列task 3. Driver向Master申請(qǐng)資源 4. Master讓調(diào)度的Worker啟動(dòng)Exeuctor 5. Exeuctor啟動(dòng)后向Driver注冊(cè)。 6. Driver將task調(diào)度到Exeuctor執(zhí)行。 7. Executor執(zhí)行結(jié)果寫入文件或返回Driver,Spark on Standalone,Spark流程,Master和Worker是standalone的角色，Driver和Executor是Spark的角色。 Master負(fù)責(zé)分配資源，分配Driver和Executor，讓W(xué)orker啟動(dòng)driver和executor，只管理到executor層，不涉及任務(wù)； Driver負(fù)責(zé)生成task，并與executor通信，進(jìn)行任務(wù)的調(diào)度和結(jié)果跟蹤，不涉及資源。,Spark on YARN,Yarn Client,Yarn Cluster,Driver運(yùn)行在AM(Application Master)： 1.客戶端把作業(yè)發(fā)布到Y(jié)ARN 2. RM為該job在某個(gè)NM上分配一個(gè)AM，NM啟動(dòng)AppMaster，App Master啟動(dòng)Driver。 3.AM啟動(dòng)后完成初始化作業(yè)。Driver生成一系列Task。 4.AM向RM申請(qǐng)資源，RM返回Executor信息。 5.AM通過RPC啟動(dòng)相應(yīng)的SparkExecutor。 6.Driver向Executor分配task。 7. Executor執(zhí)行結(jié)果寫入文件或返回Driver,Driver運(yùn)行在Cleint ： 1. .客戶端啟動(dòng)后直接運(yùn)行用戶程序，直接啟動(dòng)Driver。 2. Driver初始化并生成一系列Task 3.客戶端將App Commands發(fā)布到Y(jié)ARN。 4. RM為該job在某個(gè)NM上分配一個(gè)AM 。 5.AM向RM申請(qǐng)資源，RM返回Executor信息。 6.AM通過RPC啟動(dòng)相應(yīng)的SparkExecutor。 7.Exeuctor啟動(dòng)后向Driver注冊(cè)。 8.Driver向Executor分配task。 9.Executor執(zhí)行結(jié)果寫入文件或返回Driver,Spark流程,與Standalone類似，不過YARN架構(gòu)更通用，可以在同一集群部署Hadoop、Spark等。 RM、NM和AM是YARN的角色，Driver和Executor是Spark的角色。 RM負(fù)責(zé)資源分配，分配App Master和Executor；AM負(fù)責(zé)向RM申請(qǐng)資源，啟動(dòng)Driver和啟動(dòng)Executor Driver負(fù)責(zé)task的生成，task的調(diào)度和結(jié)果跟蹤,目錄,Spark簡介,核心模塊的實(shí)現(xiàn),Spark應(yīng)用,Spark與Hadoop的區(qū)別與聯(lián)系,Spark總體流程,Spark應(yīng)用庫,核心模塊,核心模塊的實(shí)現(xiàn),Spark的關(guān)鍵運(yùn)算組件,Spark的核心組件包括RDD、Scheduler、Storage、Shuffle四部分： RDD是Spark最核心最精髓的部分，spark將所有數(shù)據(jù)都抽象成RDD。 Scheduler是Spark的調(diào)度機(jī)制，分為DAGScheduler和TaskScheduler。 Storage模塊主要管理緩存后的RDD、shuffle中間結(jié)果數(shù)據(jù)和broadcast數(shù)據(jù) Shuffle分為Hash方式和Sort方式，兩種方式的shuffle中間數(shù)據(jù)都寫本地盤,核心模塊RDD整體介紹,RDD是什么？,RDDs are fault-tolerant, parallel data structures that let users explicitly persist intermediate results in memory, control their partitioning to optimize data placement, and manipulate them using a rich set of operators.,Resilient Distributed Datasets,RDD是Spark的基石，也是Spark的靈魂。 RDD是彈性分布式數(shù)據(jù)集，是只讀的分區(qū)記錄集合。 每個(gè)RDD有5個(gè)主要的屬性： 一組分片(Partition)：數(shù)據(jù)集的最基本組成單位 一個(gè)計(jì)算每個(gè)分片的函數(shù)：對(duì)于給定的數(shù)據(jù)集，需要做哪些計(jì)算 依賴（Dependencies）：RDD的依賴關(guān)系，描述了RDD之間的lineage preferredLocations（可選）：對(duì)于data partition的位置偏好 partitioner（可選） - 對(duì)于計(jì)算出來的數(shù)據(jù)結(jié)果如何分發(fā),核心模塊的實(shí)現(xiàn),作用于RDD上的Operation分為轉(zhuǎn)換(transformantion)和動(dòng)作(action)。 Spark中的所有“轉(zhuǎn)換”都是惰性的，在執(zhí)行“轉(zhuǎn)換”操作，并不會(huì)提交Job，只有在執(zhí)行“動(dòng)作”操作，所有operation才會(huì)被提交到cluster中真正的被執(zhí)行。這樣可以大大提升系統(tǒng)的性能。,RDD擁有的操作比MR豐富的多，不僅僅包括Map、Reduce操作，還包括右圖的filter、sort、join、save、count等操作，所以Spark比MR更容易方便完成更復(fù)雜的任務(wù)。,RDD部分Operation,核心模塊RDD操作,核心模塊的實(shí)現(xiàn),核心模塊RDD持久化,默認(rèn)情況下，每一個(gè)轉(zhuǎn)換過的RDD都會(huì)在它之上執(zhí)行一個(gè)動(dòng)作時(shí)被重新計(jì)算。 如果rdd只被使用一次或者很少次，不需要持久化。如果rdd被重復(fù)使用或者計(jì)算其代價(jià)很高，才考慮持久化。另外，shuffle后生成的rdd盡量持久化，因?yàn)閟huffle代價(jià)太高。,是否持久化,使用 persist方法(或者cache方法)，持久化一個(gè)RDD在內(nèi)存或磁盤中。 cache()過程是將RDD persist在內(nèi)存里，persist()操作可以為RDD指定StorageLevel。,持久化方法,RDD被緩存后，Spark將會(huì)在集群中，保存相關(guān)元數(shù)據(jù)，下次查詢這個(gè)RDD時(shí)，它將能更快速訪問，不需要計(jì)算。 如果持久化無謂的RDD，會(huì)浪費(fèi)內(nèi)存（或硬盤）空間，反而降低系統(tǒng)整體性能,持久化效果,核心模塊的實(shí)現(xiàn),核心模塊RDD依賴關(guān)系,RDD只能基于在穩(wěn)定物理存儲(chǔ)中的數(shù)據(jù)集和其他已有的RDD上執(zhí)行確定性操作來創(chuàng)建。 能從其他RDD通過確定操作創(chuàng)建新的RDD的原因是RDD含有從其他RDD衍生（即計(jì)算）出本RDD的相關(guān)信息（即Lineage） Dependency代表了RDD之間的依賴關(guān)系，即血緣（Lineage），分為窄依賴和寬依賴。,窄依賴： 一個(gè)父RDD最多被一個(gè)子RDD用 在一個(gè)集群節(jié)點(diǎn)上管道式執(zhí)行。 比如map、filter、union等；,寬依賴： 指子RDD的分區(qū)依賴于父RDD的所有分區(qū)，這是因?yàn)閟huffle類操作要求所有父分區(qū)可用。 比如groupByKey、reduceByKey、 sort、partitionBy等；,核心模塊的實(shí)現(xiàn),注意：一個(gè)RDD對(duì)不同的父節(jié)點(diǎn)可能有不同的依賴方式，可能對(duì)父節(jié)點(diǎn)1是寬依賴，對(duì)父節(jié)點(diǎn)2是窄依賴。,核心模塊RDD依賴關(guān)系,根據(jù)RDD依賴關(guān)系的不同，Spark將每 一個(gè)job分為不同的stage，stage之間 的依賴關(guān)系形成了DAG圖。,窄依賴 Spark將其盡量劃分在同一個(gè)stage中，因?yàn)樗鼈兛梢赃M(jìn)行流水線計(jì)算。,寬依賴 往往意味著shuffle操作，這也是Spark劃分stage的主要邊界。,一個(gè)Stage的開始就是從外部存儲(chǔ)或者shuffle結(jié)果中讀取數(shù)據(jù)；一個(gè)Stage的結(jié)束就是發(fā)生shuffle或者生成結(jié)果時(shí)。 由于rdd F是rdd G的寬依賴，所以將F與G分別劃分到不同的stage，但是B是G的窄依賴（B的每個(gè)分區(qū)只被使用一次），所以B與G還保持在一個(gè)stage。 由于rdd A是rdd B的寬依賴，所以A和B劃分在不同的stage 在圖中Stage2中，從map到union都是窄依賴，在一個(gè)stage，這兩步操作可以形成一個(gè)流水線操作，通過map操作生成的partition可以不用等待整個(gè)RDD計(jì)算結(jié)束，而是繼續(xù)進(jìn)行union操作，這樣大大提高了計(jì)算的效率。,核心模塊的實(shí)現(xiàn),核心模塊RDD容錯(cuò)機(jī)制,寬/窄依賴的概念不止用在stage劃分中，對(duì)容錯(cuò)也很有用。,核心模塊的實(shí)現(xiàn),核心模塊Scheduler整體介紹,核心模塊的實(shí)現(xiàn),Scheduler模塊作為Spark最核心的模塊之一，充分體現(xiàn)了Spark與MapReduce的不同之處，體現(xiàn)了Spark DAG思想的精巧和設(shè)計(jì)的優(yōu)雅。 Scheduler模塊分為兩大主要部分，DAGScheduler和TaskScheduler。,核心模塊Scheduler之DAGScheduler,核心模塊的實(shí)現(xiàn),對(duì)應(yīng)的任務(wù),DAGScheduler把一個(gè)spark作業(yè)轉(zhuǎn)換成成stage的DAG（Directed Acyclic Graph有向無環(huán)圖），根據(jù)RDD和stage之間的關(guān)系，找出開銷最小的調(diào)度方法，然后把stage以TaskSet的形式提交給TaskScheduler。,核心模塊Scheduler之TaskScheduler,核心模塊的實(shí)現(xiàn),TaskScheduler模塊用于與DAGScheduler交互，負(fù)責(zé)任務(wù)的具體調(diào)度和運(yùn)行。 任務(wù)調(diào)度模塊基于兩個(gè)Trait：TaskScheduler和 SchedulerBackend,TaskScheduler：定義了任務(wù)調(diào)度模塊的對(duì)外接口（submitTasks等），供DAGScheduler調(diào)用。 TaskSchedulerImpl是TaskScheduler的具體實(shí)現(xiàn)，完成資源與任務(wù)的調(diào)度。 SchedulerBackend封裝了各種backend，用于與底層資源調(diào)度系統(tǒng)交互，配合TaskSchedulerImpl實(shí)現(xiàn)任務(wù)執(zhí)行所需的資源分配。 SchedulableBuilder負(fù)責(zé)taskset的調(diào)度。 TaskSetManager負(fù)責(zé)一個(gè)taskset中task的調(diào)度。,核心模塊TaskScheduler之SchedulableBuilder,SchedulableBuilder主要負(fù)責(zé)TaskSet的調(diào)度。 核心接口是：getSortedTaskSetQueue，該接口返回排序后的TaskSetManager隊(duì)列，該接口供TaskSchedulerImpl調(diào)用。 SchedulableBuilder維護(hù)的是一顆樹，根節(jié)點(diǎn)是rootpool，葉子節(jié)點(diǎn)是TaskSetManager對(duì)象。,核心模塊的實(shí)現(xiàn),核心模塊TaskScheduler之TaskSetManager,TaskSetManager主要負(fù)責(zé)一個(gè)taskset中task的調(diào)度和跟蹤。 核心接口是：resourceOffer，該接口根據(jù)輸入的資源 在taskset內(nèi)部調(diào)度一個(gè)task，主要考慮因素是Locality，該接口供TaskSchedulerImpl調(diào)用。,Locality,NODE_LOCAL,PROCESS_LOCAL,RACK_LOCAL,NO_PREF,ANY,根據(jù)task的preferredLocations得到每個(gè)task的Locality level。 resourceOffe根據(jù)資源和maxLocality（最大寬松的本地化級(jí)別）調(diào)度task。 最終調(diào)度task的allowedLocality是該TaskSet允許的Locality(最大不超過輸入的maxLocality) ，該TaskSet允許的Locality最初默認(rèn)值是最嚴(yán)格本地化級(jí)別。如果lastLaunchTime（最近一次該taskset發(fā)布task的時(shí)間）與當(dāng)前時(shí)間差超時(shí)，會(huì)放寬locality的要求，選擇低一優(yōu)先級(jí)的locality。 在allowedLocality范圍內(nèi)，優(yōu)先調(diào)度更local的task，也就是最好在同個(gè)進(jìn)程里，次好是同個(gè)node(即機(jī)器)上，再次是同機(jī)架。在allowedLocality范圍內(nèi)，在該taskset沒有找到task，那么返回None。（上一層調(diào)用會(huì)繼續(xù)查詢其他taskset是否有滿足指定locality Level的task）,優(yōu)先級(jí)降低,核心模塊的實(shí)現(xiàn),核心模塊TaskScheduler之SchedulerBackend,SchedulerBackend是trait，封裝了多種backend，用于與底層資源調(diào)度系統(tǒng)交互（如mesos/YARN），配合TaskScheduler實(shí)現(xiàn)具體任務(wù)執(zhí)行所需的資源分配。 核心接口是： reviveOffers，與TaskSchedulerImpl交互完成task的Launch。 SchedulerBackend只關(guān)心資源，不關(guān)心task。提交資源供TaskSchedulerImpl分配task。,ReviveOffers的實(shí)現(xiàn) 將空閑資源（freeCore、executor、host）以workerOffer List形式組織。 調(diào)用TaskSchedulerImpl的resourceOffers() ，為workerOffer List空閑資源分配相應(yīng)的task 。 調(diào)用launchTasks，向executorActor發(fā)送LaunchTask消息,核心模塊的實(shí)現(xiàn),核心模塊TaskScheduler之TaskSchedulerImpl,TaskSchedulerImpl實(shí)現(xiàn)了TaskScheduler Trait，實(shí)現(xiàn)了資源和任務(wù)的調(diào)度。 核心接口是： ResourceOffers，根據(jù)提供的資源列表offers，返回滿足條件的tasks，供SchedulerBackend 調(diào)用。 資源和任務(wù)調(diào)度的核心思想：資源驅(qū)動(dòng)。即當(dāng)有空閑資源時(shí)，查看是否有task需要運(yùn)行（遵循Locality）,ResourceOffers的實(shí)現(xiàn) 將輸入的offers( SchedulerBackend返回的 workerOffer List，即可用的空閑資源 )添加到可用的資源類表（加入到不同級(jí)別的資源列表，比如executor、host、rack）。 將offers shuffle一下，調(diào)整空閑資源的順序，避免前面的空閑資源一直被分配任務(wù)。 調(diào)用rootPool的getSortedTaskSetQueue獲取需要運(yùn)行的TaskSet(SchedulableBuilder提供的接口) 對(duì)每個(gè)TaskSet循環(huán)處理（每個(gè)TaskSet都有一個(gè)TaskSetManager），調(diào)用TaskSetManager的resourceOffer給輸入的空閑資源尋找task。 返回為offers查找的tasks,submitTasks流程,核心模塊的實(shí)現(xiàn),核心模塊TaskScheduler,TaskScheduler是trait，用于與DAGScheduler交互，主要負(fù)責(zé)任務(wù)的調(diào)度和運(yùn)行，無具體實(shí)現(xiàn)，僅僅為對(duì)外統(tǒng)一接口。 核心接口是： submitTasks，具體實(shí)現(xiàn)見TaskSchedulerImpl 中的submitTasks 。 接收DAGScheduler的Task請(qǐng)求，分發(fā)Task到集群運(yùn)行并監(jiān)控運(yùn)行狀態(tài)，并將結(jié)果以event的形式匯報(bào)給DAGScheduler 。,SchedulableBuilder,TaskSetManager,SchedulerBackend,Task調(diào)度與低層的資源管理器分離，僅僅根據(jù)提供的資源調(diào)度task，不關(guān)心資源的來源。 資源調(diào)度僅僅關(guān)心資源，與多種不同的資源調(diào)度系統(tǒng)(YARN/MESOS/Standalone)交互，獲得空閑資源。 TaskSchedulerImp在接收到submitTasks時(shí)，從資源調(diào)度系統(tǒng)中獲取到空閑資源，然后將空閑資源提交到到task調(diào)度系統(tǒng)，調(diào)度滿足locality要求的task，并將task launch到executor。,資源調(diào)度,Task調(diào)度,TaskScheduler (TaskSchedulerImpl),核心模塊的實(shí)現(xiàn),核心模塊Storage,核心模塊的實(shí)現(xiàn),Storage模塊主要分為兩層： 通信層：storage模塊采用的是master-slave結(jié)構(gòu)來實(shí)現(xiàn)通信層，master和slave之間傳輸控制信息、狀態(tài)信息，這些都是通過通信層來實(shí)現(xiàn)的。 存儲(chǔ)層：storage模塊需要把數(shù)據(jù)存儲(chǔ)到disk或是memory上面，有可能還需replicate到遠(yuǎn)端，這都是由存儲(chǔ)層來實(shí)現(xiàn)和提供相應(yīng)接口。 Storage模塊提供了統(tǒng)一的操作類BlockManager，外部類與storage模塊打交道都需要通過調(diào)用BlockManager相應(yīng)接口來實(shí)現(xiàn)。 Storage模塊存取的最小單位是數(shù)據(jù)塊(Block)，Block與RDD中的Partition一一對(duì)應(yīng)，所以所有的轉(zhuǎn)換或動(dòng)作操作最終都是對(duì)Block進(jìn)行操作。,數(shù)據(jù)寫入過程分析 1. RDD的iterator調(diào)用CacheManager的getOrCompute函數(shù) 2.CacheManager調(diào)用BlockManager的put接口來寫入數(shù)據(jù) 3.BlockManager根據(jù)輸入的storageLevel來確定是寫內(nèi)存還是寫硬盤 4.通知BlockManagerMaster有新的數(shù)據(jù)寫入，在BlockManagerMaster中保存元數(shù)據(jù) 5.將寫入的數(shù)據(jù)與其它slave worker進(jìn)行同步（一般來說在本機(jī)寫入的數(shù)據(jù)，都會(huì)另先一臺(tái)機(jī)器來進(jìn)行數(shù)據(jù)的備份，即replicanumber=1）,核心模塊Storage,核心模塊的實(shí)現(xiàn),通 信 層,Driver和Executor都有一個(gè)BlockManager，里面都包含了BlockManagerMasterActor和BlockManagerSlaveActor BlockManagerMasterActor類主要負(fù)責(zé)控制消息和狀態(tài)之間的傳遞和處理。傳遞的消息包括Register、StorageStatus、updateBlockInfo、getLocation等，消息處理包括返回或更新block以及executor的元數(shù)據(jù)信息，并調(diào)用BlockManagerSlaveActor(ref )與Executor通信。 BlockManagerSlaveActor類傳遞的消息主要是removeRdd、removeBlock等消息，消息處理主要是對(duì)本executor進(jìn)行rdd和block的刪除操作。,存儲(chǔ) 層,BlockManager包含了DiskStore類和MemoryStore類 DiskStore：每一個(gè)block都被存儲(chǔ)為一個(gè)file，通過計(jì)算block id的hash值將block映射到文件中。 MemoryStore：內(nèi)部維護(hù)了一個(gè)hash map來管理所有的block，以block id為key將block存放到hash map中。,核心模塊Shuffle,核心模塊的實(shí)現(xiàn),hash方式,Sort方式,shuffle不排序，效率高 生成MXR個(gè)shuffle中間文件，一個(gè)分片一個(gè)文件 產(chǎn)生和生成這些中間文件會(huì)產(chǎn)生大量的隨機(jī)IO，磁盤效率低 shuffle時(shí)需要全部數(shù)據(jù)都放在內(nèi)存，對(duì)內(nèi)存消耗大 適合數(shù)據(jù)量能全部放到內(nèi)存，reduce操作不需要排序的場景,shuffle需要排序 生成M個(gè)shuffle中間數(shù)據(jù)文件，一個(gè)Map所有分片放到一個(gè)數(shù)據(jù)文件中，外加一個(gè)索引文件記錄每個(gè)分片在數(shù)據(jù)文件中的偏移量 shuffle能夠借助磁盤（外部排序）處理龐大的數(shù)據(jù)集 數(shù)據(jù)量大于內(nèi)存時(shí)只能使用Sort方式，也適用于Reduce操作需要排序的場景,shuffle 中Map任務(wù)產(chǎn)生的結(jié)果會(huì)根據(jù)所設(shè)置的partitioner算法填充到當(dāng)前執(zhí)行任務(wù)所在機(jī)器的每個(gè)桶中。 Reduce任務(wù)啟動(dòng)時(shí)時(shí)，會(huì)根據(jù)任務(wù)的ID，所依賴的Map任務(wù)ID以及MapStatus從遠(yuǎn)端或本地的BlockManager獲取相應(yīng)的數(shù)據(jù)作為輸入進(jìn)行處理。 Shuffle數(shù)據(jù)必須持久化磁盤，不能緩存在內(nèi)存。,目錄,Spark簡介,核心模塊的實(shí)現(xiàn),Spark應(yīng)用,Spark與Hadoop的區(qū)別與聯(lián)系,Spark總體流程,Spark應(yīng)用庫,應(yīng)用庫整體介紹,Spark應(yīng)用庫,spark之上提供了四種應(yīng)用庫： Spark SQL 是為處理結(jié)構(gòu)化數(shù)據(jù)而設(shè)計(jì)的模塊 Spark Streaming 可以很容易地創(chuàng)建可擴(kuò)展和容錯(cuò)性的流式應(yīng)用 MLlib 是Spark的可擴(kuò)展機(jī)器學(xué)習(xí)庫 GraphX 是Spark的并行圖計(jì)算API,應(yīng)用庫Spark SQL,Spark應(yīng)用庫,集成性 Spark程序可以與SQL查詢無縫銜接,統(tǒng)一的數(shù)據(jù)訪問方式 可以從多種數(shù)據(jù)源加載和查詢數(shù)據(jù),與Hive兼容性 可以在現(xiàn)有的數(shù)據(jù)倉庫上運(yùn)行不做任何修改的Hive查詢,支持標(biāo)準(zhǔn)連接 支持JDBC和ODBC 連接,應(yīng)用庫Spark Streaming,Spark應(yīng)用庫,Spark Streaming是建立在Spark上的實(shí)時(shí)計(jì)算框架，通過它提供豐富的API、基于內(nèi)存的高速執(zhí)行引擎，用戶可以結(jié)合流式、批處理和交互試查詢應(yīng)用。 Saprk的低延遲執(zhí)行引擎（100MS+）可以用于實(shí)時(shí)處理 相比于strom(基于Record)，RDD數(shù)據(jù)更容易做容錯(cuò)。 可以與kafka,Flume,ZeroMQ等進(jìn)行數(shù)據(jù)源的對(duì)接 小批量處理的方式使得它可以同時(shí)兼容批量和實(shí)時(shí)數(shù)據(jù)處理的邏輯和算法。方便了一些需要?dú)v史數(shù)據(jù)和實(shí)時(shí)數(shù)據(jù)聯(lián)合分析的特定應(yīng)用場合。,應(yīng)用庫Spark Streaming,Spark應(yīng)用庫,Spark Streaming基本原理是將流數(shù)據(jù)分成小的時(shí)間片段（幾秒），以類似批處理方式來處理這部分小數(shù)據(jù)。 處理流程： Spark Streaming把實(shí)時(shí)輸入數(shù)據(jù)流以時(shí)間片t （如1秒）為單位切分成塊 Spark Streaming會(huì)把每塊數(shù)據(jù)作為一個(gè)RDD，并使用RDD操作處理每一小塊數(shù)據(jù) 每個(gè)塊都會(huì)生成一個(gè)Spark Job處理 最終結(jié)果也返回多塊,使用Spark Streaming編寫的程序與編寫Spark程序非常相似 在Spark程序中，主要通過操作RDD（Resilient Distributed Datasets彈性分布式數(shù)據(jù)集）提供的接口，如map、reduce、filter等，實(shí)現(xiàn)數(shù)據(jù)的批處理。 而在Spark Streaming中，則通過操作DStream（表示數(shù)據(jù)流的RDD序列）提供的接口，這些接口和RDD提供的接口類似。,應(yīng)用庫MLlib和GraphX,Spark應(yīng)用庫,MLlib 是Spark對(duì)常用的機(jī)器學(xué)習(xí)算法的實(shí)現(xiàn)庫，同時(shí)包括相關(guān)的測試和數(shù)據(jù)生成器。MLlib 目前支持四種常見的機(jī)器學(xué)習(xí)問題：二元分類，回歸，聚類以及協(xié)同過濾，同時(shí)也包括一個(gè)底層的梯度下降優(yōu)化基礎(chǔ)算法。,易用性：可以使用JAVA、Scala、Python接口 高性能：比MR快100+倍 快速部署:可以運(yùn)行在已經(jīng)安裝好的Hadoop 2.0 集群上,GraphX是一個(gè)分布式圖處理框架，基于Spark平臺(tái)提供對(duì)圖計(jì)算和圖挖掘簡潔易用的而豐富多彩的接口，極大的方便了大家對(duì)分布式圖處理的需求。,靈活性：圖表和集合可以無縫銜接 高性能：比其他圖處理軟件處理快 算法種類多:提供更多的圖算法,目錄,Spark簡介,核心模塊的實(shí)現(xiàn),Spark應(yīng)用,Spark與Hadoop的區(qū)別與聯(lián)系,Spark總體流程,Spark應(yīng)用庫,spark是對(duì)MapReduce計(jì)算模型的改進(jìn)，可以說沒有HDFS，MapReduce，就沒有spark.,Spark與Hadoop的聯(lián)系,Spark可以使用YARN作為它的集群管理器，并且可以處理HDFS的數(shù)據(jù)。這對(duì)于已經(jīng)部署Hadoop集群的用戶特別重要，畢竟不需要做任何的數(shù)據(jù)遷移就可以使用Spark的強(qiáng)大處理能力。,Hadoop,Spark,MapRedcue由Map和Reduce兩個(gè)階段，并通過shuffle將兩個(gè)階段連接起來的。 但是套用MapReduce模型解決問題，不得不將問題分解為若干個(gè)有依賴關(guān)系的子問題，每個(gè)子問題對(duì)應(yīng)一個(gè)MapReduce作業(yè)，最終所有這些作業(yè)形成一個(gè)DAG。,Spark是通用的DAG框架，可以將多個(gè)有依賴關(guān)系的作業(yè)轉(zhuǎn)換為一個(gè)大的DAG。 核心思想是將Map和Reduce兩個(gè)操作進(jìn)一步拆分為多個(gè)元操作，這些元操作可以靈活組合，產(chǎn)生新的操作，并經(jīng)過一些控制程序組裝后形成一個(gè)大的DAG作業(yè)。,框架,在DAG中，由于有多個(gè)MapReduce作業(yè)組成，每個(gè)作業(yè)都會(huì)從HDFS上讀取一次數(shù)據(jù)和寫一次數(shù)據(jù)（默認(rèn)寫三份），即使這些MapReduce作業(yè)產(chǎn)生的數(shù)據(jù)是中間數(shù)據(jù)也需要寫HDFS。 這種表達(dá)作業(yè)依賴關(guān)系的方式比較低效，會(huì)浪費(fèi)大量不必要的磁盤和網(wǎng)絡(luò)IO，根本原因是作業(yè)之間產(chǎn)生的數(shù)據(jù)不是直接流動(dòng)的，而是借助HDFS作為共享數(shù)據(jù)存儲(chǔ)系統(tǒng)。,在Spark中，使用內(nèi)存（內(nèi)存不夠使用本地磁盤）替代了使用HDFS存儲(chǔ)中間結(jié)果。 對(duì)于迭代運(yùn)算效率更高。,中間結(jié)果存儲(chǔ)方式,Spark與Hadoop的區(qū)別,中間結(jié)果是指reduce操作后生成的結(jié)果,Hadoop只提供了Map和Reduce兩種操作 所有的作業(yè)都得轉(zhuǎn)換成Map和Reduce的操作,Spark提供很多種的數(shù)據(jù)集操作類型 比如Transformations 包括map, filter, flatMap, sample, groupByKey, reduceByKey, union, join, cogroup, mapValues,sort,partionBy等多種操作類型，還提供actions操作包括Count,collect, reduce, lookup, save等多種。 這些多種多樣的數(shù)據(jù)集操作類型，給開發(fā)上層應(yīng)用的用戶提供了方便。,操作模型,Hadoop就是唯一的Data Shuffle一種模式,spark用戶可以命名，物化，控制中間結(jié)果的存儲(chǔ)、分區(qū)等，編程方式更靈活,spark使用scala語言，更簡潔高效 spark對(duì)機(jī)器學(xué)習(xí)算法，圖計(jì)算能力有很好的支持。,Hadoop對(duì)迭代計(jì)算效率低,其他,Hadoop,Spark,Spark與Hadoop的區(qū)別,離線大規(guī)模分析處理,Hadoop適用的場景基本上都適合 ( 在只有map操作或者只有一次reduce操作的場景下，Spark比Hadoop的優(yōu)勢不明顯) 對(duì)于迭代計(jì)算比Hadoop有更大的優(yōu)勢,編程模型,應(yīng)用場景,總的來說，Spark采用更先進(jìn)的架構(gòu)，使得靈活性、易用性、性能等方面都比Hadoop更有優(yōu)勢，有取代Hadoop的趨勢，但其穩(wěn)定性有待進(jìn)一步提高。,*詳情見附錄,Hadoop無法緩存數(shù)據(jù)集,spark的60%內(nèi)存用來緩存RDD，對(duì)于緩存后的rdd進(jìn)行操作，節(jié)省IO，效率高,緩存,目錄,Spark簡介,核心模塊的實(shí)現(xiàn),Spark應(yīng)用,Spark與Hadoop的區(qū)別與聯(lián)系,Spark總體流程,Spark應(yīng)用庫,Spark應(yīng)用場景,適用場景,Spark是基于內(nèi)存的迭代計(jì)算框架，適用于需要多次操作特定數(shù)據(jù)集的應(yīng)用場合。 需要反復(fù)操作的次數(shù)越多，所需讀取的數(shù)據(jù)量越大，受益越大，數(shù)據(jù)量小但是計(jì)算密集度較大的場合，受益就相對(duì)較小,由于RDD的特性，Spark不適用那種異步細(xì)粒度更新狀態(tài)的應(yīng)用。 對(duì)于增量修改的應(yīng)用模型不適合。,總的來說Spark的適用面比較廣泛且比較通用。,不適用場景,目前大數(shù)據(jù)在互聯(lián)網(wǎng)公司主要把Spark應(yīng)用在廣告、報(bào)表、推薦系統(tǒng)等業(yè)務(wù)上。 在廣告業(yè)務(wù)方面需要大數(shù)據(jù)做應(yīng)用分析、效果分析、定向優(yōu)化等 在推薦系統(tǒng)方面則需要大數(shù)據(jù)優(yōu)化相關(guān)排名、個(gè)性化推薦以及熱點(diǎn)點(diǎn)擊分析等。 這些應(yīng)用場景的普遍特點(diǎn)是計(jì)算量大、效率要求高。Spark恰恰滿足了這些要求。,Spark在互聯(lián)網(wǎng)界的使用,Spark在移動(dòng)網(wǎng)優(yōu)測試結(jié)果,統(tǒng)一采集平臺(tái)耗時(shí)最長時(shí)間的匯總?cè)蝿?wù)tpa_mr_qoe_celladj_hour.sh,前端耗時(shí)最長時(shí)間的任務(wù)C/S模式下，按模板查詢，導(dǎo)出24小時(shí)LTE性能數(shù)據(jù)（小區(qū)級(jí)指標(biāo)數(shù)據(jù)）,hive性能比informix性能提升了58%；spark性能比informix性能提升了140%,hive比informix性能提升2倍。spark比informix性能提升6倍。,Spark在移動(dòng)網(wǎng)優(yōu)測試結(jié)果分析,根據(jù)spark適合批量處理整個(gè)或大部分?jǐn)?shù)據(jù)集的特點(diǎn)，重新設(shè)計(jì)數(shù)據(jù)模型 問題： 當(dāng)前業(yè)務(wù)是從一張大表中篩選出滿足指定時(shí)間條件的記錄進(jìn)行處理（從上億條數(shù)據(jù)集中篩選出幾十萬條記錄后進(jìn)行處理），spark會(huì)將整個(gè)大表讀出來，然后篩選出滿足要求的記錄，按照目前的實(shí)現(xiàn)方式，比較浪費(fèi)IO損失性能，沒有充分利用Spark的優(yōu)勢。 措施： 后續(xù)應(yīng)該重新設(shè)計(jì)數(shù)據(jù)分布，按照匯總處理粒度存儲(chǔ)文件，并周期性合并文件。如此設(shè)計(jì)可以充分利用Spark的機(jī)制，提升性能。比如匯總小時(shí)級(jí)別的數(shù)據(jù)，就可以考慮1個(gè)小時(shí)一個(gè)文件，一天或者一周整合一下文件。,優(yōu)化spark本身的配置參數(shù) 問題： 由于測試時(shí)間緊張，在默認(rèn)配置下進(jìn)行的測試，沒有對(duì)Spark參數(shù)進(jìn)行優(yōu)化 措施： 根據(jù)集群的配置情況以及系統(tǒng)的負(fù)載情況對(duì)Spark參數(shù)進(jìn)行優(yōu)化，進(jìn)一步提升系統(tǒng)性能,根據(jù)Spark適用于迭代場景的特點(diǎn)，重新設(shè)計(jì)業(yè)務(wù)處理流程和sql編寫 問題： 原有的業(yè)務(wù)處理邏輯是根據(jù)數(shù)據(jù)庫的特點(diǎn)，建立多張臨時(shí)表，分級(jí)分步多次匯總完成最終的任務(wù)，spark對(duì)中間過程的每步操作都看作一個(gè)job，沒有使用到迭代的特性。 措施： 根據(jù)spark適用場景重新設(shè)計(jì)處理流程和sql編寫，使得中間處理數(shù)據(jù)被迭代使用，提升效率,本次spark測試結(jié)果是在沒有修改原有處理邏輯的基礎(chǔ)上進(jìn)行了spark的測試，提升了部分性能，后續(xù)如果網(wǎng)管系統(tǒng)使用Spark需要根據(jù)其本身特