[[189321]]

Spark：

Apache Spark 是專為大規(guī)模數(shù)據(jù)處理而設(shè)計(jì)的快速通用的計(jì)算引擎。

Spark是UC Berkeley AMP lab (加州大學(xué)伯克利分校的AMP實(shí)驗(yàn)室)所開源的類Hadoop MapReduce的通用并行框架，Spark擁有Hadoop MapReduce所具有的優(yōu)點(diǎn)；但不同于MapReduce的是Job中間輸出結(jié)果可以保存在內(nèi)存中，從而不再需要讀寫HDFS，因此Spark能更好地適用于數(shù)據(jù)挖掘與機(jī)器學(xué)習(xí)等需要迭代的MapReduce的算法。

Spark 是一種與 Hadoop 相似的開源集群計(jì)算環(huán)境，但是兩者之間還存在一些不同之處，這些有用的不同之處使 Spark 在某些工作負(fù)載方面表現(xiàn)得更加優(yōu)越，換句話說，Spark 啟用了內(nèi)存分布數(shù)據(jù)集，除了能夠提供交互式查詢外，它還可以優(yōu)化迭代工作負(fù)載。

Spark 是在 Scala 語言中實(shí)現(xiàn)的，它將 Scala 用作其應(yīng)用程序框架。與 Hadoop 不同，Spark 和 Scala 能夠緊密集成，其中的 Scala 可以像操作本地集合對(duì)象一樣輕松地操作分布式數(shù)據(jù)集。

盡管創(chuàng)建 Spark 是為了支持分布式數(shù)據(jù)集上的迭代作業(yè)，但是實(shí)際上它是對(duì) Hadoop 的補(bǔ)充，可以在 Hadoop 文件系統(tǒng)中并行運(yùn)行。通過名為 Mesos 的第三方集群框架可以支持此行為。Spark 由加州大學(xué)伯克利分校 AMP 實(shí)驗(yàn)室 (Algorithms, Machines, and People Lab) 開發(fā)，可用來構(gòu)建大型的、低延遲的數(shù)據(jù)分析應(yīng)用程序。

Spark的性能特點(diǎn)：

1.更快的速度：內(nèi)存計(jì)算下，Spark 比 Hadoop 快100倍。

• 內(nèi)存計(jì)算引擎，提供Cache機(jī)制來支持需要反復(fù)迭代計(jì)算或者多次數(shù)據(jù)共享，減少數(shù)據(jù)讀取的I/O開銷
• DAG引擎，減少多次計(jì)算之間中間結(jié)果寫到HDFS的開銷;
• 使用多線程池模型來減少task啟動(dòng)開銷，shuffle過程中避免不必要的sort操作已經(jīng)減少磁盤I/O操作;

2.易用性：

• Spark 提供了80多個(gè)高級(jí)運(yùn)算符。
• 提供了豐富的API，支持JAVA,Scala,Python和R四種語言;
• 代碼量比MapReduce少2~5倍;

3.通用性：Spark 提供了大量的庫，包括SQL、DataFrames、MLlib、GraphX、Spark Streaming。 開發(fā)者可以在同一個(gè)應(yīng)用程序中無縫組合使用這些庫。

4.支持多種資源管理器：Spark 支持 Hadoop YARN，Apache Mesos，及其自帶的獨(dú)立集群管理器

Spark基本原理：

Spark Streaming：構(gòu)建在Spark上處理Stream數(shù)據(jù)的框架，基本的原理是將Stream數(shù)據(jù)分成小的時(shí)間片斷(幾秒)，以類似batch批量處理的方式來處理這小部分?jǐn)?shù)據(jù)。Spark Streaming構(gòu)建在Spark上，一方面是因?yàn)镾park的低延遲執(zhí)行引擎(100ms+)，雖然比不上專門的流式數(shù)據(jù)處理軟件，也可以用于實(shí)時(shí)計(jì)算，另一方面相比基于Record的其它處理框架(如Storm)，一部分窄依賴的RDD數(shù)據(jù)集可以從源數(shù)據(jù)重新計(jì)算達(dá)到容錯(cuò)處理目的。此外小批量處理的方式使得它可以同時(shí)兼容批量和實(shí)時(shí)數(shù)據(jù)處理的邏輯和算法。方便了一些需要?dú)v史數(shù)據(jù)和實(shí)時(shí)數(shù)據(jù)聯(lián)合分析的特定應(yīng)用場(chǎng)合。

Spark背景：

1.MapReduce局限性:

1.僅支持Map和Reduce兩種操作;

2.處理效率低效;不適合迭代計(jì)算(如機(jī)器學(xué)習(xí)、圖計(jì)算等)，交互式處理(數(shù)據(jù)挖掘)和流失處理(日志分析)

3.Map中間結(jié)果需要寫磁盤，Reduce寫HDFS，多個(gè)MR之間通過HDFS交換數(shù)據(jù);

4.任務(wù)調(diào)度和啟動(dòng)開銷大;

5.無法充分利用內(nèi)存;(與MR產(chǎn)生時(shí)代有關(guān)，MR出現(xiàn)時(shí)內(nèi)存價(jià)格比較高，采用磁盤存儲(chǔ)代價(jià)小)

