Feeling

　　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語法解析器。

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

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

參數可以通過spark_home/conf/spark-default.conf配置文件設置。

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

下面是官網的相關描述：

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）

歸并排序最差運行時間是O(nlogn)，它是利用遞歸設計程序的典型例子。

歸并排序的最基礎的操作就是合并兩個已經排好序的序列。

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



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

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

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

（1）首先設定一個新的數列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]復制到C中，那么C[7] = 87。歸并完成。

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

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

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

2 堆排序（HeapSort）

堆排序屬于百萬俱樂部的成員。它特別適合超大數據量（百萬條記錄以上）的排序。因為它并不使用遞歸（因為超大數據量的遞歸可能會導致堆棧溢出），而且它的時間也是O(nlogn)。還有它并不需要大量的額外存儲空間。

堆排序的思路是:

(1)將原始未排序的數據建成一個堆。
(2)建成堆以后，最大值在堆頂，也就是第0個元素，這時候將第零個元素和最后一個元素交換。
(3)這時候將從0到倒數第二個元素的所有數據當成一個新的序列，建一個新的堆，再次交換第一個和最后一個元素，依次類推，就可以將所有元素排序完畢。

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


堆排序的具體算法如下：