Spark簡介

　　Spark是整個BDAS的核心組件，是一個大數據分布式編程框架，不僅實現了MapReduce的算子map 函數和reduce函數及計算模型，還提供更為豐富的算子，如filter、join、groupByKey等。是一個用來實現快速而同用的集群計算的平臺。

　　Spark將分布式數據抽象為彈性分布式數據集（RDD），實現了應用任務調度、RPC、序列化和壓縮，并為運行在其上的上層組件提供API。其底層采用Scala這種函數式語言書寫而成，并且所提供的API深度借鑒Scala函數式的編程思想，提供與Scala類似的編程接口

　　Sparkon Yarn

　　從用戶提交作業到作業運行結束整個運行期間的過程分析。

　　一、客戶端進行操作

1. 根據yarnConf來初始化yarnClient，并啟動yarnClient

2. 創建客戶端Application，并獲取Application的ID，進一步判斷集群中的資源是否滿足executor和ApplicationMaster申請的資源，如果不滿足則拋出IllegalArgumentException；

3. 設置資源、環境變量：其中包括了設置Application的Staging目錄、準備本地資源（jar文件、log4j.properties）、設置Application其中的環境變量、創建Container啟動的Context等；

4. 設置Application提交的Context，包括設置應用的名字、隊列、AM的申請的Container、標記該作業的類型為Spark；

5. 申請Memory，并最終通過yarnClient.submitApplication向ResourceManager提交該Application。

　　當作業提交到YARN上之后，客戶端就沒事了，甚至在終端關掉那個進程也沒事，因為整個作業運行在YARN集群上進行，運行的結果將會保存到HDFS或者日志中。

　　二、提交到YARN集群，YARN操作

1. 運行ApplicationMaster的run方法；

2. 設置好相關的環境變量。

3. 創建amClient，并啟動；

4. 在Spark UI啟動之前設置Spark UI的AmIpFilter；

5. 在startUserClass函數專門啟動了一個線程（名稱為Driver的線程）來啟動用戶提交的Application，也就是啟動了Driver。在Driver中將會初始化SparkContext；

6. 等待SparkContext初始化完成，最多等待spark.yarn.applicationMaster.waitTries次數（默認為10），如果等待了的次數超過了配置的，程序將會退出；否則用SparkContext初始化yarnAllocator；

7. 當SparkContext、Driver初始化完成的時候，通過amClient向ResourceManager注冊ApplicationMaster

8. 分配并啟動Executeors。在啟動Executeors之前，先要通過yarnAllocator獲取到numExecutors個Container，然后在Container中啟動Executeors。

   　　那么這個Application將失敗，將Application Status標明為FAILED，并將關閉SparkContext。其實，啟動Executeors是通過ExecutorRunnable實現的，而ExecutorRunnable內部是啟動CoarseGrainedExecutorBackend的。

9. 最后，Task將在CoarseGrainedExecutorBackend里面運行，然后運行狀況會通過Akka通知CoarseGrainedScheduler，直到作業運行完成。

　　Spark節點的概念

　　一、Spark驅動器是執行程序中的main()方法的進程。它執行用戶編寫的用來創建SparkContext(初始化)、創建RDD，以及運行RDD的轉化操作和行動操作的代碼。

　　驅動器節點driver的職責：

1. 把用戶程序轉為任務task(driver)

   　　Spark驅動器程序負責把用戶程序轉化為多個物理執行單元，這些單元也被稱之為任務task(詳解見備注)

2. 為執行器節點調度任務(executor)

   　　有了物理計劃之后，Spark驅動器在各個執行器節點進程間協調任務的調度。Spark驅動器程序會根據當前的執行器節點，把所有任務基于數據所在位置分配給合適的執行器進程。當執行任務時，執行器進程會把緩存的數據存儲起來，而驅動器進程同樣會跟蹤這些緩存數據的位置，并利用這些位置信息來調度以后的任務，以盡量減少數據的網絡傳輸。（就是所謂的移動計算，而不移動數據)。

　　二、執行器節點

　　作用：

1. 負責運行組成Spark應用的任務，并將結果返回給驅動器進程；

2. 通過自身的塊管理器(blockManager)為用戶程序中要求緩存的RDD提供內存式存儲。RDD是直接緩存在執行器進程內的，因此任務可以在運行時充分利用緩存數據加快運算。

　　驅動器的職責：

