什么是Spark

• Spark是一個(gè)用于大規(guī)模數(shù)據(jù)處理的統(tǒng)一計(jì)算引擎
• Spark一個(gè)重要的特性就是基于內(nèi)存計(jì)算，從而它的速度可以達(dá)到MapReduce的幾十倍甚至百倍

對(duì)比Hadoop

• Spark是一個(gè)綜合性質(zhì)的計(jì)算引擎，Hadoop既包含Mapreduce(計(jì)算)還包含HDFS(存儲(chǔ))和YARN(資源管理)，兩個(gè)框架定位不同，從綜合能力來說Hadoop更勝一籌
• 計(jì)算模型：Spark任務(wù)可以包含多個(gè)計(jì)算操作，輕松實(shí)現(xiàn)復(fù)雜迭代計(jì)算，Hadoop中的mapreduce任務(wù)只包含Map和Reduce階段，不夠靈活
• 處理速度：Spark任務(wù)的數(shù)據(jù)是存放在內(nèi)存里面的，而Hadoop中的MapReduce任務(wù)是基于磁盤的

在實(shí)際工作中Hadoop會(huì)作為一個(gè)提供分布式存儲(chǔ)和分布式資源管理的一個(gè)角色存在，Spark會(huì)依賴于Hadoop去做計(jì)算。

Spark應(yīng)用場(chǎng)景

• 低延時(shí)的海量數(shù)據(jù)計(jì)算需求
• 低延時(shí)的SQL交互查詢需求
• 準(zhǔn)實(shí)時(shí)計(jì)算需求

Spark數(shù)據(jù)處理流程

什么是RDD

• 通常通過Hadoop上的文件，即HDFS文件進(jìn)行創(chuàng)建，也可以通過程序中的集合來創(chuàng)建
• 是Spark提供的核心抽象，全稱為Resillient Distributed Dataset，即彈性分布式數(shù)據(jù)集
  • 彈性：RDD數(shù)據(jù)在默認(rèn)的情況下存放內(nèi)存中，但是在內(nèi)存資源不足時(shí)，Spark也會(huì)自動(dòng)將RDD數(shù)據(jù)寫入磁盤
  • RDD在抽象上來說是一種元素?cái)?shù)據(jù)的集合，它是被分區(qū)的，每個(gè)分區(qū)分布在集群中的不同節(jié)點(diǎn)上，從而RDD中的數(shù)據(jù)可以被并行操作
  • 容錯(cuò)性：最重要的特性就是提供了容錯(cuò)性，可以自動(dòng)從節(jié)點(diǎn)失敗中恢復(fù)過來。比如某個(gè)節(jié)點(diǎn)的數(shù)據(jù)由于故障導(dǎo)致分區(qū)的數(shù)據(jù)丟了，RDD會(huì)自動(dòng)通過數(shù)據(jù)來源重新計(jì)算數(shù)據(jù)

Spark架構(gòu)相關(guān)進(jìn)程

• Driver：我們編寫的Spark程序由Driver進(jìn)程負(fù)責(zé)執(zhí)行
• Master：集群的主節(jié)點(diǎn)中啟動(dòng)的進(jìn)程
• Worker：集群的從節(jié)點(diǎn)中啟動(dòng)的進(jìn)程
• Executor：由Worker負(fù)責(zé)啟動(dòng)的進(jìn)程，執(zhí)行數(shù)據(jù)處理和數(shù)據(jù)計(jì)算
• Task：由Executor負(fù)責(zé)啟動(dòng)的線程，是真正干活的

入門案例：統(tǒng)計(jì)單詞數(shù)量

# scala 代碼
object WordCountScala {def main(args: Array[String]): Unit = {val conf = new SparkConf();conf.setAppName("wordCount").setMaster("local")val context = new SparkContext(conf);val linesRDD = context.textFile("D:\\hadoop\\logs\\hello.txt");var wordsRDD = linesRDD.flatMap(line => line.split(" "))val pairRDD = wordsRDD.map(word => (word, 1))val wordCountRDD = pairRDD.reduceByKey(_ + _)wordCountRDD.foreach(wordCount => println(wordCount._1 + "---" + wordCount._2))context.stop()}
}

public class WordCountJava {public static void main(String[] args) {SparkConf sparkConf = new SparkConf();sparkConf.setAppName("worldCount").setMaster("local");JavaSparkContext javaSparkContext = new JavaSparkContext();JavaRDD<String> stringJavaRDD = javaSparkContext.textFile("D:\\hadoop\\logs\\hello.txt");// 數(shù)據(jù)切割，把一行數(shù)據(jù)拆分為一個(gè)個(gè)的單詞// 第一個(gè)是輸入數(shù)據(jù)類型，第二個(gè)是輸出數(shù)據(jù)類型JavaRDD<String> wordRDD = stringJavaRDD.flatMap(new FlatMapFunction<String, String>() {@Overridepublic Iterator<String> call(String line) throws Exception {return Arrays.asList(line.split(" ")).iterator();}});// 迭代word，裝換成（word，1）這種形式// 第一個(gè)是輸入?yún)?shù)，第二個(gè)是輸出第一個(gè)參數(shù)類型，第三個(gè)是輸出第二個(gè)參數(shù)類型JavaPairRDD<String, Integer> pairRDD = wordRDD.mapToPair(new PairFunction<String, String, Integer>() {@Overridepublic Tuple2<String, Integer> call(String word) throws Exception {return new Tuple2<>(word, 1);}});// 根據(jù)key進(jìn)行分組聚合JavaPairRDD<String, Integer> wordCountRDD = pairRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {@Overridepublic Integer call(Integer v1, Integer v2) throws Exception {return v1 + v2;}});// 輸出控制臺(tái)wordCountRDD.foreach(new VoidFunction<Tuple2<String, Integer>>() {@Overridepublic void call(Tuple2<String, Integer> tuple2) throws Exception {System.out.println(tuple2._1 + "=:=" + tuple2._2);}});javaSparkContext.stop();}}

Spark開啟historyServer

[root@hadoop04 conf]# vim spark-env.sh 
export SPARK_HISTORY_OPTS="-Dspark.history.ui.port=18080 -Dspark.history.fs.logDirectory=hdfs://hadoop01:9000/tmp/logs/root/logs"[root@hadoop04 conf]# vim spark-defaults.conf 
spark.eventLof.enable=true
spark.eventLog.compress=true
spark.eventLog.dir=hdfs://hadoop01:9000/tmp/logs/root/logs
spark.history.fs.logDirectory=hdfs://hadoop01:9000/tmp/logs/root/logs# 啟動(dòng)
[root@hadoop04 conf]# sbin/start-history-server.sh # 訪問
http://hadoop04:18080/