6.Map端和Reduce端均需要排序;

7.MapReduce編程不夠靈活。(比較Scala函數(shù)式編程而言)

8.框架多樣化[采用一種框架技術(shù)(Spark)同時(shí)實(shí)現(xiàn)批處理、流式計(jì)算、交互式計(jì)算]：

• 批處理：MapReduce、Hive、Pig;
• 流式計(jì)算:Storm
• 交互式計(jì)算:Impala

Spark核心概念：

• RDD：Resilient Distributed Datasets,彈性分布式數(shù)據(jù)集

• 分布在集群中的只讀對(duì)象集合(由多個(gè)Partition　構(gòu)成);
• 可以存儲(chǔ)在磁盤或內(nèi)存中(多種存儲(chǔ)級(jí)別);
• 通過并行“轉(zhuǎn)換”操作構(gòu)造;
• 失效后自動(dòng)重構(gòu);
• RDD基本操作(operator)

Transformation具體內(nèi)容

• map(func) :返回一個(gè)新的分布式數(shù)據(jù)集，由每個(gè)原元素經(jīng)過func函數(shù)轉(zhuǎn)換后組成
• filter(func) : 返回一個(gè)新的數(shù)據(jù)集，由經(jīng)過func函數(shù)后返回值為true的原元素組成

*flatMap(func) : 類似于map，但是每一個(gè)輸入元素，會(huì)被映射為0到多個(gè)輸出元素(因此，func函數(shù)的返回值是一個(gè)Seq，而不是單一元素)

• flatMap(func) : 類似于map，但是每一個(gè)輸入元素，會(huì)被映射為0到多個(gè)輸出元素(因此，func函數(shù)的返回值是一個(gè)Seq，而不是單一元素)
• sample(withReplacement, frac, seed) :

根據(jù)給定的隨機(jī)種子seed，隨機(jī)抽樣出數(shù)量為frac的數(shù)據(jù)。

• union(otherDataset) : 返回一個(gè)新的數(shù)據(jù)集，由原數(shù)據(jù)集和參數(shù)聯(lián)合而成
• groupByKey([numTasks]) :

在一個(gè)由(K,V)對(duì)組成的數(shù)據(jù)集上調(diào)用，返回一個(gè)(K，Seq[V])對(duì)的數(shù)據(jù)集。注意：默認(rèn)情況下，使用8個(gè)并行任務(wù)進(jìn)行分組，你可以傳入numTask可選參數(shù)，根據(jù)數(shù)據(jù)量設(shè)置不同數(shù)目的Task

• reduceByKey(func, [numTasks]) : 在一個(gè)(K，V)對(duì)的數(shù)據(jù)集上使用，返回一個(gè)(K，V)對(duì)的數(shù)據(jù)集，key相同的值，都被使用指定的reduce函數(shù)聚合到一起。和groupbykey類似，任務(wù)的個(gè)數(shù)是可以通過第二個(gè)可選參數(shù)來配置的。
• join(otherDataset, [numTasks]) :

在類型為(K,V)和(K,W)類型的數(shù)據(jù)集上調(diào)用，返回一個(gè)(K,(V,W))對(duì)，每個(gè)key中的所有元素都在一起的數(shù)據(jù)集

• groupWith(otherDataset, [numTasks]) : 在類型為(K,V)和(K,W)類型的數(shù)據(jù)集上調(diào)用，返回一個(gè)數(shù)據(jù)集，組成元素為(K, Seq[V], Seq[W]) Tuples。這個(gè)操作在其它框架，稱為CoGroup

cartesian(otherDataset) : 笛卡爾積。但在數(shù)據(jù)集T和U上調(diào)用時(shí)，返回一個(gè)(T，U)對(duì)的數(shù)據(jù)集，所有元素交互進(jìn)行笛卡爾積。

• flatMap(func) :

類似于map，但是每一個(gè)輸入元素，會(huì)被映射為0到多個(gè)輸出元素(因此，func函數(shù)的返回值是一個(gè)Seq，而不是單一元素)

Actions具體內(nèi)容

• reduce(func) : 通過函數(shù)func聚集數(shù)據(jù)集中的所有元素。Func函數(shù)接受2個(gè)參數(shù)，返回一個(gè)值。這個(gè)函數(shù)必須是關(guān)聯(lián)性的，確?？梢员徽_的并發(fā)執(zhí)行
• collect() : 在Driver的程序中，以數(shù)組的形式，返回?cái)?shù)據(jù)集的所有元素。這通常會(huì)在使用filter或者其它操作后，返回一個(gè)足夠小的數(shù)據(jù)子集再使用，直接將整個(gè)RDD集Collect返回，很可能會(huì)讓Driver程序OOM
• count() : 返回?cái)?shù)據(jù)集的元素個(gè)數(shù)
• take(n) : 返回一個(gè)數(shù)組，由數(shù)據(jù)集的前n個(gè)元素組成。注意，這個(gè)操作目前并非在多個(gè)節(jié)點(diǎn)上，并行執(zhí)行，而是Driver程序所在機(jī)器，單機(jī)計(jì)算所有的元素(Gateway的內(nèi)存壓力會(huì)增大，需要謹(jǐn)慎使用)
• first() : 返回?cái)?shù)據(jù)集的***個(gè)元素(類似于take(1))

