Spark追妻系列(Spark初了解)

Posted 2022-02-13 数仓白菜白

感觉每天又有了希望，又有动力，感觉学习尽头十足

Spark是什么

        Spark是一个用来实现快速而通用的集群计算的平台。

        在之前，学习了MR，学习了hadoop，用mapreduce来对数据进行处理，但是hadoop是用批处理的，而且还有延迟，况且，出来了Hive，Hive将sql转化为mr算子。可以不用去写mr代码就可以对数据进行分析。

        Spark是集群计算，是在hadoop集群中进行的，Spark的一个主要特点就是能够在内存中进行计算，在内存中计算比在磁盘上运算要快很多，同时现在学习Spark的代价也很小了，大数据相关的软件都在慢慢实现SQL化，SPark,Flink 都支持SQL进行处理。

 

        Spark是一个软件的大集合，从上面的图中可以看到，在以Spark Core为基础和重点的基础上，增加了SQL,实时计算，机器学习，图计算。

Spark Core是什么

        spark core实现了spark的基本功能，包括任务调度，内存管理，错误恢复和存储系统交互等模块。

Spark SQL是什么

        SPark SQL是用Spark来操作结构化数据的程序包，通过Spark SQL可以使用Hive SQL来查询数据，同时Hive的默认引擎是mr，也可以换成Spark。

        同时呢，只介绍这两个，因为这两个将会在后面的博客出现几个月甚至一年。等实力够了再去学习其它的。

初识Spark

        在博客中讲解的Spark都是基于windows本地的，不是在命令行上面sbt。

        关于spark的本地安装，可以去b站看看尚硅谷的Spark视频。

命令行和本地运行的差别

1. 在本地运行时，可以在idea上写代码，但是需要自行初始化SparkContext
2. 命令行提交之后就会出结果，本地运行直接在Idea中出结果

初始化SparkContext

        想要在本地编写Spark Core代码，需要本地的Hadoop 和 Spark环境。

        导入Spark的包，创建SparkContext。

        先创建一个SparkConf对象来配置你的应用，然后基于这个SparkConf来创建SparkContext对象。

        val cf =new SparkConf().setMaster("local").setAppName("") val sc = new SparkContext(cf)

为什么要这样做呢？

1. 集群url，告诉Spark如何连接集群，因为是本地集群，所以SetMaster的时候是local
2. 为此次程序命名

        初始化Spark Context后，就可以创建RDD来进行代码了。

        关闭的时候 通过 sc.stop()就可以推出应用

        通过RDD来实现W C

首先看一下数据源

        在MapReduce中，word count就是第一个入门案例，在Spark里面，WordCount也是第一个入门案例。

代码实现

val wordCount = new SparkConf().setMaster("local").setAppName("WordCount") 
val sparkContext = new SparkContext(wordCount) 
val value = sparkContext.textFile("date/1.txt") 
val value1 = value.flatMap(_.split(" ")) 
val value2 = value1.groupBy(value1 => value1) 
val value3 = value2.map  case (word, list) => (word, list.size)  value3.collect().foreach(println(_)) s
parkContext.stop()

        上面就是用Scala代码来实现的WordCount。

首先进行层层分析

1. 创建SparkConf对象，对集群进行配置，创建SparkContext对象

val wordCount = new SparkConf().setMaster("local").setAppName("WordCount") 
val sparkContext = new SparkContext(wordCount)

2.通过textFile来获取数据源

val value = sparkContext.textFile("date/1.txt")

3.逻辑代码

//flatMap函数，将数据根据空格分开
val value1 = value.flatMap(_.split(" ")) 
//groupBy分组，对每个单词进行分组 
val value2 = value1.groupBy(value1 => value1) 
//map + match进行匹配 value2的类型是(String,Iterable) 
val value3 = value2.map  case (word, list) => (word, list.size)  
//如果匹配到老了，就转换类型 
(String,Iterable) => (String,Int)

RDD是什么

        RDD是弹性 分布式 数据集。

        在Spark中，对数据的所有操作就是创建RDD,转化已有的RDD,调用RDD进行求值

如何创建RDD

两种方法创建RDD

1. 读取一个外部数据集
2. 在驱动程序里分发驱动程序中的对象集合

        上面的例子中，通过读取外部文件来创建RDD

        创建RDD过后，RDD支持两种操作，转换和动作。

        转错操作就是各种算子，由一个RDD变成另外一个新的RDD。行动算子会对RDD计算出结果，并将结果返回控制台。

        上面的例子中，flatMap,map,groupBy都是转换算子，foreach,collect等都是动作算子

        虽然一个程序会有多个算子，但是Spark是惰性运算RDD的，只有在第一次用到的时候，这个RDD算子才会运作，还有一点先说一下，RDD算子并不存储数据，只是一个工具，数据都是在各个节点根据RDD进行运算的。

        默认情况下，Spark中的RDD会在每次行动算子运算的时候会重新运算，如果想多个行动操作用一个RDD，那么可以RDD.persist（）让Spark对RDD进行缓存。也就是持久化。

总结：

        先写一篇简短的文章，明天会输出篇幅比较大的文章。