Feeling

　　Spark簡(jiǎn)介

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

　　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é)束整個(gè)運(yùn)行期間的過程分析。

　　一、客戶端進(jìn)行操作

1. 根據(jù)yarnConf來初始化yarnClient，并啟動(dòng)yarnClient

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

3. 設(shè)置資源、環(huán)境變量：其中包括了設(shè)置Application的Staging目錄、準(zhǔn)備本地資源（jar文件、log4j.properties）、設(shè)置Application其中的環(huán)境變量、創(chuàng)建Container啟動(dòng)的Context等；

4. 設(shè)置Application提交的Context，包括設(shè)置應(yīng)用的名字、隊(duì)列、AM的申請(qǐng)的Container、標(biāo)記該作業(yè)的類型為Spark；

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

　　當(dāng)作業(yè)提交到Y(jié)ARN上之后，客戶端就沒事了，甚至在終端關(guān)掉那個(gè)進(jìn)程也沒事，因?yàn)檎麄€(gè)作業(yè)運(yùn)行在YARN集群上進(jìn)行，運(yùn)行的結(jié)果將會(huì)保存到HDFS或者日志中。

　　二、提交到Y(jié)ARN集群，YARN操作

1. 運(yùn)行ApplicationMaster的run方法；

2. 設(shè)置好相關(guān)的環(huán)境變量。

3. 創(chuàng)建amClient，并啟動(dòng)；

4. 在Spark UI啟動(dòng)之前設(shè)置Spark UI的AmIpFilter；

5. 在startUserClass函數(shù)專門啟動(dòng)了一個(gè)線程（名稱為Driver的線程）來啟動(dòng)用戶提交的Application，也就是啟動(dòng)了Driver。在Driver中將會(huì)初始化SparkContext；

6. 等待SparkContext初始化完成，最多等待spark.yarn.applicationMaster.waitTries次數(shù)（默認(rèn)為10），如果等待了的次數(shù)超過了配置的，程序?qū)?huì)退出；否則用SparkContext初始化yarnAllocator；

7. 當(dāng)SparkContext、Driver初始化完成的時(shí)候，通過amClient向ResourceManager注冊(cè)ApplicationMaster

8. 分配并啟動(dòng)Executeors。在啟動(dòng)Executeors之前，先要通過yarnAllocator獲取到numExecutors個(gè)Container，然后在Container中啟動(dòng)Executeors。

   　　那么這個(gè)Application將失敗，將Application Status標(biāo)明為FAILED，并將關(guān)閉SparkContext。其實(shí)，啟動(dòng)Executeors是通過ExecutorRunnable實(shí)現(xiàn)的，而ExecutorRunnable內(nèi)部是啟動(dòng)CoarseGrainedExecutorBackend的。

9. 最后，Task將在CoarseGrainedExecutorBackend里面運(yùn)行，然后運(yùn)行狀況會(huì)通過Akka通知CoarseGrainedScheduler，直到作業(yè)運(yùn)行完成。

　　Spark節(jié)點(diǎn)的概念

　　一、Spark驅(qū)動(dòng)器是執(zhí)行程序中的main()方法的進(jìn)程。它執(zhí)行用戶編寫的用來創(chuàng)建SparkContext(初始化)、創(chuàng)建RDD，以及運(yùn)行RDD的轉(zhuǎn)化操作和行動(dòng)操作的代碼。

　　驅(qū)動(dòng)器節(jié)點(diǎn)driver的職責(zé)：

1. 把用戶程序轉(zhuǎn)為任務(wù)task(driver)

   　　Spark驅(qū)動(dòng)器程序負(fù)責(zé)把用戶程序轉(zhuǎn)化為多個(gè)物理執(zhí)行單元，這些單元也被稱之為任務(wù)task(詳解見備注)

