大數據與機器學習

目錄第八章Hive介紹 31.1.1HIVE產生背景 31.1.2組件生態架構 31.1.3HIVE架構 41.1.4部署架構-主備模式 51.1.5處理流程 6第九章Spark介紹 71.1.6Spark產生背景 71.1.7Spark業務架構 91.1.8Spark部署 101.1.9Spark原理 111.1.10Spark-core的RDD 111.1.11產生背景 171.1.12Spark-streaming 181.1.13Spark-graphX圖計算 20第八章Hive介紹HIVE產生背景為了使用大數據Hadoop組件，減少MapReduce函數使用的學習成本，Facebook開發了通過使用SQL語言轉換為MR函數的工具HIVE。從而可以直接使用MR作業進行數據統計作業，或者直接調用HDFS。組件生態架構HIVE架構Command-lineshell:命令行模式調用顯示數據計算結果Thrift與JDBC：支持SQL語句調用顯示數據計算結果HueWeb：大數據交互平臺，支持Web調用HDFS以圖表類型顯示數據統計結果。Metastore:元數據存儲。主要使用的是關系型數據庫（MySQL）。元數據主要是描述數據資源屬性信息的數據。如：存儲位置、數據字段、數據目錄。部署架構-主備模式主備模式，以防HIVE中的MySQL崩潰而丟失數據。處理流程Hive是基于Hadoop的一個數據倉庫工具，用來進行數據提取、轉化、加載。Hive是將結構化的數據文件映射為數據庫表，通過SQL語句底層調用MapReduce進行數據分析。hive十分適合對數據倉庫進行統計分析。Hive客戶端：由CLI命令行、JDBC(java數據庫連接器-JAVA開發)/ODBC（面向對面數據庫連接器-C語言開發）、WebGUI網頁模式，其中CLI和WebGUI都可以直接命令Driver執行任務提取數據。而JDBC/ODBC方式需要啟動ThriftServer(RPC遠程調用)，可以允許客戶端使用包括Java、C++、Ruby和其他很多種語言，通過編程的方式遠程訪問Hive

。（即前端服務可以遠程調用后端數據）Metastore:因為將數據映射為表，所以需要將元數據（表的列和分區及其屬性，表的屬性（是否為外部表等），表的數據所在目錄等）存入數據庫，目前Metastore用的是MySQL關系數據庫。將元數據保存在關系數據庫中，大大減少了在查詢過程中執行語義檢查的時間（即數據庫會去判斷SQL語句是否真正具有相應的含義，比如說sql語句涉及的表，或者字段是否存在等。）Driver：包含解釋器、Compiler編譯器、Optimizer優化器、Executor執行器解釋器：解析sql語句，檢查表，字段是否存在。編譯器：生成Job任務，當前sql語句會生成幾個job優化器：檢查sql語句是否可以優化執行器：執行sql,調用MapReduce任務MapReduce任務全過程客戶端向JobTracker申請任務，由JobTracker向多個TaskTracker分配任務。再由客戶端向YARN的SourceManager申請資源，SourceManager在隨機向一臺NodeManager發布建立里ApplicationMaster的命令，NodeManager本地建立一個container，然后啟動ApplicationMaster，AM再向SM注冊和申請資源，SM收到請求后分配資源，并返回給AM。AM收到后，分別到各個NM要求其啟動container執行任務，并為任務設置資源。各個NM執行任務，讀取本地DataNode的HDFS文件。讀取文件過程為，客戶端先向NameNode請求訪問，請求通過后，由NM將本地最新鏡像文件返回給客戶端，客戶端按照鏡像文件的元數據建立各個NM讀取通道。第九章Spark介紹Spark產生背景Spark是在MapReduce基礎上完善出來的。為了彌補MR的應用不足（實時處理、內存高效、易用提升、迭代處理）。本質上與MR一樣，都是大數據處理引擎。以下是與MR相比的優勢性提升100倍處理效率：因Spark在內存中處理，比MR處理快100倍。或者在硬盤處理也要快上10倍。提升易用性：Spark提供了80多種high-level操作（ioin、group、sum、count...）更容易建立并行應用，比MR的Map和Reduce的低等級操作要多。完善迭代處理：MapReduce每次操作都要對磁盤進行I\O操作，Spark的通過RDD可以控制中間結果，將結果應用到后續的計算中。最后統計寫入HDFS，大大減少了中間結果的存儲，降低了數據復制、磁盤I\O和數據序列化開銷。完善實時流處理：Streaming流式處理完善實時計算Spark業務架構SparkCore：

實現了Spark的基本功能，包含任務調度、內存管理、錯誤恢復、與存儲系統交互等模塊。SparkCore中還包含了對彈性分布式數據集(resilientdistributeddataset，簡稱RDD)的API定義。

SparkSQL：

是Spark用來操作結構化數據的程序包。通過SparkSQL，我們可以使用SQL或者ApacheHive版本的SQL方言(HQL)來查詢數據。SparkSQL支持多種數據源，比如Hive表、Parquet以及JSON等。

SparkStreaming：

