Spark簡介
Spark是整個BDAS的核心組件，是一個大數(shù)據(jù)分布式編程框架，不僅實(shí)現(xiàn)了MapReduce的算子map 函數(shù)和reduce函數(shù)及計(jì)算模型，還提供更為豐富的算子，如filter、join、groupByKey等。是一個用來實(shí)現(xiàn)快速而同用的集群計(jì)算的平臺。
Spark將分布式數(shù)據(jù)抽象為彈性分布式數(shù)據(jù)集（RDD），實(shí)現(xiàn)了應(yīng)用任務(wù)調(diào)度、RPC、序列化和壓縮，并為運(yùn)行在其上的上層組件提供API。其底層采用Scala這種函數(shù)式語言書寫而成，并且所提供的API深度借鑒Scala函數(shù)式的編程思想，提供與Scala類似的編程接口
Sparkon Yarn
從用戶提交作業(yè)到作業(yè)運(yùn)行結(jié)束整個運(yùn)行期間的過程分析。
一、客戶端進(jìn)行操作
根據(jù)yarnConf來初始化yarnClient，并啟動yarnClient
創(chuàng)建客戶端Application，并獲取Application的ID，進(jìn)一步判斷集群中的資源是否滿足executor和ApplicationMaster申請的資源，如果不滿足則拋出IllegalArgumentException；
設(shè)置資源、環(huán)境變量：其中包括了設(shè)置Application的Staging目錄、準(zhǔn)備本地資源（jar文件、log4j.properties）、設(shè)置Application其中的環(huán)境變量、創(chuàng)建Container啟動的Context等；
設(shè)置Application提交的Context，包括設(shè)置應(yīng)用的名字、隊(duì)列、AM的申請的Container、標(biāo)記該作業(yè)的類型為Spark；
申請Memory，并最終通過yarnClient.submitApplication向ResourceManager提交該Application。
當(dāng)作業(yè)提交到Y(jié)ARN上之后，客戶端就沒事了，甚至在終端關(guān)掉那個進(jìn)程也沒事，因?yàn)檎麄€作業(yè)運(yùn)行在YARN集群上進(jìn)行，運(yùn)行的結(jié)果將會保存到HDFS或者日志中。
二、提交到Y(jié)ARN集群，YARN操作
運(yùn)行ApplicationMaster的run方法；
設(shè)置好相關(guān)的環(huán)境變量。
創(chuàng)建amClient，并啟動；
在Spark UI啟動之前設(shè)置Spark UI的AmIpFilter；
在startUserClass函數(shù)專門啟動了一個線程（名稱為Driver的線程）來啟動用戶提交的Application，也就是啟動了Driver。在Driver中將會初始化SparkContext；
等待SparkContext初始化完成，最多等待spark.yarn.applicationMaster.waitTries次數(shù)（默認(rèn)為10），如果等待了的次數(shù)超過了配置的，程序?qū)顺觯环駝t用SparkContext初始化yarnAllocator；
當(dāng)SparkContext、Driver初始化完成的時候，通過amClient向ResourceManager注冊ApplicationMaster
分配并啟動Executeors。在啟動Executeors之前，先要通過yarnAllocator獲取到numExecutors個Container，然后在Container中啟動Executeors。
那么這個Application將失敗，將Application Status標(biāo)明為FAILED，并將關(guān)閉SparkContext。其實(shí)，啟動Executeors是通過ExecutorRunnable實(shí)現(xiàn)的，而ExecutorRunnable內(nèi)部是啟動CoarseGrainedExecutorBackend的。
最后，Task將在CoarseGrainedExecutorBackend里面運(yùn)行，然后運(yùn)行狀況會通過Akka通知CoarseGrainedScheduler，直到作業(yè)運(yùn)行完成。
Spark節(jié)點(diǎn)的概念
一、Spark驅(qū)動器是執(zhí)行程序中的main()方法的進(jìn)程。它執(zhí)行用戶編寫的用來創(chuàng)建SparkContext(初始化)、創(chuàng)建RDD，以及運(yùn)行RDD的轉(zhuǎn)化操作和行動操作的代碼。
驅(qū)動器節(jié)點(diǎn)driver的職責(zé)：
把用戶程序轉(zhuǎn)為任務(wù)task(driver)
Spark驅(qū)動器程序負(fù)責(zé)把用戶程序轉(zhuǎn)化為多個物理執(zhí)行單元，這些單元也被稱之為任務(wù)task(詳解見備注)
為執(zhí)行器節(jié)點(diǎn)調(diào)度任務(wù)(executor)
有了物理計(jì)劃之后，Spark驅(qū)動器在各個執(zhí)行器節(jié)點(diǎn)進(jìn)程間協(xié)調(diào)任務(wù)的調(diào)度。Spark驅(qū)動器程序會根據(jù)當(dāng)前的執(zhí)行器節(jié)點(diǎn)，把所有任務(wù)基于數(shù)據(jù)所在位置分配給合適的執(zhí)行器進(jìn)程。當(dāng)執(zhí)行任務(wù)時，執(zhí)行器進(jìn)程會把緩存的數(shù)據(jù)存儲起來，而驅(qū)動器進(jìn)程同樣會跟蹤這些緩存數(shù)據(jù)的位置，并利用這些位置信息來調(diào)度以后的任務(wù)，以盡量減少數(shù)據(jù)的網(wǎng)絡(luò)傳輸。（就是所謂的移動計(jì)算，而不移動數(shù)據(jù))。
二、執(zhí)行器節(jié)點(diǎn)
作用：
負(fù)責(zé)運(yùn)行組成Spark應(yīng)用的任務(wù)，并將結(jié)果返回給驅(qū)動器進(jìn)程；
通過自身的塊管理器(blockManager)為用戶程序中要求緩存的RDD提供內(nèi)存式存儲。RDD是直接緩存在執(zhí)行器進(jìn)程內(nèi)的，因此任務(wù)可以在運(yùn)行時充分利用緩存數(shù)據(jù)加快運(yùn)算。
驅(qū)動器的職責(zé)：
所有的Spark程序都遵循同樣的結(jié)構(gòu)：程序從輸入數(shù)據(jù)創(chuàng)建一系列RDD，再使用轉(zhuǎn)化操作派生成新的RDD，最后使用行動操作手機(jī)或存儲結(jié)果RDD，Spark程序其實(shí)是隱式地創(chuàng)建出了一個由操作組成的邏輯上的有向無環(huán)圖DAG。當(dāng)驅(qū)動器程序執(zhí)行時，它會把這個邏輯圖轉(zhuǎn)為物理執(zhí)行計(jì)劃。
這樣 Spark就把邏輯計(jì)劃轉(zhuǎn)為一系列步驟(stage)，而每個步驟又由多個任務(wù)組成。這些任務(wù)會被打包送到集群中。
Spark初始化
每個Spark應(yīng)用都由一個驅(qū)動器程序來發(fā)起集群上的各種并行操作。驅(qū)動器程序包含應(yīng)用的main函數(shù)，并且定義了集群上的分布式數(shù)據(jù)集，以及對該分布式數(shù)據(jù)集應(yīng)用了相關(guān)操作。
驅(qū)動器程序通過一個SparkContext對象來訪問spark,這個對象代表對計(jì)算集群的一個連接。（比如在sparkshell啟動時已經(jīng)自動創(chuàng)建了一個SparkContext對象，是一個叫做SC的變量。(下圖，查看變量sc)
一旦創(chuàng)建了sparkContext，就可以用它來創(chuàng)建RDD。比如調(diào)用sc.textFile()來創(chuàng)建一個代表文本中各行文本的RDD。（比如vallinesRDD = sc.textFile(“yangsy.text”),val spark = linesRDD.filter(line=>line.contains(“spark”),spark.count()）
執(zhí)行這些操作，驅(qū)動器程序一般要管理多個執(zhí)行器,就是我們所說的executor節(jié)點(diǎn)。
在初始化SparkContext的同時，加載sparkConf對象來加載集群的配置，從而創(chuàng)建sparkContext對象。
從源碼中可以看到，在啟動thriftserver時，調(diào)用了spark- daemon.sh文件，該文件源碼如左圖，加載spark_home下的conf中的文件。
（在執(zhí)行后臺代碼時，需要首先創(chuàng)建conf對象，加載相應(yīng)參數(shù)， val sparkConf = newSparkConf().setMaster("local").setAppName("cocapp").set("spark.executor.memory","1g"), val sc: SparkContext = new SparkContext(sparkConf))
RDD工作原理：
RDD(Resilient DistributedDatasets)[1] ,彈性分布式數(shù)據(jù)集，是分布式內(nèi)存的一個抽象概念，RDD提供了一種高度受限的共享內(nèi)存模型，即RDD是只讀的記錄分區(qū)的集合，只能通過在其他RDD執(zhí)行確定的轉(zhuǎn)換操作（如map、join和group by）而創(chuàng)建，然而這些限制使得實(shí)現(xiàn)容錯的開銷很低。對開發(fā)者而言，RDD可以看作是Spark的一個對象，它本身運(yùn)行于內(nèi)存中，如讀文件是一個RDD，對文件計(jì)算是一個RDD，結(jié)果集也是一個RDD ，不同的分片、數(shù)據(jù)之間的依賴、key-value類型的map數(shù)據(jù)都可以看做RDD。
主要分為三部分：創(chuàng)建RDD對象，DAG調(diào)度器創(chuàng)建執(zhí)行計(jì)劃，Task調(diào)度器分配任務(wù)并調(diào)度Worker開始運(yùn)行。
SparkContext(RDD相關(guān)操作)→通過(提交作業(yè))→(遍歷RDD拆分stage→生成作業(yè))DAGScheduler→通過（提交任務(wù)集）→任務(wù)調(diào)度管理(TaskScheduler)→通過（按照資源獲取任務(wù))→任務(wù)調(diào)度管理(TaskSetManager)
Transformation返回值還是一個RDD。它使用了鏈?zhǔn)秸{(diào)用的設(shè)計(jì)模式，對一個RDD進(jìn)行計(jì)算后，變換成另外一個RDD，然后這個RDD又可以進(jìn)行另外一次轉(zhuǎn)換。這個過程是分布式的。
Action返回值不是一個RDD。它要么是一個Scala的普通集合，要么是一個值，要么是空，最終或返回到Driver程序，或把RDD寫入到文件系統(tǒng)中
轉(zhuǎn)換(Transformations)(如：map, filter, groupBy, join等)，Transformations操作是Lazy的，也就是說從一個RDD轉(zhuǎn)換生成另一個RDD的操作不是馬上執(zhí)行，Spark在遇到Transformations操作時只會記錄需要這樣的操作，并不會去執(zhí)行，需要等到有Actions操作的時候才會真正啟動計(jì)算過程進(jìn)行計(jì)算。
操作(Actions)(如：count, collect, save等)，Actions操作會返回結(jié)果或把RDD數(shù)據(jù)寫到存儲系統(tǒng)中。Actions是觸發(fā)Spark啟動計(jì)算的動因。
它們本質(zhì)區(qū)別是：Transformation返回值還是一個RDD。它使用了鏈?zhǔn)秸{(diào)用的設(shè)計(jì)模式，對一個RDD進(jìn)行計(jì)算后，變換成另外一個RDD，然后這個RDD又可以進(jìn)行另外一次轉(zhuǎn)換。這個過程是分布式的。Action返回值不是一個RDD。它要么是一個Scala的普通集合，要么是一個值，要么是空，最終或返回到Driver程序，或把RDD寫入到文件系統(tǒng)中。關(guān)于這兩個動作，在Spark開發(fā)指南中會有就進(jìn)一步的詳細(xì)介紹，它們是基于Spark開發(fā)的核心。
RDD基礎(chǔ)
Spark中的RDD就是一個不可變的分布式對象集合。每個RDD都被分為多個分區(qū)，這些分區(qū)運(yùn)行在集群的不同節(jié)點(diǎn)上。創(chuàng)建RDD的方法有兩種：一種是讀取一個外部數(shù)據(jù)集；一種是在群東程序里分發(fā)驅(qū)動器程序中的對象集合，不如剛才的示例，讀取文本文件作為一個字符串的RDD的示例。
創(chuàng)建出來后，RDD支持兩種類型的操作:轉(zhuǎn)化操作和行動操作
轉(zhuǎn)化操作會由一個RDD生成一個新的RDD。（比如剛才的根據(jù)謂詞篩選）
行動操作會對RDD計(jì)算出一個結(jié)果，并把結(jié)果返回到驅(qū)動器程序中，或把結(jié)果存儲到外部存儲系統(tǒng)（比如HDFS）中。比如first()操作就是一個行動操作，會返回RDD的第一個元素。
注：轉(zhuǎn)化操作與行動操作的區(qū)別在于Spark計(jì)算RDD的方式不同。雖然你可以在任何時候定義一個新的RDD，但Spark只會惰性計(jì)算這些RDD。它們只有第一個在一個行動操作中用到時，才會真正的計(jì)算。之所以這樣設(shè)計(jì)，是因?yàn)楸热鐒偛耪{(diào)用sc.textFile(...)時就把文件中的所有行都讀取并存儲起來，就會消耗很多存儲空間，而我們馬上又要篩選掉其中的很多數(shù)據(jù)。
這里還需要注意的一點(diǎn)是，spark會在你每次對它們進(jìn)行行動操作時重新計(jì)算。如果想在多個行動操作中重用同一個RDD，那么可以使用RDD.persist()或RDD.collect()讓Spark把這個RDD緩存下來。（可以是內(nèi)存，也可以是磁盤)
Spark會使用譜系圖來記錄這些不同RDD之間的依賴關(guān)系，Spark需要用這些信息來按需計(jì)算每個RDD，也可以依靠譜系圖在持久化的RDD丟失部分?jǐn)?shù)據(jù)時用來恢復(fù)所丟失的數(shù)據(jù)。(如下圖，過濾errorsRDD與warningsRDD,最終調(diào)用union()函數(shù))
RDD計(jì)算方式
RDD的寬窄依賴
窄依賴 (narrowdependencies) 和寬依賴 (widedependencies) 。窄依賴是指 父 RDD 的每個分區(qū)都只被子 RDD 的一個分區(qū)所使用 。相應(yīng)的，那么寬依賴就是指父 RDD 的分區(qū)被多個子 RDD 的分區(qū)所依賴。例如， map 就是一種窄依賴，而 join 則會導(dǎo)致寬依賴
這種劃分有兩個用處。首先，窄依賴支持在一個結(jié)點(diǎn)上管道化執(zhí)行。例如基于一對一的關(guān)系，可以在 filter 之后執(zhí)行 map 。其次，窄依賴支持更高效的故障還原。因?yàn)閷τ谡蕾嚕挥衼G失的父 RDD 的分區(qū)需要重新計(jì)算。而對于寬依賴，一個結(jié)點(diǎn)的故障可能導(dǎo)致來自所有父 RDD 的分區(qū)丟失，因此就需要完全重新執(zhí)行。因此對于寬依賴，Spark 會在持有各個父分區(qū)的結(jié)點(diǎn)上，將中間數(shù)據(jù)持久化來簡化故障還原，就像 MapReduce 會持久化 map 的輸出一樣。
SparkExample
步驟 1 ：創(chuàng)建 RDD 。上面的例子除去最后一個 collect 是個動作，不會創(chuàng)建 RDD 之外，前面四個轉(zhuǎn)換都會創(chuàng)建出新的 RDD 。因此第一步就是創(chuàng)建好所有 RDD( 內(nèi)部的五項(xiàng)信息 ) 。
步驟 2 ：創(chuàng)建執(zhí)行計(jì)劃。Spark 會盡可能地管道化，并基于是否要重新組織數(shù)據(jù)來劃分 階段 (stage) ，例如本例中的 groupBy() 轉(zhuǎn)換就會將整個執(zhí)行計(jì)劃劃分成兩階段執(zhí)行。最終會產(chǎn)生一個 DAG(directedacyclic graph ，有向無環(huán)圖 ) 作為邏輯執(zhí)行計(jì)劃。
步驟 3 ：調(diào)度任務(wù)。 將各階段劃分成不同的 任務(wù) (task) ，每個任務(wù)都是數(shù)據(jù)和計(jì)算的合體。在進(jìn)行下一階段前，當(dāng)前階段的所有任務(wù)都要執(zhí)行完成。因?yàn)橄乱浑A段的第一個轉(zhuǎn)換一定是重新組織數(shù)據(jù)的，所以必須等當(dāng)前階段所有結(jié)果數(shù)據(jù)都計(jì)算出來了才能繼續(xù)。
假設(shè)本例中的 hdfs://names 下有四個文件塊，那么 HadoopRDD 中 partitions 就會有四個分區(qū)對應(yīng)這四個塊數(shù)據(jù)，同時 preferedLocations 會指明這四個塊的最佳位置。現(xiàn)在，就可以創(chuàng)建出四個任務(wù)，并調(diào)度到合適的集群結(jié)點(diǎn)上。
Spark數(shù)據(jù)分區(qū)
Spark的特性是對數(shù)據(jù)集在節(jié)點(diǎn)間的分區(qū)進(jìn)行控制。在分布式系統(tǒng)中，通訊的代價是巨大的，控制數(shù)據(jù)分布以獲得最少的網(wǎng)絡(luò)傳輸可以極大地提升整體性能。Spark程序可以通過控制RDD分區(qū)方式來減少通訊的開銷。
Spark中所有的鍵值對RDD都可以進(jìn)行分區(qū)。確保同一組的鍵出現(xiàn)在同一個節(jié)點(diǎn)上。比如，使用哈希分區(qū)將一個RDD分成了100個分區(qū)，此時鍵的哈希值對100取模的結(jié)果相同的記錄會被放在一個節(jié)點(diǎn)上。
（可使用partitionBy(newHashPartitioner(100)).persist()來構(gòu)造100個分區(qū))
Spark中的許多操作都引入了將數(shù)據(jù)根據(jù)鍵跨界點(diǎn)進(jìn)行混洗的過程。(比如：join(),leftOuterJoin(),groupByKey(),reducebyKey()等)對于像reduceByKey()這樣只作用于單個RDD的操作，運(yùn)行在未分區(qū)的RDD上的時候會導(dǎo)致每個鍵的所有對應(yīng)值都在每臺機(jī)器上進(jìn)行本地計(jì)算。
SparkSQL的shuffle過程
Spark SQL的核心是把已有的RDD，帶上Schema信息，然后注冊成類似sql里的”Table”，對其進(jìn)行sql查詢。這里面主要分兩部分，一是生成SchemaRD，二是執(zhí)行查詢。
如果是spark-hive項(xiàng)目，那么讀取metadata信息作為Schema、讀取hdfs上數(shù)據(jù)的過程交給Hive完成，然后根據(jù)這倆部分生成SchemaRDD，在HiveContext下進(jìn)行hql()查詢。
SparkSQL結(jié)構(gòu)化數(shù)據(jù)
首先說一下ApacheHive，Hive可以在HDFS內(nèi)或者在其他存儲系統(tǒng)上存儲多種格式的表。SparkSQL可以讀取Hive支持的任何表。要把Spark SQL連接已有的hive上，需要提供Hive的配置文件。hive-site.xml文件復(fù)制到spark的conf文件夾下。再創(chuàng)建出HiveContext對象(sparksql的入口)，然后就可以使用HQL來對表進(jìn)行查詢，并以由行足證的RDD的形式拿到返回的數(shù)據(jù)。
創(chuàng)建Hivecontext并查詢數(shù)據(jù)
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字段
通過jdbc連接外部數(shù)據(jù)源更新與加載
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()
//加載外部數(shù)據(jù)源數(shù)據(jù)到內(nèi)存
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解析
首先說下傳統(tǒng)數(shù)據(jù)庫的解析，傳統(tǒng)數(shù)據(jù)庫的解析過程是按Rusult、Data Source、Operation的次序來解析的。傳統(tǒng)數(shù)據(jù)庫先將讀入的SQL語句進(jìn)行解析，分辨出SQL語句中哪些詞是關(guān)鍵字（如select,from,where)，哪些是表達(dá)式，哪些是Projection，哪些是Data Source等等。進(jìn)一步判斷SQL語句是否規(guī)范，不規(guī)范就報錯，規(guī)范則按照下一步過程綁定（Bind)。過程綁定是將SQL語句和數(shù)據(jù)庫的數(shù)據(jù)字典(列,表,視圖等）進(jìn)行綁定，如果相關(guān)的Projection、Data Source等都存在，就表示這個SQL語句是可以執(zhí)行的。在執(zhí)行過程中，有時候甚至不需要讀取物理表就可以返回結(jié)果，比如重新運(yùn)行剛運(yùn)行過的SQL語句，直接從數(shù)據(jù)庫的緩沖池中獲取返回結(jié)果。在數(shù)據(jù)庫解析的過程中SQL語句時，將會把SQL語句轉(zhuǎn)化成一個樹形結(jié)構(gòu)來進(jìn)行處理，會形成一個或含有多個節(jié)點(diǎn)(TreeNode)的Tree,然后再后續(xù)的處理政對該Tree進(jìn)行一系列的操作。
Spark SQL對SQL語句的處理和關(guān)系數(shù)據(jù)庫對SQL語句的解析采用了類似的方法，首先會將SQL語句進(jìn)行解析，然后形成一個Tree，后續(xù)如綁定、優(yōu)化等處理過程都是對Tree的操作，而操作方法是采用Rule,通過模式匹配，對不同類型的節(jié)點(diǎn)采用不同的操作。SparkSQL有兩個分支，sqlContext和hiveContext。sqlContext現(xiàn)在只支持SQL語法解析器（Catalyst)，hiveContext支持SQL語法和HiveContext語法解析器。
原文地址：http://mt.sohu.com/20160522/n450849016.shtml