2. 為執(zhí)行器節(jié)點(diǎn)調(diào)度任務(wù)(executor)

   　　有了物理計(jì)劃之后，Spark驅(qū)動(dòng)器在各個(gè)執(zhí)行器節(jié)點(diǎn)進(jìn)程間協(xié)調(diào)任務(wù)的調(diào)度。Spark驅(qū)動(dòng)器程序會(huì)根據(jù)當(dāng)前的執(zhí)行器節(jié)點(diǎn)，把所有任務(wù)基于數(shù)據(jù)所在位置分配給合適的執(zhí)行器進(jìn)程。當(dāng)執(zhí)行任務(wù)時(shí)，執(zhí)行器進(jìn)程會(huì)把緩存的數(shù)據(jù)存儲(chǔ)起來，而驅(qū)動(dòng)器進(jìn)程同樣會(huì)跟蹤這些緩存數(shù)據(jù)的位置，并利用這些位置信息來調(diào)度以后的任務(wù)，以盡量減少數(shù)據(jù)的網(wǎng)絡(luò)傳輸。（就是所謂的移動(dòng)計(jì)算，而不移動(dòng)數(shù)據(jù))。

　　二、執(zhí)行器節(jié)點(diǎn)

　　作用：

1. 負(fù)責(zé)運(yùn)行組成Spark應(yīng)用的任務(wù)，并將結(jié)果返回給驅(qū)動(dòng)器進(jìn)程；

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

　　驅(qū)動(dòng)器的職責(zé)：

　　所有的Spark程序都遵循同樣的結(jié)構(gòu)：程序從輸入數(shù)據(jù)創(chuàng)建一系列RDD，再使用轉(zhuǎn)化操作派生成新的RDD，最后使用行動(dòng)操作手機(jī)或存儲(chǔ)結(jié)果RDD，Spark程序其實(shí)是隱式地創(chuàng)建出了一個(gè)由操作組成的邏輯上的有向無環(huán)圖DAG。當(dāng)驅(qū)動(dòng)器程序執(zhí)行時(shí)，它會(huì)把這個(gè)邏輯圖轉(zhuǎn)為物理執(zhí)行計(jì)劃。

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

　　Spark初始化

1. 每個(gè)Spark應(yīng)用都由一個(gè)驅(qū)動(dòng)器程序來發(fā)起集群上的各種并行操作。驅(qū)動(dòng)器程序包含應(yīng)用的main函數(shù)，并且定義了集群上的分布式數(shù)據(jù)集，以及對(duì)該分布式數(shù)據(jù)集應(yīng)用了相關(guān)操作。

2. 驅(qū)動(dòng)器程序通過一個(gè)SparkContext對(duì)象來訪問spark,這個(gè)對(duì)象代表對(duì)計(jì)算集群的一個(gè)連接。（比如在sparkshell啟動(dòng)時(shí)已經(jīng)自動(dòng)創(chuàng)建了一個(gè)SparkContext對(duì)象，是一個(gè)叫做SC的變量。(下圖，查看變量sc)

   　　

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

   　　執(zhí)行這些操作，驅(qū)動(dòng)器程序一般要管理多個(gè)執(zhí)行器,就是我們所說的executor節(jié)點(diǎn)。

4. 在初始化SparkContext的同時(shí)，加載sparkConf對(duì)象來加載集群的配置，從而創(chuàng)建sparkContext對(duì)象。

   　　從源碼中可以看到，在啟動(dòng)thriftserver時(shí)，調(diào)用了spark- daemon.sh文件，該文件源碼如左圖，加載spark_home下的conf中的文件。

   　　

   　　（在執(zhí)行后臺(tái)代碼時(shí)，需要首先創(chuàng)建conf對(duì)象，加載相應(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)存的一個(gè)抽象概念，RDD提供了一種高度受限的共享內(nèi)存模型，即RDD是只讀的記錄分區(qū)的集合，只能通過在其他RDD執(zhí)行確定的轉(zhuǎn)換操作（如map、join和group by）而創(chuàng)建，然而這些限制使得實(shí)現(xiàn)容錯(cuò)的開銷很低。對(duì)開發(fā)者而言，RDD可以看作是Spark的一個(gè)對(duì)象，它本身運(yùn)行于內(nèi)存中，如讀文件是一個(gè)RDD，對(duì)文件計(jì)算是一個(gè)RDD，結(jié)果集也是一個(gè)RDD ，不同的分片、數(shù)據(jù)之間的依賴、key-value類型的map數(shù)據(jù)都可以看做RDD。