　　所有的Spark程序都遵循同樣的結構：程序從輸入數據創建一系列RDD，再使用轉化操作派生成新的RDD，最后使用行動操作手機或存儲結果RDD，Spark程序其實是隱式地創建出了一個由操作組成的邏輯上的有向無環圖DAG。當驅動器程序執行時，它會把這個邏輯圖轉為物理執行計劃。

　　這樣 Spark就把邏輯計劃轉為一系列步驟(stage)，而每個步驟又由多個任務組成。這些任務會被打包送到集群中。

　　Spark初始化

1. 每個Spark應用都由一個驅動器程序來發起集群上的各種并行操作。驅動器程序包含應用的main函數，并且定義了集群上的分布式數據集，以及對該分布式數據集應用了相關操作。

2. 驅動器程序通過一個SparkContext對象來訪問spark,這個對象代表對計算集群的一個連接。（比如在sparkshell啟動時已經自動創建了一個SparkContext對象，是一個叫做SC的變量。(下圖，查看變量sc)

   　　

3. 一旦創建了sparkContext，就可以用它來創建RDD。比如調用sc.textFile()來創建一個代表文本中各行文本的RDD。（比如vallinesRDD = sc.textFile(“yangsy.text”),val spark = linesRDD.filter(line=>line.contains(“spark”),spark.count()）

   　　執行這些操作，驅動器程序一般要管理多個執行器,就是我們所說的executor節點。

4. 在初始化SparkContext的同時，加載sparkConf對象來加載集群的配置，從而創建sparkContext對象。

   　　從源碼中可以看到，在啟動thriftserver時，調用了spark- daemon.sh文件，該文件源碼如左圖，加載spark_home下的conf中的文件。

   　　

   　　（在執行后臺代碼時，需要首先創建conf對象，加載相應參數， val sparkConf = newSparkConf().setMaster("local").setAppName("cocapp").set("spark.executor.memory","1g"), val sc: SparkContext = new SparkContext(sparkConf))

　　RDD工作原理：

　　RDD(Resilient DistributedDatasets)[1] ,彈性分布式數據集，是分布式內存的一個抽象概念，RDD提供了一種高度受限的共享內存模型，即RDD是只讀的記錄分區的集合，只能通過在其他RDD執行確定的轉換操作（如map、join和group by）而創建，然而這些限制使得實現容錯的開銷很低。對開發者而言，RDD可以看作是Spark的一個對象，它本身運行于內存中，如讀文件是一個RDD，對文件計算是一個RDD，結果集也是一個RDD ，不同的分片、數據之間的依賴、key-value類型的map數據都可以看做RDD。

　　主要分為三部分：創建RDD對象，DAG調度器創建執行計劃，Task調度器分配任務并調度Worker開始運行。

　　SparkContext(RDD相關操作)→通過(提交作業)→(遍歷RDD拆分stage→生成作業)DAGScheduler→通過（提交任務集）→任務調度管理(TaskScheduler)→通過（按照資源獲取任務)→任務調度管理(TaskSetManager)

　　Transformation返回值還是一個RDD。它使用了鏈式調用的設計模式，對一個RDD進行計算后，變換成另外一個RDD，然后這個RDD又可以進行另外一次轉換。這個過程是分布式的。

　　Action返回值不是一個RDD。它要么是一個Scala的普通集合，要么是一個值，要么是空，最終或返回到Driver程序，或把RDD寫入到文件系統中

　　轉換(Transformations)(如：map, filter, groupBy, join等)，Transformations操作是Lazy的，也就是說從一個RDD轉換生成另一個RDD的操作不是馬上執行，Spark在遇到Transformations操作時只會記錄需要這樣的操作，并不會去執行，需要等到有Actions操作的時候才會真正啟動計算過程進行計算。

　　操作(Actions)(如：count, collect, save等)，Actions操作會返回結果或把RDD數據寫到存儲系統中。Actions是觸發Spark啟動計算的動因。

　　它們本質區別是：Transformation返回值還是一個RDD。它使用了鏈式調用的設計模式，對一個RDD進行計算后，變換成另外一個RDD，然后這個RDD又可以進行另外一次轉換。這個過程是分布式的。Action返回值不是一個RDD。它要么是一個Scala的普通集合，要么是一個值，要么是空，最終或返回到Driver程序，或把RDD寫入到文件系統中。關于這兩個動作，在Spark開發指南中會有就進一步的詳細介紹，它們是基于Spark開發的核心。

　　RDD基礎

