什么是Spark
- Spark是一个用于大规模数据处理的统一计算引擎
- Spark一个重要的特性就是基于内存计算，从而它的速度可以达到MapReduce的几十倍甚至百倍
对比Hadoop
- Spark是一个综合性质的计算引擎，Hadoop既包含Mapreduce(计算)还包含HDFS(存储)和YARN(资源管理)，两个框架定位不同，从综合能力来说Hadoop更胜一筹
- 计算模型：Spark任务可以包含多个计算操作，轻松实现复杂迭代计算，Hadoop中的mapreduce任务只包含Map和Reduce阶段，不够灵活
- 处理速度：Spark任务的数据是存放在内存里面的，而Hadoop中的MapReduce任务是基于磁盘的
在实际工作中Hadoop会作为一个提供分布式存储和分布式资源管理的一个角色存在，Spark会依赖于Hadoop去做计算。
Spark应用场景
- 低延时的海量数据计算需求
- 低延时的SQL交互查询需求
- 准实时计算需求
Spark数据处理流程
什么是RDD
- 通常通过Hadoop上的文件，即HDFS文件进行创建，也可以通过程序中的集合来创建
- 是Spark提供的核心抽象，全称为Resillient Distributed Dataset，即弹性分布式数据集
- 弹性：RDD数据在默认的情况下存放内存中，但是在内存资源不足时，Spark也会自动将RDD数据写入磁盘
- RDD在抽象上来说是一种元素数据的集合，它是被分区的，每个分区分布在集群中的不同节点上，从而RDD中的数据可以被并行操作
- 容错性：最重要的特性就是提供了容错性，可以自动从节点失败中恢复过来。比如某个节点的数据由于故障导致分区的数据丢了，RDD会自动通过数据来源重新计算数据
Spark架构相关进程
- Driver：我们编写的Spark程序由Driver进程负责执行
- Master：集群的主节点中启动的进程
- Worker：集群的从节点中启动的进程
- Executor：由Worker负责启动的进程，执行数据处理和数据计算
- Task：由Executor负责启动的线程，是真正干活的
入门案例：统计单词数量
# 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");
// 数据切割，把一行数据拆分为一个个的单词
// 第一个是输入数据类型，第二个是输出数据类型
JavaRDD<String> wordRDD = stringJavaRDD.flatMap(new FlatMapFunction<String, String>() {
@Override
public Iterator<String> call(String line) throws Exception {
return Arrays.asList(line.split(" ")).iterator();
}
});
// 迭代word，装换成（word，1）这种形式
// 第一个是输入参数，第二个是输出第一个参数类型，第三个是输出第二个参数类型
JavaPairRDD<String, Integer> pairRDD = wordRDD.mapToPair(new PairFunction<String, String, Integer>() {
@Override
public Tuple2<String, Integer> call(String word) throws Exception {
return new Tuple2<>(word, 1);
}
});
// 根据key进行分组聚合
JavaPairRDD<String, Integer> wordCountRDD = pairRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer v1, Integer v2) throws Exception {
return v1 + v2;
}
});
// 输出控制台
wordCountRDD.foreach(new VoidFunction<Tuple2<String, Integer>>() {
@Override
public 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
# 启动
[root@hadoop04 conf]# sbin/start-history-server.sh
# 访问
http://hadoop04:18080/