　　主要分為三部分：創(chuàng)建RDD對(duì)象，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返回值還是一個(gè)RDD。它使用了鏈?zhǔn)秸{(diào)用的設(shè)計(jì)模式，對(duì)一個(gè)RDD進(jìn)行計(jì)算后，變換成另外一個(gè)RDD，然后這個(gè)RDD又可以進(jìn)行另外一次轉(zhuǎn)換。這個(gè)過程是分布式的。

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

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

　　操作(Actions)(如：count, collect, save等)，Actions操作會(huì)返回結(jié)果或把RDD數(shù)據(jù)寫到存儲(chǔ)系統(tǒng)中。Actions是觸發(fā)Spark啟動(dòng)計(jì)算的動(dòng)因。

　　它們本質(zhì)區(qū)別是：Transformation返回值還是一個(gè)RDD。它使用了鏈?zhǔn)秸{(diào)用的設(shè)計(jì)模式，對(duì)一個(gè)RDD進(jìn)行計(jì)算后，變換成另外一個(gè)RDD，然后這個(gè)RDD又可以進(jìn)行另外一次轉(zhuǎn)換。這個(gè)過程是分布式的。Action返回值不是一個(gè)RDD。它要么是一個(gè)Scala的普通集合，要么是一個(gè)值，要么是空，最終或返回到Driver程序，或把RDD寫入到文件系統(tǒng)中。關(guān)于這兩個(gè)動(dòng)作，在Spark開發(fā)指南中會(huì)有就進(jìn)一步的詳細(xì)介紹，它們是基于Spark開發(fā)的核心。

　　RDD基礎(chǔ)

1. Spark中的RDD就是一個(gè)不可變的分布式對(duì)象集合。每個(gè)RDD都被分為多個(gè)分區(qū)，這些分區(qū)運(yùn)行在集群的不同節(jié)點(diǎn)上。創(chuàng)建RDD的方法有兩種：一種是讀取一個(gè)外部數(shù)據(jù)集；一種是在群東程序里分發(fā)驅(qū)動(dòng)器程序中的對(duì)象集合，不如剛才的示例，讀取文本文件作為一個(gè)字符串的RDD的示例。

2. 創(chuàng)建出來后，RDD支持兩種類型的操作:轉(zhuǎn)化操作和行動(dòng)操作

   　　轉(zhuǎn)化操作會(huì)由一個(gè)RDD生成一個(gè)新的RDD。（比如剛才的根據(jù)謂詞篩選）

   　　行動(dòng)操作會(huì)對(duì)RDD計(jì)算出一個(gè)結(jié)果，并把結(jié)果返回到驅(qū)動(dòng)器程序中，或把結(jié)果存儲(chǔ)到外部存儲(chǔ)系統(tǒng)（比如HDFS）中。比如first()操作就是一個(gè)行動(dòng)操作，會(huì)返回RDD的第一個(gè)元素。

   　　注：轉(zhuǎn)化操作與行動(dòng)操作的區(qū)別在于Spark計(jì)算RDD的方式不同。雖然你可以在任何時(shí)候定義一個(gè)新的RDD，但Spark只會(huì)惰性計(jì)算這些RDD。它們只有第一個(gè)在一個(gè)行動(dòng)操作中用到時(shí)，才會(huì)真正的計(jì)算。之所以這樣設(shè)計(jì)，是因?yàn)楸热鐒偛耪{(diào)用sc.textFile(...)時(shí)就把文件中的所有行都讀取并存儲(chǔ)起來，就會(huì)消耗很多存儲(chǔ)空間，而我們馬上又要篩選掉其中的很多數(shù)據(jù)。

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

3. Spark會(huì)使用譜系圖來記錄這些不同RDD之間的依賴關(guān)系，Spark需要用這些信息來按需計(jì)算每個(gè)RDD，也可以依靠譜系圖在持久化的RDD丟失部分?jǐn)?shù)據(jù)時(shí)用來恢復(fù)所丟失的數(shù)據(jù)。(如下圖，過濾errorsRDD與warningsRDD,最終調(diào)用union()函數(shù))

   　　

　　RDD計(jì)算方式

　　RDD的寬窄依賴

　　窄依賴 (narrowdependencies) 和寬依賴 (widedependencies) 。窄依賴是指 父 RDD 的每個(gè)分區(qū)都只被子 RDD 的一個(gè)分區(qū)所使用 。相應(yīng)的，那么寬依賴就是指父 RDD 的分區(qū)被多個(gè)子 RDD 的分區(qū)所依賴。例如， map 就是一種窄依賴，而 join 則會(huì)導(dǎo)致寬依賴