是Spark提供的對實時數據進行流式計算的組件。提供了用來操作數據流的API，并且與SparkCore中的RDDAPI高度對應。

SparkMLlib：

提供常見的機器學習(ML)功能的程序庫。包括分類、回歸、聚類、協同過濾等，還提供了模型評估、數據導入等額外的支持功能。

集群管理器：

Spark設計為可以高效地在一個計算節點到數千個計算節點之間伸縮計算。為了實現這樣的要求，同時獲得最大靈活性，Spark支持在各種集群管理器(clustermanager)上運行，包括HadoopYARN、ApacheMesos，以及Spark自帶的一個簡易調度器，叫作獨立調度器。Spark部署從物理部署層面上來看，Spark主要分為兩種類型的節點，Master節點和Worker節點，Master節點主要運行集群管理器的中心化部分，所承載的作用是分配Application到Worker節點，維護Worker節點，Driver，Application的狀態。Worker節點負責具體的業務運行。從Spark程序運行的層面來看，Spark主要分為驅動器節點和執行器節點。Spark原理RDD基本操作RDD的操作分為轉化（Transformation）操作和行動（Action）操作。轉化操作就是從一個RDD產生一個新的RDD，而行動操作就是進行實際的計算。RDD的操作是惰性的，當RDD執行轉化操作的時候，實際計算并沒有被執行，只有當RDD執行行動操作時才會促發計算任務提交，從而執行相應的計算操作。1.構建操作Spark里的計算都是通過操作RDD完成的，學習RDD的第一個問題就是如何構建RDD，構建RDD的方式從數據來源角度分為以下兩類。從內存里直接讀取數據。從文件系統里讀取數據，文件系統的種類很多，常見的就是HDFS及本地文件系統。第一類方式是從內存里構造RDD，需要使用makeRDD方法，代碼如下所示。valrdd01=sc.makeRDD(List(l,2,3,4,5,6))這個語句創建了一個由“1,2,3,4,5,6”六個元素組成的RDD。第二類方式是通過文件系統構造RDD，代碼如下所示valrdd:RDD[String]==sc.textFile(“file:///D:/sparkdata.txt”,1)這里例子使用的是本地文件系統，所以文件路徑協議前綴是file://。2.轉換操作RDD的轉換操作是返回新的RDD的操作。轉換出來的RDD是惰性求值的，只有在行動操作中用到這些RDD時才會被計算。許多轉換操作都是針對各個元素的，也就是說，這些轉換操作每次只會操作RDD中的一個元素，不過并不是所有的轉換操作都是這樣的。表1描述了常用的RDD轉換操作。3.行動操作行動操作用于執行計算并按指定的方式輸出結果。行動操作接受RDD，但是返回非RDD，即輸出一個值或者結果。在RDD執行過程中，真正的計算發生在行動操作。表2描述了常用的RDD行動操作。RDD血緣關系RDD的最重要的特性之一就是血緣關系（Lineage)，它描述了一個RDD是如何從父RDD計算得來的。如果某個RDD丟失了，則可以根據血緣關系，從父RDD計算得來。圖2給出了一個RDD執行過程的實例。系統從輸入中邏輯上生成了A和C兩個RDD，經過一系列轉換操作，邏輯上生成了F這個RDD。Spark記錄了RDD之間的生成和依賴關系。當F進行行動操作時，Spark才會根據RDD的依賴關系生成DAG，并從起點開始真正的計算。上述一系列處理稱為一個血緣關系（Lineage），即DAG拓撲排序的結果。在血緣關系中，下一代的RDD依賴于上一代的RDD。例如，在圖2中，B依賴于A，D依賴于C，而E依賴于B和D。RDD依賴類型根據不同的轉換操作，RDD血緣關系的依賴分為窄依賴和寬依賴。窄依賴是指父RDD的每個分區都只被子RDD的一個分區所使用（一對一）。寬依賴是指父RDD的每個分區都被多個子RDD的分區所依賴（一對多）。map、filter、union等操作是窄依賴，而groupByKey、reduceByKey等操作是寬依賴，如圖3所示join操作有兩種情況，如果join操作中使用的每個Partition僅僅和固定個Partition進行join，則該join操作是窄依賴，其他情況下的join操作是寬依賴。所以可得出一個結論，窄依賴不僅包含一對一的窄依賴，還包含一對固定個數的窄依賴，也就是說，對父RDD依賴的Partition不會隨著RDD數據規模的改變而改變。1.窄依賴1）子RDD的每個分區依賴于常數個父分區（即與數據規模無關)。2）輸入輸出一對一的算子，且結果RDD的分區結構不變，如map、flatMap。3）輸入輸出一對一的算子，但結果RDD的分區結構發生了變化，如union。4）從輸入中選擇部分元素的算子，如filter、distinct、subtract、sample。2.寬依賴對單個RDD基于Key進行重組和reduce，如groupByKey、reduceByKey。3）對兩個RDD基于Key進行join和重組，如join。Spark的這種依賴關系設計，使其具有了天生的容錯性，大大加快了Spark的執行速度。RDD通過血緣關系記住了它是如何從其他RDD中演變過來的。當這個RDD的部分分區數據丟失時，它可以通過血緣關系獲取足夠的信息來重新運算和恢復丟失的數據分區，從而帶來性能的提升。相對而言，窄依賴的失敗恢復更為高效，它只需要根據父RDD分區重新計算丟失的分區即可，而不需要重新計算父RDD的所有分區。而對于寬依賴來講，單個結點失效，即使只是RDD的一個分區失效，也需要重新計算父RDD的所有分區，開銷較大。寬依賴操作就像是將父RDD中所有分區的記錄進行了“洗牌”，數據被打散，然后在子RDD中進行重組。階段劃分DAGDAG（DirectedAcyclicGraph有向無環圖）又稱“血統”Lineage。用戶提交的計算任務是一個由RDD構成的DAG，如果RDD的轉換是寬依賴，那么這個寬依賴轉換就將這個DAG分為了不同的階段（Stage)。由于寬依賴會帶來“洗牌”，所以不同的Stage是不能并行計算的，后面Stage的RDD的計算需要等待前面Stage的RDD的所有分區全部計算完畢以后才能進行。這點就類似于在MapReduce中，Reduce階段的計算必須等待所有Map任務完成后才能開始一樣。在對Job中的所有操作劃分Stage時，一般會按照倒序進行，即從Action開始，遇到窄依賴操作，則劃分到同一個執行階段，遇到寬依賴操作，則劃分一個新的執行階段。后面的Stage需要等待所有的前面的Stage執行完之后才可以執行，這樣Stage之間根據依賴關系就構成了一個大粒度的DAG。下面通過圖4詳細解釋一下階段劃分。假設從HDFS中讀入數據生成3個不同的RDD(A、C和E)，通過一系列轉換操作后得到新的RDD(G)，并把結果保存到HDFS中。可以看到這幅DAG中只有join操作是一個寬依賴，Spark會以此為邊界將其前后劃分成不同的階段。同時可以注意到，在Stage2中，從map到union都是窄依賴，這兩步操作可以形成一個流水線操作，通過map操作生成的分區可以不用等待整個RDD計算結束，而是繼續進行union操作，這樣大大提高了計算的效率。把一個DAG圖劃分成多個Stage以后，每個Stage都代表了一組由關聯的、相互之間沒有寬依賴關系的任務組成的任務集合。在運行的時候，Spark會把每個任務集合提交給任務調度器進行處理。產生背景ApacheFlink是由Apache軟件基金會開發的開源流處理框架，其核心是用Java和Scala編寫的分布式流數據處理引擎。Flink以數據并行和流水線方式執行任意流數據程序，Flink的流水線運行時系統可以執行批處理和流處理程序。此外，Flink的運行時本身也支持迭代算法的執行。Flink提供高吞吐量、低延遲的流數據處理引擎以及對事件-時間處理和狀態管理的支持。Flink應用程序在發生機器故障時具有容錯能力，并且支持exactly-once語義。（嚴格一次）Spark-streamingStreaming：是一種數據傳送技術，它把客戶機收到的數據變成一個穩定連續的流，源源不斷地送出，使用戶聽到的聲音或看到的圖象十分平穩，而且用戶在整個文件送完之前就可以開始在屏幕上瀏覽文件。常見的流式計算框架ApacheStormSparkStreamingApacheSamzaSparkStreaming是SparkCore的擴展應用，它具有可擴展，高吞吐量，對于流數據的可容錯性等特點。可以監控來自Kafka,Flume,HDFS。Kinesis,Twitter，ZeroMQ或者Scoket套接字的數據通過復雜的算法以及一系列的計算分析數據，并且可以將分析結果存入到HDFS文件系統，數據庫以及前端頁面中。SparkStreaming有以下特點：高可擴展性，可以運行在上百臺機器上（Scalestohundredsofnodes）低延遲，可以在秒級別上對數據進行處理（Achieveslowlatency）高可容錯性（Efficientlyrecoverfromfailures）能夠集成并行計算程序，比如SparkCore（Integrateswithbatchandinteractiveprocessing）工作原理對于SparkCore它的核心就是RDD，對于SparkStreaming來說，它的核心是DStream，DStream類似于RDD，它實質上一系列的RDD的集合，DStream可以按照秒數將數據流進行批量的劃分。首先從接收到流數據之后，將其劃分為多個batch，然后提交給Spark集群進行計算，最后將結果批量輸出到HDFS或者數據庫以及前端頁面展示等等。可以參考下面這幅圖來幫助理解：DStream是一系列連續的RDD，它是建立在Spark之上的不可變的，分布式數據集，在DStream中的每一個RDD包含著一定時間間隔的數據，如下圖所示：SparkCore在初始化時會生成一個SparkContext對象來對數據進行后續的處理，相對應的SparkStreaming會創建一個StreamingContext，它的底層是SparkContext，也就是說它會將任務提交給SparkContext來執行，這也很好的解釋了DStream是一系列的RDD。當啟動SparkStreaming應用的時候，首先會在一個節點的Executor上啟動一個Receiver接受者，然后當從數據源寫入數據的時候會被Receiver接收，接收到數據之后Receiv