1. Spark中的RDD就是一個不可變的分布式對象集合。每個RDD都被分為多個分區，這些分區運行在集群的不同節點上。創建RDD的方法有兩種：一種是讀取一個外部數據集；一種是在群東程序里分發驅動器程序中的對象集合，不如剛才的示例，讀取文本文件作為一個字符串的RDD的示例。

2. 創建出來后，RDD支持兩種類型的操作:轉化操作和行動操作

   　　轉化操作會由一個RDD生成一個新的RDD。（比如剛才的根據謂詞篩選）

   　　行動操作會對RDD計算出一個結果，并把結果返回到驅動器程序中，或把結果存儲到外部存儲系統（比如HDFS）中。比如first()操作就是一個行動操作，會返回RDD的第一個元素。

   　　注：轉化操作與行動操作的區別在于Spark計算RDD的方式不同。雖然你可以在任何時候定義一個新的RDD，但Spark只會惰性計算這些RDD。它們只有第一個在一個行動操作中用到時，才會真正的計算。之所以這樣設計，是因為比如剛才調用sc.textFile(...)時就把文件中的所有行都讀取并存儲起來，就會消耗很多存儲空間，而我們馬上又要篩選掉其中的很多數據。

   　　這里還需要注意的一點是，spark會在你每次對它們進行行動操作時重新計算。如果想在多個行動操作中重用同一個RDD，那么可以使用RDD.persist()或RDD.collect()讓Spark把這個RDD緩存下來。（可以是內存，也可以是磁盤)

3. Spark會使用譜系圖來記錄這些不同RDD之間的依賴關系，Spark需要用這些信息來按需計算每個RDD，也可以依靠譜系圖在持久化的RDD丟失部分數據時用來恢復所丟失的數據。(如下圖，過濾errorsRDD與warningsRDD,最終調用union()函數)

   　　

　　RDD計算方式

　　RDD的寬窄依賴

　　窄依賴 (narrowdependencies) 和寬依賴 (widedependencies) 。窄依賴是指 父 RDD 的每個分區都只被子 RDD 的一個分區所使用 。相應的，那么寬依賴就是指父 RDD 的分區被多個子 RDD 的分區所依賴。例如， map 就是一種窄依賴，而 join 則會導致寬依賴

　　這種劃分有兩個用處。首先，窄依賴支持在一個結點上管道化執行。例如基于一對一的關系，可以在 filter 之后執行 map 。其次，窄依賴支持更高效的故障還原。因為對于窄依賴，只有丟失的父 RDD 的分區需要重新計算。而對于寬依賴，一個結點的故障可能導致來自所有父 RDD 的分區丟失，因此就需要完全重新執行。因此對于寬依賴，Spark 會在持有各個父分區的結點上，將中間數據持久化來簡化故障還原，就像 MapReduce 會持久化 map 的輸出一樣。

　　SparkExample

　　步驟 1 ：創建 RDD 。上面的例子除去最后一個 collect 是個動作，不會創建 RDD 之外，前面四個轉換都會創建出新的 RDD 。因此第一步就是創建好所有 RDD( 內部的五項信息 ) 。

　　步驟 2 ：創建執行計劃。Spark 會盡可能地管道化，并基于是否要重新組織數據來劃分 階段 (stage) ，例如本例中的 groupBy() 轉換就會將整個執行計劃劃分成兩階段執行。最終會產生一個 DAG(directedacyclic graph ，有向無環圖 ) 作為邏輯執行計劃。

　　步驟 3 ：調度任務。 將各階段劃分成不同的 任務 (task) ，每個任務都是數據和計算的合體。在進行下一階段前，當前階段的所有任務都要執行完成。因為下一階段的第一個轉換一定是重新組織數據的，所以必須等當前階段所有結果數據都計算出來了才能繼續。

　　假設本例中的 hdfs://names 下有四個文件塊，那么 HadoopRDD 中 partitions 就會有四個分區對應這四個塊數據，同時 preferedLocations 會指明這四個塊的最佳位置。現在，就可以創建出四個任務，并調度到合適的集群結點上。

　　Spark數據分區

1. Spark的特性是對數據集在節點間的分區進行控制。在分布式系統中，通訊的代價是巨大的，控制數據分布以獲得最少的網絡傳輸可以極大地提升整體性能。Spark程序可以通過控制RDD分區方式來減少通訊的開銷。