　　這種劃分有兩個(gè)用處。首先，窄依賴支持在一個(gè)結(jié)點(diǎn)上管道化執(zhí)行。例如基于一對(duì)一的關(guān)系，可以在 filter 之后執(zhí)行 map 。其次，窄依賴支持更高效的故障還原。因?yàn)閷?duì)于窄依賴，只有丟失的父 RDD 的分區(qū)需要重新計(jì)算。而對(duì)于寬依賴，一個(gè)結(jié)點(diǎn)的故障可能導(dǎo)致來自所有父 RDD 的分區(qū)丟失，因此就需要完全重新執(zhí)行。因此對(duì)于寬依賴，Spark 會(huì)在持有各個(gè)父分區(qū)的結(jié)點(diǎn)上，將中間數(shù)據(jù)持久化來簡(jiǎn)化故障還原，就像 MapReduce 會(huì)持久化 map 的輸出一樣。

　　SparkExample

　　步驟 1 ：創(chuàng)建 RDD 。上面的例子除去最后一個(gè) collect 是個(gè)動(dòng)作，不會(huì)創(chuàng)建 RDD 之外，前面四個(gè)轉(zhuǎn)換都會(huì)創(chuàng)建出新的 RDD 。因此第一步就是創(chuàng)建好所有 RDD( 內(nèi)部的五項(xiàng)信息 ) 。

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

　　步驟 3 ：調(diào)度任務(wù)。 將各階段劃分成不同的 任務(wù) (task) ，每個(gè)任務(wù)都是數(shù)據(jù)和計(jì)算的合體。在進(jìn)行下一階段前，當(dāng)前階段的所有任務(wù)都要執(zhí)行完成。因?yàn)橄乱浑A段的第一個(gè)轉(zhuǎn)換一定是重新組織數(shù)據(jù)的，所以必須等當(dāng)前階段所有結(jié)果數(shù)據(jù)都計(jì)算出來了才能繼續(xù)。

　　假設(shè)本例中的 hdfs://names 下有四個(gè)文件塊，那么 HadoopRDD 中 partitions 就會(huì)有四個(gè)分區(qū)對(duì)應(yīng)這四個(gè)塊數(shù)據(jù)，同時(shí) preferedLocations 會(huì)指明這四個(gè)塊的最佳位置。現(xiàn)在，就可以創(chuàng)建出四個(gè)任務(wù)，并調(diào)度到合適的集群結(jié)點(diǎn)上。

　　Spark數(shù)據(jù)分區(qū)

1. Spark的特性是對(duì)數(shù)據(jù)集在節(jié)點(diǎn)間的分區(qū)進(jìn)行控制。在分布式系統(tǒng)中，通訊的代價(jià)是巨大的，控制數(shù)據(jù)分布以獲得最少的網(wǎng)絡(luò)傳輸可以極大地提升整體性能。Spark程序可以通過控制RDD分區(qū)方式來減少通訊的開銷。

2. Spark中所有的鍵值對(duì)RDD都可以進(jìn)行分區(qū)。確保同一組的鍵出現(xiàn)在同一個(gè)節(jié)點(diǎn)上。比如，使用哈希分區(qū)將一個(gè)RDD分成了100個(gè)分區(qū)，此時(shí)鍵的哈希值對(duì)100取模的結(jié)果相同的記錄會(huì)被放在一個(gè)節(jié)點(diǎn)上。

   　　（可使用partitionBy(newHashPartitioner(100)).persist()來構(gòu)造100個(gè)分區(qū))

3. Spark中的許多操作都引入了將數(shù)據(jù)根據(jù)鍵跨界點(diǎn)進(jìn)行混洗的過程。(比如：join(),leftOuterJoin(),groupByKey(),reducebyKey()等)對(duì)于像reduceByKey()這樣只作用于單個(gè)RDD的操作，運(yùn)行在未分區(qū)的RDD上的時(shí)候會(huì)導(dǎo)致每個(gè)鍵的所有對(duì)應(yīng)值都在每臺(tái)機(jī)器上進(jìn)行本地計(jì)算。

　　SparkSQL的shuffle過程

　　Spark SQL的核心是把已有的RDD，帶上Schema信息，然后注冊(cè)成類似sql里的”Table”，對(duì)其進(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ù)

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