saveAsTextFile(path) : 將數(shù)據(jù)集的元素，以textfile的形式，保存到本地文件系統(tǒng)，hdfs或者任何其它hadoop支持的文件系統(tǒng)。Spark將會(huì)調(diào)用每個(gè)元素的toString方法，并將它轉(zhuǎn)換為文件中的一行文本

• saveAsSequenceFile(path) : 將數(shù)據(jù)集的元素，以sequencefile的格式，保存到指定的目錄下，本地系統(tǒng)，hdfs或者任何其它hadoop支持的文件系統(tǒng)。RDD的元素必須由key-value對(duì)組成，并都實(shí)現(xiàn)了Hadoop的Writable接口，或隱式可以轉(zhuǎn)換為Writable(Spark包括了基本類型的轉(zhuǎn)換，例如Int，Double，String等等)
• foreach(func) : 在數(shù)據(jù)集的每一個(gè)元素上，運(yùn)行函數(shù)func。這通常用于更新一個(gè)累加器變量，或者和外部存儲(chǔ)系統(tǒng)做交互

算子分類

大致可以分為三大類算子：

• Value數(shù)據(jù)類型的Transformation算子，這種變換并不觸發(fā)提交作業(yè)，針對(duì)處理的數(shù)據(jù)項(xiàng)是Value型的數(shù)據(jù)。
• Key-Value數(shù)據(jù)類型的Transfromation算子，這種變換并不觸發(fā)提交作業(yè)，針對(duì)處理的數(shù)據(jù)項(xiàng)是Key-Value型的數(shù)據(jù)對(duì)。
• Action算子，這類算子會(huì)觸發(fā)SparkContext提交Job作業(yè)。

• Spark RDD cache/persist

Spark RDD cache

1.允許將RDD緩存到內(nèi)存中或磁盤上，以便于重用

2.提供了多種緩存級(jí)別，以便于用戶根據(jù)實(shí)際需求進(jìn)行調(diào)整

3.cache使用

• 之前用MapReduce實(shí)現(xiàn)過WordCount，現(xiàn)在我們用Scala實(shí)現(xiàn)下wordCount.是不是很簡潔呢?!

import org.apache.spark.{SparkConf, SparkContext} 
 
object SparkWordCount{ 
 def main(args: Array[String]) { 
 if (args.length == 0) { 
 System.err.println("Usage: SparkWordCount <inputfile> <outputfile>") 
 System.exit(1) 
 } 
 
 val conf = new SparkConf().setAppName("SparkWordCount") 
 val sc = new SparkContext(conf) 
 
 val file=sc.textFile("file:///hadoopLearning/spark-1.5.1-bin-hadoop2.4/README.md") 
 val counts=file.flatMap(line=>line.split(" ")) 
 .map(word=>(word,1)) 
 .reduceByKey(_+_) 
 counts.saveAsTextFile("file:///hadoopLearning/spark-1.5.1-bin-hadoop2.4/countReslut.txt") 
 
 } 
} 

• 關(guān)于RDD的Transformation與Action的特點(diǎn)我們介紹下;

1.接口定義方式不同：

Transformation: RDD[X]–>RDD[y]

Action:RDD[x]–>Z (Z不是一個(gè)RDD，可能是一個(gè)基本類型，數(shù)組等)

2.惰性執(zhí)行：

Transformation：只會(huì)記錄RDD轉(zhuǎn)化關(guān)系，并不會(huì)觸發(fā)計(jì)算

Action:是觸發(fā)程序執(zhí)行(分布式)的算子。

程序的執(zhí)行流程：

Spark運(yùn)行模式：

Local(本地模式)：

1.單機(jī)運(yùn)行，通常用于測(cè)試;

• local:只啟動(dòng)一個(gè)executor
• local[k]:啟動(dòng)k個(gè)executor
• local[*]:啟動(dòng)跟cpu數(shù)目相同的executor

2.standalone(獨(dú)立模式)

獨(dú)立運(yùn)行在一個(gè)集群中

3.Yarn/mesos

1.運(yùn)行在資源管理系統(tǒng)上，比如Yarn或mesos

2.Spark On Yarn存在兩種模式

yarn-client

yanr-cluster

兩種方式的區(qū)別：

Spark在企業(yè)中的應(yīng)用場(chǎng)景

• 基于日志數(shù)據(jù)的快速查詢系統(tǒng)業(yè)務(wù);

構(gòu)建于Spark之上的SparkSQL ,利用其快速以及內(nèi)存表等優(yōu)勢(shì)，承擔(dān)了日志數(shù)據(jù)的即席查詢工作。

• 典型算法的Spark實(shí)現(xiàn)

預(yù)測(cè)用戶的廣告點(diǎn)擊概率;

計(jì)算兩個(gè)好友間的共同好友數(shù);

用于ETL的SparkSQL和DAG任務(wù)。