Spark

Spark 是什么?

Apache Spark?是用于大規(guī)模數(shù)據(jù)處理的快速和通用引擎.

速度:在內(nèi)存中,運(yùn)行程序比Hadoop MapReduce快100倍，在磁盤上則要快10倍.

Apache Spark具有支持非循環(huán)數(shù)據(jù)流和內(nèi)存計(jì)算的高級DAG執(zhí)行引擎.

易用:可以使用Java，Scala，Python，R快速編寫程序.

Spark提供80+高級操作方法，可以輕松構(gòu)建并行應(yīng)用程序.

Spark提供了一堆庫，包括SQL和DataFrame,MLlib，GraphX和Spark Streaming。您可以在相同的應(yīng)用程序中無縫地組合這些庫. Spark在Hadoop，Mesos，獨(dú)立或云端運(yùn)行。它可以訪問各種數(shù)據(jù)源，包括HDFS，Cassandra，HBase和S3

一,RDD 彈性分布式數(shù)據(jù)集

定義, TA 容錯(cuò)的,并行的數(shù)據(jù)結(jié)構(gòu),存儲(chǔ)到磁盤和內(nèi)存，控制數(shù)據(jù)分區(qū)。本質(zhì)上是一個(gè)只讀的分區(qū)記錄集合,RDD包含多個(gè)分區(qū)，每個(gè)分區(qū)是一個(gè)dataset片段.

依賴, RDD可以相互依賴。如果RDD的每個(gè)分區(qū)最多只能被一個(gè)Child RDD的一個(gè)分區(qū)使用，窄依賴；若多個(gè)Child RDD分區(qū)都可以依賴,寬依賴.

首先，窄依賴被劃分到同一個(gè)stage,支持在同一個(gè)cluster node上以管道形式執(zhí)行多條命令，eg,先map,緊接著filter.相反,寬依賴由于依賴的上游節(jié)點(diǎn)不止一個(gè),往往跨界點(diǎn)傳輸數(shù)據(jù).

其次從容災(zāi)角度講,窄依賴的只需要執(zhí)行父RDD的丟失分區(qū)的計(jì)算即可恢復(fù).而寬依賴需要考慮恢復(fù)所有父RDD的丟失分區(qū).

?本質(zhì), RDD是Spark中的抽象數(shù)據(jù)結(jié)構(gòu)類型,從編程的角度來看，RDD可以簡單看成是一個(gè)數(shù)組。和普通數(shù)組的區(qū)別是，RDD中的數(shù)據(jù)是分區(qū)存儲(chǔ)的，這樣不同分區(qū)的數(shù)據(jù)就可以分布在不同的機(jī)器上，同時(shí)可以被并行處理。因此，Spark應(yīng)用程序所做的無非是把需要處理的數(shù)據(jù)轉(zhuǎn)換為RDD，然后對RDD進(jìn)行一系列的變換和操作從而得到結(jié)果。本質(zhì)是一個(gè)抽象類,如下:

abstract class RDD[T: ClassTag](@transient private var _sc: SparkContext,@transient private var deps: Seq[Dependency[_]]) extends Serializable with Logging {} typefunctionuse

transformation	map()	函數(shù)應(yīng)用于RDD每一個(gè)元素，返回值是新的RDD
transformation	flatMap()	函數(shù)應(yīng)用于RDD每一個(gè)元素，將元素?cái)?shù)據(jù)進(jìn)行拆分變成迭代器返回值是新的RDD
transformation	filter()	過濾,返回值是新的RDD
transformation	distinct()	去重，返回值是新的RDD
transformation	union()	并集，返回值是新的RDD
transformation	intersection()	交集，返回值是新的RDD
transformation	subtract()	原RDD里和參數(shù)RDD里相同的元素去掉
transformation	cartesian()	函數(shù)應(yīng)用于RDD每一個(gè)元素，返回值是新的RDD

typefunctionuse

action	collect()	返回RDD所有元素
action	count()	RDD里元素個(gè)數(shù)
action	countByValue()	各元素在RDD中出現(xiàn)次數(shù)
action	reduce()	并行整合所有RDD數(shù)據(jù)，例如求和操作
action	fold(0)(func)	和reduce功能一樣，不過fold帶有初始值
action	aggregate(0)(seqOp,combop)	和reduce功能一樣，但是返回的RDD數(shù)據(jù)類型和原RDD不一樣
action	foreach(func)	對RDD每個(gè)元素都是使用特定函數(shù)

DAG 有向無環(huán)圖

容錯(cuò)處理

傳統(tǒng)關(guān)系型數(shù)據(jù)庫:采用日志記錄容災(zāi),數(shù)據(jù)恢復(fù)都依賴于重新執(zhí)行日志中的SQL;

Hadoop:通過把數(shù)據(jù)備份到其他機(jī)器來容災(zāi);