2. 創(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字段

3. 通過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ù)庫(kù)的解析，傳統(tǒng)數(shù)據(jù)庫(kù)的解析過程是按Rusult、Data Source、Operation的次序來解析的。傳統(tǒng)數(shù)據(jù)庫(kù)先將讀入的SQL語句進(jìn)行解析，分辨出SQL語句中哪些詞是關(guān)鍵字（如select,from,where)，哪些是表達(dá)式，哪些是Projection，哪些是Data Source等等。進(jìn)一步判斷SQL語句是否規(guī)范，不規(guī)范就報(bào)錯(cuò)，規(guī)范則按照下一步過程綁定（Bind)。過程綁定是將SQL語句和數(shù)據(jù)庫(kù)的數(shù)據(jù)字典(列,表,視圖等）進(jìn)行綁定，如果相關(guān)的Projection、Data Source等都存在，就表示這個(gè)SQL語句是可以執(zhí)行的。在執(zhí)行過程中，有時(shí)候甚至不需要讀取物理表就可以返回結(jié)果，比如重新運(yùn)行剛運(yùn)行過的SQL語句，直接從數(shù)據(jù)庫(kù)的緩沖池中獲取返回結(jié)果。在數(shù)據(jù)庫(kù)解析的過程中SQL語句時(shí)，將會(huì)把SQL語句轉(zhuǎn)化成一個(gè)樹形結(jié)構(gòu)來進(jìn)行處理，會(huì)形成一個(gè)或含有多個(gè)節(jié)點(diǎn)(TreeNode)的Tree,然后再后續(xù)的處理政對(duì)該Tree進(jìn)行一系列的操作。

　　Spark SQL對(duì)SQL語句的處理和關(guān)系數(shù)據(jù)庫(kù)對(duì)SQL語句的解析采用了類似的方法，首先會(huì)將SQL語句進(jìn)行解析，然后形成一個(gè)Tree，后續(xù)如綁定、優(yōu)化等處理過程都是對(duì)Tree的操作，而操作方法是采用Rule,通過模式匹配，對(duì)不同類型的節(jié)點(diǎn)采用不同的操作。SparkSQL有兩個(gè)分支，sqlContext和hiveContext。sqlContext現(xiàn)在只支持SQL語法解析器（Catalyst)，hiveContext支持SQL語法和HiveContext語法解析器。

原文地址：http://mt.sohu.com/20160522/n450849016.shtml

spark中有partition的概念（和slice是同一個(gè)概念，在spark1.2中官網(wǎng)已經(jīng)做出了說明），一般每個(gè)partition對(duì)應(yīng)一個(gè)task。在我的測(cè)試過程中，如果沒有設(shè)置spark.default.parallelism參數(shù)，spark計(jì)算出來的partition非常巨大，與我的cores非常不搭。我在兩臺(tái)機(jī)器上（8cores *2 +6g * 2）上，spark計(jì)算出來的partition達(dá)到2.8萬個(gè)，也就是2.9萬個(gè)tasks，每個(gè)task完成時(shí)間都是幾毫秒或者零點(diǎn)幾毫秒，執(zhí)行起來非常緩慢。在我嘗試設(shè)置了 spark.default.parallelism 后，任務(wù)數(shù)減少到10，執(zhí)行一次計(jì)算過程從minute降到20second。

參數(shù)可以通過spark_home/conf/spark-default.conf配置文件設(shè)置。

eg.

spark.master                  spark://master:7077 

spark.default.parallelism     10 

spark.driver.memory           2g 

spark.serializer              org.apache.spark.serializer.KryoSerializer 

spark.sql.shuffle.partitions  50

下面是官網(wǎng)的相關(guān)描述：

from:http://spark.apache.org/docs/latest/configuration.html

from:http://spark.apache.org/docs/latest/tuning.html

Level of Parallelism

Clusters will not be fully utilized unless you set the level of parallelism for each operation high enough. Spark automatically sets the number of “map” tasks to run on each file according to its size (though you can control it through optional parameters to SparkContext.textFile, etc), and for distributed “reduce” operations, such as groupByKey and reduceByKey, it uses the largest parent RDD’s number of partitions. You can pass the level of parallelism as a second argument (see the spark.PairRDDFunctions documentation), or set the config propertyspark.default.parallelism to change the default. In general, we recommend 2-3 tasks per CPU core in your cluster.


