Spark框架学习
一：Spark概述
官网：http://spark.apache.org/
Apache Spark™是用于大规模数据处理的统一分析引擎。
为大数据处理而设计的快速通用的计算引擎。
Spark加州大学伯克利分校AMP实验室。不同于mapreduce的是一个Spark任务的中间
结果保存到内存中。空间换时间。
Spark启用的是内存分布式数据集。
用scala语言实现，与spark紧密继承。用scala可以轻松的处理分布式数据集。
Spark并不是为了替代hadoop，而为了补充hadoop。
Spark并没有存储。可以集成HDFS。
二：Spark特点
1）速度快
与mr对比，磁盘运行的话10倍以上。
内存运行的话，100倍以上。
2）便于使用
支持java/scala/python/R
3)通用
不仅支持批处理（SparkSQL）
而且支持流处理（SparkStreaming）
4)兼容
兼容其它组件
Spark实现了Standalone作为内置的资源管理和调度框架。hdfs/yarn。
三：Spark安装部署
集群规划：
主节点：Master bigdata112
从节点：Worker bigdata113 bigdata114
1）准备工作
关闭防火墙
设置主机名/etc/hostname
映射文件/etc/hosts
免密登录scp
ssh-keygen
安装jdk(scala依赖jvm)
2)安装Spark集群
-》上传
-》解压
-》修改配置文件
spark-env.sh（直接在该文件的最后加入如下配置）
jdk=
master_host=
master_port=
slaves 加入从节点
Bigdata112
Bigdata113
这样集群就搭建好了！！！比hadoop的搭建方便多了
-》启动集群
sbin/start-all.sh
四：启动sparkshell
本地模式：bin/spark-shell
集群启动：bin/spark-shell --master spark://spark-01:7077
--total-executor-cores 2
--executor-memory 500mb
最后2个参数是可选的！！可以加可不加，不加的话系统会根据你虚拟机配置的核心数来决定！
Spark Shell版的WordCount：
五：Spark集群角色：
|
Yarn |
Spark |
作用 |
|
ResourceManager |
Master |
管理子节点 |
|
NodeManager |
Worker |
管理当前节点 |
|
YarnChild |
Executor |
处理计算任务 |
|
Client+ApplicationMaster |
SparkSubmit |
提交计算任务 |
案例一：Spark版的WordCount程序
Step1：创建一个Maven工程。
编写Pom文件：
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.dawn.spark</groupId> <artifactId>SparkWC</artifactId> <version>1.0-SNAPSHOT</version> <properties> <maven.compiler.source>1.8</maven.compiler.source> <maven.compiler.target>1.8</maven.compiler.target> <scala.version>2.11.8</scala.version> <spark.version>2.2.0</spark.version> <hadoop.version>2.8.4</hadoop.version> <encoding>UTF-8</encoding> </properties> <dependencies> <!-- scala的依赖导入 --> <dependency> <groupId>org.scala-lang</groupId> <artifactId>scala-library</artifactId> <version>${scala.version}</version> </dependency> <!-- spark的依赖导入 --> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-core_2.11</artifactId> <version>${spark.version}</version> </dependency> <!-- hadoop-client API的导入 --> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>${hadoop.version}</version> </dependency> </dependencies> <build> <pluginManagement> <plugins> <!-- scala的编译插件 --> <plugin> <groupId>net.alchim31.maven</groupId> <artifactId>scala-maven-plugin</artifactId> <version>3.2.2</version> </plugin> <!-- ava的编译插件 --> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <version>3.5.1</version> </plugin> </plugins> </pluginManagement> <plugins> <plugin> <groupId>net.alchim31.maven</groupId> <artifactId>scala-maven-plugin</artifactId> <executions> <execution> <id>scala-compile-first</id> <phase>process-resources</phase> <goals> <goal>add-source</goal> <goal>compile</goal> </goals> </execution> <execution> <id>scala-test-compile</id> <phase>process-test-resources</phase> <goals> <goal>testCompile</goal> </goals> </execution> </executions> </plugin> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-compiler-plugin</artifactId> <executions> <execution> <phase>compile</phase> <goals> <goal>compile</goal> </goals> </execution> </executions> </plugin> <!-- 打jar包插件 --> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <version>2.4.3</version> <executions> <execution> <phase>package</phase> <goals> <goal>shade</goal> </goals> <configuration> <filters> <filter> <artifact>*:*</artifact> <excludes> <exclude>META-INF/*.SF</exclude> <exclude>META-INF/*.DSA</exclude> <exclude>META-INF/*.RSA</exclude> </excludes> </filter> </filters> </configuration> </execution> </executions> </plugin> </plugins> </build> </project>
Step2：编写WordCount代码
import org.apache.spark.{SparkConf, SparkContext}
/**
* @author Dawn
* 2019年6月18日15:39:39
* @version 1.0
* spark-WordCount本地模式测试
*/
object ScalaWordCount {
def main(args: Array[String]): Unit = {
//2.设置参数 setAppName设计程序名 setMaster本地测试设置线程数 *多个
val conf:SparkConf=new SparkConf().setAppName("ScalaWordCount").setMaster("local[*]")
//1.创建spark执行程序的入口
val sc:SparkContext=new SparkContext(conf)
//3.加载数据 并且处理
sc.textFile("f:/temp/data.txt").flatMap(_.split(" ")).map((_,1))
.reduceByKey(_+_)
.sortBy(_._2,false)
.foreach(println)
//保存文件
// .saveAsTextFile("f:/temp/scalaWC/")
//4.关闭资源
sc.stop()
}
}
注意：
运行结果如下：
