一、Spark概述
1、Spark簡介
Spark是專為大規模數據處理而設計的，基于內存快速通用,可擴展的集群計算引擎，實現了高效的DAG執行引擎,可以通過基于內存來高效處理數據流，運算速度相比于MapReduce得到了顯著的提高。
2、運行結構
Driver
運行Spark的Applicaion中main()函數，會創建SparkContext，SparkContext負責和Cluster-Manager進行通信，并負責申請資源、任務分配和監控等。
ClusterManager
負責申請和管理在WorkerNode上運行應用所需的資源，可以高效地在一個計算節點到數千個計算節點之間伸縮計算，目前包括Spark原生的ClusterManager、ApacheMesos和HadoopYARN。
Executor
Application運行在WorkerNode上的一個進程，作為工作節點負責運行Task任務，并且負責將數據存在內存或者磁盤上，每個 Application都有各自獨立的一批Executor，任務間相互獨立。
二、環境部署
1、Scala環境
安裝包管理
[root@hop01 opt]# tar -zxvf scala-2.12.2.tgz
[root@hop01 opt]# mv scala-2.12.2 scala2.12
配置變量
[root@hop01 opt]# vim /etc/profile
export SCALA_HOME=/opt/scala2.12
export PATH=$PATH:$SCALA_HOME/bin
[root@hop01 opt]# source /etc/profile
版本查看
[root@hop01 opt]# scala -version
Scala環境需要部署在Spark運行的相關服務節點上。
2、Spark基礎環境
安裝包管理
[root@hop01 opt]# tar -zxvf spark-2.1.1-bin-hadoop2.7.tgz
[root@hop01 opt]# mv spark-2.1.1-bin-hadoop2.7 spark2.1
配置變量
[root@hop01 opt]# vim /etc/profile
export SPARK_HOME=/opt/spark2.1
export PATH=$PATH:$SPARK_HOME/bin
[root@hop01 opt]# source /etc/profile
版本查看
[root@hop01 opt]# spark-shell
3、Spark集群配置
服務節點
[root@hop01 opt]# cd /opt/spark2.1/conf/
[root@hop01 conf]# cp slaves.template slaves
[root@hop01 conf]# vim slaves
hop01
hop02
hop03
環境配置
[root@hop01 conf]# cp spark-env.sh.template spark-env.sh
[root@hop01 conf]# vim spark-env.sh
export JAVA_HOME=/opt/jdk1.8
export SCALA_HOME=/opt/scala2.12
export SPARK_MASTER_IP=hop01
export SPARK_LOCAL_IP=安裝節點IP
export SPARK_WORKER_MEMORY=1g
export HADOOP_CONF_DIR=/opt/hadoop2.7/etc/hadoop
注意SPARK_LOCAL_IP的配置。
4、Spark啟動
依賴Hadoop相關環境，所以要先啟動。
啟動：/opt/spark2.1/sbin/start-all.sh
停止：/opt/spark2.1/sbin/stop-all.sh
這里在主節點會啟動兩個進程：Master和Worker，其他節點只啟動一個Worker進程。
5、訪問Spark集群
默認端口是：8080。
http://hop01:8080/
運行基礎案例：
[root@hop01 spark2.1]# cd /opt/spark2.1/
[root@hop01 spark2.1]# bin/spark-submit --class org.apache.spark.examples.SparkPi --master local examples/jars/spark-examples_2.11-2.1.1.jar
運行結果：Pi is roughly 3.1455357276786384
三、開發案例
1、核心依賴
依賴Spark2.1.1版本：
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.11</artifactId>
<version>2.1.1</version>
</dependency>
引入Scala編譯插件：
<plugin>
<groupId>net.alchim31.maven</groupId>
<artifactId>scala-maven-plugin</artifactId>
<version>3.2.2</version>
<executions>
<execution>
<goals>
<goal>compile</goal>
<goal>testCompile</goal>
</goals>
</execution>
</executions>
</plugin>
2、案例代碼開發
讀取指定位置的文件，并輸出文件內容單詞統計結果。
@RestController
public class wordWeb implements Serializable {
@GetMapping("/word/web")
public String getWeb (){
// 1、創建Spark的配置對象
SparkConf sparkConf = new SparkConf().setAppName("LocalCount")
.setMaster("local[*]");
// 2、創建SparkContext對象
JavaSparkContext sc = new JavaSparkContext(sparkConf);
sc.setLogLevel("WARN");
// 3、讀取測試文件
JavaRDD lineRdd = sc.textFile("/var/spark/test/word.txt");
// 4、行內容進行切分
JavaRDD wordsRdd = lineRdd.flatMap(new FlatMapFunction() {
@Override
public Iterator call(Object obj) throws Exception {
String value = String.valueOf(obj);
String[] words = value.split(",");
return Arrays.asList(words).iterator();
}
});
// 5、切分的單詞進行標注
JavaPairRDD wordAndOneRdd = wordsRdd.mapToPair(new PairFunction() {
@Override
public Tuple2 call(Object obj) throws Exception {
//將單詞進行標記：
return new Tuple2(String.valueOf(obj), 1);
}
});
// 6、統計單詞出現次數
JavaPairRDD wordAndCountRdd = wordAndOneRdd.reduceByKey(new Function2() {
@Override
public Object call(Object obj1, Object obj2) throws Exception {
return Integer.parseInt(obj1.toString()) + Integer.parseInt(obj2.toString());
}
});
// 7、排序
JavaPairRDD sortedRdd = wordAndCountRdd.sortByKey();
List<Tuple2> finalResult = sortedRdd.collect();
// 8、結果打印
for (Tuple2 tuple2 : finalResult) {
System.out.println(tuple2._1 + " ===> " + tuple2._2);
}
// 9、保存統計結果
sortedRdd.saveAsTextFile("/var/spark/output");
sc.stop();
return "success" ;
}
}
打包執行結果：
查看文件輸出：
[root@hop01 output]# vim /var/spark/output/part-00000
