今天主要学习了Spark环境的搭建以及一些RDD算子的学习
Spark环境搭建
比起hadoop的环境搭建，要搭建起一个Spark的学习环境不要太简单。我们使用Idea创建一个Maven项目，导入scala的支持，然后导入如下依赖：
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.12</artifactId>
<version>3.0.1</version>
</dependency>
</dependencies>
编写一
/**
* @Description:
* @author: LiuGe
* @date: 2021/1/25
*/
object Spark03_WordCount {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount")
val sc = new SparkContext(sparkConf)
val lines: RDD[String] = sc.textFile("datas")
val words: RDD[String] = lines.flatMap(_.split(" "))
val wordToOne = words.map {
word => (word, 1)
}
// Spark框架提供了更多的功能，可以把分组和聚合用一个方法实现
// reduceByKey：相同的key的数据可以对value进行reduce聚合
val wordToCount = wordToOne.reduceByKey(_ + _)
// 5.将转换结果采集到控制台打印出来
val array = wordToCount.collect()
array.foreach(println)
// 关闭连接
sc.stop()
}
}
如果控制台能打印出结果，说明我们的本地测试环境就搭建完成了。但其实真正的spark运行环境不是这样的，我这里就不记录了。
常见RDD算子学习
什么是RDD算子
首先要明确的概念是RDD：
RDD（Resilient Distributed Dataset）叫做弹性分布式数据集，是 Spark 中最基本的数据处理模型。代码中是一个抽象类，它代表一个弹性的、不可变、可分区、里面的元素可并行计算的集合。
简单说，就是Spark中最小的单位。而RDD算子指的其实就是RDD的方法（函数）
map && flatmap && groupby && filter
基本和scala中的功能一致，不再细说
mapPartitions
首先先来看代码：
/**
* @Description:
* @author: LiuGe
* @date: 2021/1/25
*/
object Spark02_RDD_Operator_Transform {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
val sc = new SparkContext(sparkConf)
// 算法 — mapPartitions
val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)
// 可以以分区为单位进行数据转换操作
// 但会将整个分区的数组加载到内存中进行引用
// 处理完的数据是不会释放的，存在对象的引用
// 内存较小，数据量较大的场合下容易出现内存溢出
val mapRDD: RDD[Int] = rdd.mapPartitions(iter => {
println(">>>>>>>>>>")
iter.map(_ * 2)
})
mapRDD.collect()
sc.stop()
}
}
可以看到，mapPartitions相比较于map，其实就是把对每个元素的处理变成了对整个分区的处理。这样的效率会更高，但由于它处理完的数据不会自动释放，在内存不够，数据量大的时候会出现内存溢出，要注意map和mapPartitions的选择。
mapPartitionsWithIndex
/**
* @Description:
* @author: LiuGe
* @date: 2021/1/25
*/
object Spark03_RDD_Operator_Transform {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
val sc = new SparkContext(sparkConf)
// 算法 — mapPartitions
val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4), 2)
// [1,2], [3,4]
val mpiRdd: RDD[Int] = rdd.mapPartitionsWithIndex((index, iter) => {
if (index == 1) iter else Nil.iterator
})
mpiRdd.collect().foreach(println)
sc.stop()
}
}
类似的，mapPartitionsWithIndex就是加上了每个分区的编号，让我们可以实现一些特殊的需求。
glom
/**
* @Description:
* @author: LiuGe
* @date: 2021/1/25
*/
object Spark05_RDD_Operator_Transform {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
val sc = new SparkContext(sparkConf)
// 算法 — glom
val rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4),2)
// List => Int
// Int => Array
val glomRDD: RDD[Array[Int]] = rdd.glom()
glomRDD.collect().foreach(data => println(data.mkString(",")))
sc.stop()
}
}
glom，指的是将同一个分组的数据再转换成Array。
sample
/**
* @Description:
* @author: LiuGe
* @date: 2021/1/25
*/
object Spark08_RDD_Operator_Transform {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
val sc = new SparkContext(sparkConf)
// 算法 — sample
val rdd = sc.makeRDD(List(1,2,3,4,5,6,7,8,9,10))
// sample算子需要传递三个参数
// 1.第一个参数表示，抽取数据后是否将数据返回
// 2.第二个参数表示 如果是抽取不放回的场合（基准值的概念），抽取放回的场合（条件概率）
// 3.第三个参数 随机算法的种子 如果不传递时，会使用当前系统时间
// println(rdd.sample(withReplacement = false, 0.4).collect().mkString(","))
println(rdd.sample(withReplacement = true, 2).collect().mkString(","))
sc.stop()
}
}
sample，指采样方法，可以随机抽取一些数据，主要用来应对数据倾斜的情况。
distinct
/**
* @Description:
* @author: LiuGe
* @date: 2021/1/25
*/
object Spark09_RDD_Operator_Transform {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
val sc = new SparkContext(sparkConf)
// 算法 — distinct
val rdd = sc.makeRDD(List(1,2,3,4,1,2,3,4))
val rdd1: RDD[Int] = rdd.distinct()
rdd1.collect().foreach(println)
sc.stop()
}
}
去重，非常常规的功能。
总结
总的来说，在花费了大量时间学习scala后，发现Spark学习起来十分简单，因为和scala的语法相似度十分高，基本就相当于在写scala的程序。也是没有白学那么久的scala