RDD:本身是一個(gè)不可變的數(shù)據(jù)集,當(dāng)某個(gè)worker節(jié)點(diǎn)上的任務(wù)失敗時(shí),可以利用DAG重新調(diào)度計(jì)算這個(gè)失敗的任務(wù),由于不用復(fù)制數(shù)據(jù),從而大大降低了網(wǎng)絡(luò)通信.在流式計(jì)算場景中,Spark需要記錄日志和檢查點(diǎn),以便利用checkpoint和日志對數(shù)據(jù)進(jìn)行恢復(fù);

二,Discretized Streams (DStreams)

DStream是一系列連續(xù)的RDD,是Spark Streaming提供的基本抽象如下圖所示:?

對DStream應(yīng)用的任何操作都將轉(zhuǎn)換為底層RDD上的操作

三,Initializing StreamingContext

要初始化Spark Streaming程序，必須創(chuàng)建一個(gè)StreamingContext對象，它是所有Spark Streaming功能的主要入口.

SparkConf conf = new SparkConf().setAppName(appName).setMaster(master); JavaStreamingContext ssc = new JavaStreamingContext(conf, new Duration(1000));

?

四,Input DStreams and Receivers

Spark Streaming提供兩類內(nèi)置流式傳輸源:

基本數(shù)據(jù)源:StreamingContext API中直接提供的源.比如：文件系統(tǒng)和套接字連接.(file 和 socket) 高級源:Kafka，Flume，Kinesis等資源可以通過額外的實(shí)用類來獲得.

Spark Streaming 提供兩種接收器:

可靠的接收器 - 當(dāng)數(shù)據(jù)已被接收并且通過復(fù)制存儲(chǔ)在Spark中時(shí)，可靠的接收器正確地向可靠的源發(fā)送確認(rèn)。 不可靠的接收器 - 不可靠的接收器不向源發(fā)送確認(rèn)。這可以用于不支持確認(rèn)的源，或者甚至當(dāng)不需要或需要進(jìn)入確認(rèn)的復(fù)雜性時(shí)，用于可靠的源。

五,Transformations on DStreams

TransformationMeaning

map(func)	通過func傳遞源DStream的每個(gè)元素,返回新的DStream
flatMap(func)	與map類似,但每個(gè)輸入項(xiàng)可以映射到0個(gè)或更多的輸出項(xiàng)
filter(func)	過濾
repartition(numPartitions)	通過修改分區(qū)來更改DStream中的并發(fā)數(shù)
union(otherStream)	求兩個(gè)DStream的并集
count()	計(jì)算源DStream的每個(gè)RDD中的元素?cái)?shù)量，返回RDD的新DStream
reduce(func)	使用函數(shù)func聚合源DStream的每個(gè)RDD中的元素來返回單個(gè)元素RDD的新DStream
countByValue()	根據(jù)value計(jì)算key.
reduceByKey(func, [numTasks])	根據(jù)Key進(jìn)行特定的計(jì)算
join(otherStream, [numTasks])	當(dāng)（K，V）和（K，W）對的兩個(gè)DStream被調(diào)用時(shí)，返回一個(gè)新的（K，（V，W））對的DStream與每個(gè)鍵的所有元素對
transform(func)	通過對源DStream的每個(gè)RDD應(yīng)用RDD到RDD函數(shù)來返回新的DStream。這可以用于對DStream進(jìn)行任意RDD操作
updateStateByKey(func)	返回一個(gè)新的“狀態(tài)”DStream，其中每個(gè)key的狀態(tài)通過在key的先前狀態(tài)應(yīng)用給定的功能和key的新值來更新。這可以用于維護(hù)每個(gè)key的任意狀態(tài)數(shù)據(jù)

六,Output Operations on DStreams

Output OperationMeaning

print()	打印10個(gè)元素,用于調(diào)試
saveAsTextFiles(prefix, [suffix])	將此DStream的內(nèi)容另存為文本文件。每個(gè)批處理間隔的文件名是根據(jù)前綴和后綴“prefix-TIME_IN_MS [.suffix]”生成的
saveAsObjectFiles(prefix, [suffix])	將此DStream的內(nèi)容保存為序列化Java對象的SequenceFiles。每個(gè)批處理間隔的文件名是根據(jù)前綴和后綴“prefix-TIME_IN_MS [.suffix]”生成的。
saveAsHadoopFiles(prefix, [suffix])	將此DStream的內(nèi)容另存為Hadoop文件。每個(gè)批處理間隔的文件名是根據(jù)前綴和后綴“prefix-TIME_IN_MS [.suffix]”生成的。
foreachRDD(func)	對從流中生成的每個(gè)RDD應(yīng)用函數(shù)func的最通用的輸出運(yùn)算符。此功能應(yīng)將每個(gè)RDD中的數(shù)據(jù)推送到外部系統(tǒng)，例如將RDD保存到文件，或?qū)⑵渫ㄟ^網(wǎng)絡(luò)寫入數(shù)據(jù)庫

轉(zhuǎn)載：https://www.2cto.com/net/201711/695363.html

總結(jié)

以上是生活随笔為你收集整理的Spark介绍的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。