深入淺出spark.pptx

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