2. Spark中所有的鍵值對RDD都可以進行分區。確保同一組的鍵出現在同一個節點上。比如，使用哈希分區將一個RDD分成了100個分區，此時鍵的哈希值對100取模的結果相同的記錄會被放在一個節點上。

   　　（可使用partitionBy(newHashPartitioner(100)).persist()來構造100個分區)

3. Spark中的許多操作都引入了將數據根據鍵跨界點進行混洗的過程。(比如：join(),leftOuterJoin(),groupByKey(),reducebyKey()等)對于像reduceByKey()這樣只作用于單個RDD的操作，運行在未分區的RDD上的時候會導致每個鍵的所有對應值都在每臺機器上進行本地計算。

　　SparkSQL的shuffle過程

　　Spark SQL的核心是把已有的RDD，帶上Schema信息，然后注冊成類似sql里的”Table”，對其進行sql查詢。這里面主要分兩部分，一是生成SchemaRD，二是執行查詢。

　　如果是spark-hive項目，那么讀取metadata信息作為Schema、讀取hdfs上數據的過程交給Hive完成，然后根據這倆部分生成SchemaRDD，在HiveContext下進行hql()查詢。

　　SparkSQL結構化數據

1. 首先說一下ApacheHive，Hive可以在HDFS內或者在其他存儲系統上存儲多種格式的表。SparkSQL可以讀取Hive支持的任何表。要把Spark SQL連接已有的hive上，需要提供Hive的配置文件。hive-site.xml文件復制到spark的conf文件夾下。再創建出HiveContext對象(sparksql的入口)，然后就可以使用HQL來對表進行查詢，并以由行足證的RDD的形式拿到返回的數據。

2. 創建Hivecontext并查詢數據

   　　importorg.apache.spark.sql.hive.HiveContext

   　　valhiveCtx = new org.apache.spark.sql.hive.HiveContext(sc)

   　　valrows = hiveCtx.sql(“SELECT name,age FROM users”)

   　　valfitstRow – rows.first()

   　　println(fitstRow.getSgtring(0)) //字段0是name字段

3. 通過jdbc連接外部數據源更新與加載

   　　Class.forName("com.mysql.jdbc.Driver")

   　　val conn =DriverManager.getConnection(mySQLUrl)

   　　val stat1 =conn.createStatement()

   　　stat1.execute("UPDATE CI_LABEL_INFO set DATA_STATUS_ID = 2 , DATA_DATE ='" + dataDate +"' where LABEL_ID in ("+allCreatedLabels.mkString(",")+")")

   　　stat1.close()

   　　//加載外部數據源數據到內存

   　　valDIM_COC_INDEX_MODEL_TABLE_CONF =sqlContext.jdbc(mySQLUrl,"DIM_COC_INDEX_MODEL_TABLE_CONF").cache()

   　　val targets =DIM_COC_INDEX_MODEL_TABLE_CONF.filter("TABLE_DATA_CYCLE ="+TABLE_DATA_CYCLE).collect

　　SparkSQL解析

　　首先說下傳統數據庫的解析，傳統數據庫的解析過程是按Rusult、Data Source、Operation的次序來解析的。傳統數據庫先將讀入的SQL語句進行解析，分辨出SQL語句中哪些詞是關鍵字（如select,from,where)，哪些是表達式，哪些是Projection，哪些是Data Source等等。進一步判斷SQL語句是否規范，不規范就報錯，規范則按照下一步過程綁定（Bind)。過程綁定是將SQL語句和數據庫的數據字典(列,表,視圖等）進行綁定，如果相關的Projection、Data Source等都存在，就表示這個SQL語句是可以執行的。在執行過程中，有時候甚至不需要讀取物理表就可以返回結果，比如重新運行剛運行過的SQL語句，直接從數據庫的緩沖池中獲取返回結果。在數據庫解析的過程中SQL語句時，將會把SQL語句轉化成一個樹形結構來進行處理，會形成一個或含有多個節點(TreeNode)的Tree,然后再后續的處理政對該Tree進行一系列的操作。

　　Spark SQL對SQL語句的處理和關系數據庫對SQL語句的解析采用了類似的方法，首先會將SQL語句進行解析，然后形成一個Tree，后續如綁定、優化等處理過程都是對Tree的操作，而操作方法是采用Rule,通過模式匹配，對不同類型的節點采用不同的操作。SparkSQL有兩個分支，sqlContext和hiveContext。sqlContext現在只支持SQL語法解析器（Catalyst)，hiveContext支持SQL語法和HiveContext語法解析器。