原文地址：http://www.cnblogs.com/wrencai/p/4231966.html

1 歸并排序（MergeSort）

歸并排序最差運(yùn)行時(shí)間是O(nlogn)，它是利用遞歸設(shè)計(jì)程序的典型例子。

歸并排序的最基礎(chǔ)的操作就是合并兩個(gè)已經(jīng)排好序的序列。

假設(shè)我們有一個(gè)沒有排好序的序列，那么首先我們使用分割的辦法將這個(gè)序列分割成一個(gè)一個(gè)已經(jīng)排好序的子序列。然后再利用歸并的方法將一個(gè)個(gè)的子序列合并成排序好的序列。分割和歸并的過程可以看下面的圖例。



從上圖可以看出，我們首先把一個(gè)未排序的序列從中間分割成2部分，再把2部分分成4部分，依次分割下去，直到分割成一個(gè)一個(gè)的數(shù)據(jù)，再把這些數(shù)據(jù)兩兩歸并到一起，使之有序，不停的歸并，最后成為一個(gè)排好序的序列。

如何把兩個(gè)已經(jīng)排序好的子序列歸并成一個(gè)排好序的序列呢？可以參看下面的方法。

假設(shè)我們有兩個(gè)已經(jīng)排序好的子序列。
序列A：1 23 34 65
序列B：2 13 14 87
那么可以按照下面的步驟將它們歸并到一個(gè)序列中。

（1）首先設(shè)定一個(gè)新的數(shù)列C[8]。
（2）A[0]和B[0]比較，A[0] = 1，B[0] = 2，A[0] < B[0]，那么C[0] = 1
（3）A[1]和B[0]比較，A[1] = 23，B[0] = 2，A[1] > B[0]，那么C[1] = 2
（4）A[1]和B[1]比較，A[1] = 23，B[1] = 13，A[1] > B[1]，那么C[2] = 13
（5）A[1]和B[2]比較，A[1] = 23，B[2] = 14，A[1] > B[2]，那么C[3] = 14
（6）A[1]和B[3]比較，A[1] = 23，B[3] = 87，A[1] < B[3]，那么C[4] = 23
（7）A[2]和B[3]比較，A[2] = 34，B[3] = 87，A[2] < B[3]，那么C[5] = 34
（8）A[3]和B[3]比較，A[3] = 65，B[3] = 87，A[3] < B[3]，那么C[6] = 65
（9）最后將B[3]復(fù)制到C中，那么C[7] = 87。歸并完成。

如果我們清楚了上面的分割和歸并過程，那么我們就可以用遞歸的方法得到歸并算法的實(shí)現(xiàn)。

歸并排序算法是一種O(nlogn)的算法。它的最差，平均，最好時(shí)間都是O(nlogn)。但是它需要額外的存儲(chǔ)空間，這在某些內(nèi)存緊張的機(jī)器上會(huì)受到限制。

歸并算法是又分割和歸并兩部分組成的。對(duì)于分割部分，如果我們使用二分查找的話，時(shí)間是O(logn)，在最后歸并的時(shí)候，時(shí)間是O(n)，所以總的時(shí)間是O(nlogn)。

2 堆排序（HeapSort）

堆排序?qū)儆诎偃f俱樂部的成員。它特別適合超大數(shù)據(jù)量（百萬條記錄以上）的排序。因?yàn)樗⒉皇褂眠f歸（因?yàn)槌髷?shù)據(jù)量的遞歸可能會(huì)導(dǎo)致堆棧溢出），而且它的時(shí)間也是O(nlogn)。還有它并不需要大量的額外存儲(chǔ)空間。

堆排序的思路是:

(1)將原始未排序的數(shù)據(jù)建成一個(gè)堆。
(2)建成堆以后，最大值在堆頂，也就是第0個(gè)元素，這時(shí)候?qū)⒌诹銈€(gè)元素和最后一個(gè)元素交換。
(3)這時(shí)候?qū)?到倒數(shù)第二個(gè)元素的所有數(shù)據(jù)當(dāng)成一個(gè)新的序列，建一個(gè)新的堆，再次交換第一個(gè)和最后一個(gè)元素，依次類推，就可以將所有元素排序完畢。

建立堆的過程如下面的圖所示:


堆排序的具體算法如下：