【Spark】Spark编程体验，RDD转换算子、执行算子操作（六）

Spark编程体验

项目依赖管理

<dependencies><dependency><groupId>org.scala-lang</groupId><artifactId>scala-library</artifactId><version>2.12.10</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.12</artifactId><version>3.2.1</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_2.12</artifactId><version>3.2.1</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.23</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-hive_2.12</artifactId><version>3.2.1</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming_2.12</artifactId><version>3.2.1</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-kafka-0-10_2.12</artifactId><version>3.2.1</version></dependency>
</dependencies><build><finalName>chapter1.WordCount</finalName><plugins><plugin><groupId>net.alchim31.maven</groupId><artifactId>scala-maven-plugin</artifactId><version>3.4.6</version><executions><execution><goals><goal>compile</goal><goal>testCompile</goal></goals></execution></executions></plugin></plugins>
</build>

项目编码

spark入门程序wordcount：

package com.fesco.bigdata.spark
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
/*** scala版本的wordcount*/
object ScalaWordCountApp {
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
.setAppName(s"${ScalaWordCountApp.getClass.getSimpleName}")
.setMaster("local[*]")
val sc = new SparkContext(conf)
//加载数据
val file: RDD[String] = sc.textFile("file:/E:/data/spark/hello.txt")//按照分隔符进行切分
val words:RDD[String] = lines.flatMap(line => line.split("\\s+"))
//每个单词记为1次
val pairs:RDD[(String, Int)] = words.map(word => (word, 1))
//聚合数据
val ret:RDD[(String, Int)] = pairs.reduceByKey(myReduceFunc)
//export data to external system
ret.foreach(println)
}
sc.stop()
}
def myReduceFunc(v1: Int, v2: Int): Int = {
v1 + v2
}
}

Master URL说明
首先在编程过程中，至少需要给spark程序传递一个参数master-url，通过sparkConf.setMaster来完成。改参数，代表的是spark作业的执行方式，或者指定的spark程序的cluster-manager的类型。
表-1 模式选择
在这里插入图片描述

spark程序的其他提交方式

加载hdfs中的文件：

object RemoteSparkWordCountOps {
def main(args: Array[String]): Unit = {//创建程序入口val conf = new SparkConf().setAppName("wc").setMaster("local[*]")val sc = new SparkContext(conf)//设置日志级别sc.setLogLevel("WARN")//加载数据val file = sc.textFile("hdfs://hadoop101:8020//wordcount//words.txt")//切分val spliFile: RDD[String] = file.flatMap(_.split(" "))//每个单词记为1次val wordAndOne: RDD[(String, Int)] = spliFile.map((_, 1))//聚合val wordAndCount: RDD[(String, Int)] = wordAndOne.reduceByKey(_ + _)//打印输出wordAndCount.foreach(println)//释放资源sc.stop()
}
}

提交spark程序到集群中

首先需要将spark-core模块进行打包，其次上传到集群中，才可以进行提交作业到spark或者yarn集群中运行。
1)Client：

bin/spark-submit \
--class chapter1.WordCount \
--master spark://hadoop101:7077 \
/root/word.jar \
hdfs://hadoop101:8020/wordcount/words.txt

2)Cluster：

bin/spark-submit \
--class chapter1.WordCount \
--master spark://hadoop101:7077 \
/root/word.jar \
hdfs://hadoop101:8020/wordcount/words.txt \
hdfs://hadoop101:8020/wordcount/output1

Spark RDD操作

Spark执行流程
在上一讲中，我们知道了什么是Spark，什么是RDD、Spark的核心构成组件，以及Spark案例程序。在这一讲中，我们将继续需要Spark作业的执行过程，以及编程模型RDD的各种花式操作，首先来学习Spark作业执行流程。
WordCount执行流程
Wordcount执行流程如图-1所示。
在这里插入图片描述

图-1 wordcount执行流程
在上图中我们可以看到rdd（partition）和rdd之间是有依赖关系的，大致分为两种：窄依赖（Narrow Dependency）和宽依赖（Wide/Shuffle Dependency）。
1)窄依赖：rdd中partition中的数据，只依赖于父rdd中的一个分区，把这种依赖关系，称之为窄依赖，常见的窄依赖操作有：flatMap、map、filter、union、coalesce等等。
2)宽依赖：与窄依赖对应的，partition中的数据，依赖于父rdd中所有的partition，把这种依赖关系称之为宽依赖，常见的宽依赖操作：distinct、reduceByKey、groupByKey、repartition等等。
总而言之，rdd和rdd是有依赖关系的，我们把rdd和rdd之间关系构成的一个图或者依赖的链条，称之为rdd的lineage（血统），是保障spark容错的一个重要支撑。
Spark作业提交流程
下面我们一同来看spark作业提交流程，流程如图-2所示:
在这里插入图片描述

图-2 spark作业提交流程
RDD的基本操作
RDD概述
在较高的层次上，每个Spark应用程序都由一个驱动程序组成，该驱动程序运行用户的主功能并在集群上执行各种并行操作。Spark提供的主要抽象是一个弹性分布式数据集（RDD），它是一个跨集群节点划分的元素集合，可以并行操作。RDD是从Hadoop文件系统（或任何其他支持Hadoop的文件系统）中的一个文件或驱动程序中现有的Scala集合开始创建的，并对其进行转换。用户还可以要求Spark将RDD持久化在内存中，这样就可以跨并行操作高效地重用RDD。最后，RDD会自动从节点故障中恢复。
RDD分类
需要知道RDD操作算子的分类，基本上分为两类：transformation和action，当然更加细致的分，可以分为输入算子，转换算子，缓存算子，行动算子，整个RDD原生数据空间如图-3所示。

在这里插入图片描述
图-3 RDD原生数据空间
1)输入：在Spark程序运行中，数据从外部数据空间（如分布式存储：textFile读取HDFS等，parallelize方法输入Scala集合或数据）输入Spark，数据进入Spark运行时数据空间，转化为Spark中的数据块，通过BlockManager进行管理。
2)运行：在Spark数据输入形成RDD后便可以通过转换算子，如filter等，对数据进行操作并将RDD转化为新的RDD，通过Action算子，触发Spark提交作业。如果数据需要复用，可以通过Cache算子，将数据缓存到内存。
3)输出：程序运行结束数据会输出Spark运行时空间，存储到分布式存储中（如saveAsTextFile输出到HDFS），或Scala数据或集合中（collect输出到Scala集合，count返回Scala int型数据）。
RDD初始化
RDD的初始化，原生api提供的2种创建方式，一种就是读取文件textFile，还有一种就是加载一个scala集合parallelize。当然，也可以通过transformation算子来创建的RDD。
RDD算子使用
transformation转换操作
map算子
rdd.map(p: A => B):RDD，对rdd集合中的每一个元素，都作用一次该func函数，之后返回值为生成元素构成的一个新的RDD。

val sc = new SparkContext(conf)
//map 原集合*7
val list = 1 to 7
//构建一个rdd
val listRDD:RDD[Int] = sc.parallelize(list)
//listRDD.map((num:Int) => num * 7)
//listRDD.map(num => num * 7)
val ret = listRDD.map(_ * 7)
ret.foreach(println)

flatMap算子
rdd.flatMap(p: A => 集合):RDD ==>rdd集合中的每一个元素，都要作用func函数，返回0到多个新的元素，这些新的元素共同构成一个新的RDD。

def  flatMapOps(sc:SparkContext): Unit = {
val list = List(
"jia jing kan kan kan",
"gao di di  di di",
"zhan yuan qi qi"
)
val listRDD = sc.parallelize(list)
listRDD.flatMap(line => line.split("\\s+"))
.foreach(println)
}

mapPartitions算子
mapPartitions(p: Iterator[A] =>Iterator[B])，上面的map操作，一次处理一条记录；而mapPartitions一次性处理一个partition分区中的数据。
注意：虽说mapPartitions的执行性能要高于map，但是其一次性将一个分区的数据加载到执行内存空间，如果该分区数据集比较大，存在OOM的风险。

//创建RDD并指定分区数
val array = 1 to 10
val listRDD:RDD[Int] = sc.parallelize(array,2)	
//通过-将分区之间的数据连接
val result: RDD[String] = rdd.mapPartitions(x=>Iterator(x.mkString("-")))
//打印输出
println(result.collect().toBuffer)
mapPartitionsWithIndex算子
mapPartitionsWithIndex((index,p:Iterator[A]=>Iterator[B]))，该操作比mapPartitions多了一个index，代表就是后面p所对应的分区编号:rdd的分区编号，命名规范，如果有N个分区，分区编号就从0，...，N-1。
val rdd: RDD[Int] = sc.parallelize(1 to 16,4)
//查看每个分区当中都保存了哪些数据
val result: RDD[String] = rdd.mapPartitionsWithIndex((index,item)=>Iterator(index+":"+item.mkString(",")))
//打印输出
result.foreach(println)

sample算子
1)sample(withReplacement, fraction, seed):随机抽样算子，sample主要工作就是为了来研究数据本身，去代替全量研究会出现类似数据倾斜(dataSkew)等问题，无法进行全量研究，只能用样本去评估整体。
2)withReplacement：Boolean：有放回的抽样和无放回的抽样。
3)fraction：Double：样本空间占整体数据量的比例，大小在[0,1]。
4)seed：Long：是一个随机数的种子，有默认值，通常不需要传参。
需要说明一点的是，这个抽样是一个不准确的抽样，抽取的结果数可能在准确的结果上下浮动。

def sampleOps(sc: SparkContext): Unit = {
val list = sc.parallelize(1 to 100000)
val sampled1 = list.sample(true, 0.01)
println("sampled1 count: " + sampled1.count())
val sampled2 = list.sample(false, 0.01)
println("sampled2 count: " + sampled2.count())
}

union算子
rdd1.union(rdd2)相当于sql中的union all，进行两个rdd数据间的联合，需要说明一点是，该union是一个窄依赖操作，rdd1如果有N个分区，rdd2有M个分区，那么union之后的分区个数就为N+M。

val rdd1 = sc.parallelize(1 to 5,3)
//查看每个分区当中都保存了哪些数据
rdd1.mapPartitionsWithIndex((index,item)=>Iterator(index+":"+item.mkString(","))).foreach(println)
val rdd2: RDD[Int] = sc.parallelize(5 to 7,2)
//查看每个分区当中都保存了哪些数据
rdd2.mapPartitionsWithIndex((index,item)=>Iterator(index+":"+item.mkString(","))).foreach(println)
//union整合
val result: RDD[Int] = rdd1.union(rdd2)
//获取数据
println(result.collect().toBuffer)
//查看每个分区当中都保存了哪些数据
result.mapPartitionsWithIndex((index,item)=>Iterator(index+":"+item.mkString(","))).foreach(println)
//获取分区数
println(result.getNumPartitions)

join算子
1)rdd1.join(rdd2) 相当于sql中的join连接操作
A(id) a, B(aid) b
select * from A a join B b on a.id = b.aid
2)交叉连接: across join
select * from A a across join B ====>这回产生笛卡尔积。
3)内连接： inner join，提取左右两张表中的交集
select * from A a inner join B on a.id = b.aid 或者
select * from A a, B b where a.id = b.aid
4)外连接：outer join
5)左外连接：left outer join 返回左表所有，右表匹配返回，匹配不上返回null
select * from A a left outer join B on a.id = b.aid
6)右外连接：right outer join 刚好是左外连接的相反
select * from A a left outer join B on a.id = b.aid
7)全连接：full join
8)全外连接：full outer join = left outer join + right outer join
前提：要先进行join，rdd的类型必须是K-V。
对join操作可以归纳为如图-4所示。
在这里插入图片描述

图-4 sql 各种join连接图

def main(args: Array[String]): Unit = {
val conf: SparkConf = new SparkConf().setAppName("demo").setMaster("local[*]")
val sc = new SparkContext(conf)
sc.setLogLevel("WARN")
val rdd1: RDD[(Int, String)] = sc.parallelize(Array((1,"a"),(2,"b"),(3,"c")))
val rdd2: RDD[(Int, Int)] = sc.parallelize(Array((1,4),(2,5),(5,6)))
//join操作
val result: RDD[(Int, (String, Int))] = rdd1.join(rdd2)
//打印输出
println(result.collect().toBuffer)
//leftOutJoin操作
val result1: RDD[(Int, (String, Option[Int]))] = rdd1.leftOuterJoin(rdd2)
//打印输出
println(result1.collect().toBuffer)
//rightOuterJoin
val result2: RDD[(Int, (Option[String], Int))] = rdd1.rightOuterJoin(rdd2)
//打印输出
println(result2.collect().toBuffer)
//fullOuterJoin
val result3: RDD[(Int, (Option[String], Option[Int]))] = rdd1.fullOuterJoin(rdd2)
//打印输出
println(result3.collect().toBuffer)
}

coalesce算子
1)coalesce(numPartition, shuffle=false):分区合并的意思。
2)numPartition:分区后的分区个数。
3)shuffle：此次重分区是否开启shuffle，决定当前的操作是宽(true)依赖还是窄(false)依赖。

def main(args: Array[String]): Unit = {
val conf: SparkConf = new SparkConf().setAppName("wc").setMaster("local[*]")
val sc = new SparkContext(conf)
sc.setLogLevel("WARN")
val rdd = sc.parallelize(1 to 16,4)
//获取分区数
println(rdd.getNumPartitions)
//查看每个分区当中保存了哪些数据
rdd.mapPartitionsWithIndex((index,item)=>Iterator(index+":"+item.mkString(","))).foreach(println)
//缩减分区
val rdd1: RDD[Int] = rdd.coalesce(3)
//获取分区数
println(rdd1.getNumPartitions)
//查看每个分区当中保存的数据变化
rdd1.mapPartitionsWithIndex((index,item)=>Iterator(index+":"+item.mkString(","))).foreach(println)
//释放资源
sc.stop()
}

repartition(numPartitions)算子
根据分区数，从新通过网络随机洗牌所有数据。

def main(args: Array[String]): Unit = {
val conf: SparkConf = new SparkConf().setAppName("wc").setMaster("local[*]")
val sc = new SparkContext(conf)
//设置控制台日志级别
sc.setLogLevel("WARN")
val rdd = sc.parallelize(1 to 16,4)
//获取分区数
println(rdd.getNumPartitions)
//查看每个分区当中保存了哪些数据
rdd.mapPartitionsWithIndex((index,item)=>Iterator(index+":"+item.mkString(","))).foreach(println)
//缩减分区
val rdd1: RDD[Int] = rdd.repartition(3)
//获取分区数
println(rdd1.getNumPartitions)
//查看每个分区当中保存的数据变化
rdd1.mapPartitionsWithIndex((index,item)=>Iterator(index+":"+item.mkString(","))).foreach(println)
//释放资源
sc.stop()
}

sortBy(func,[ascending], [numTasks])算子
用func先对数据进行处理，按照处理后的数据比较结果排序。

def main(args: Array[String]): Unit = {
val conf: SparkConf = new SparkConf().setAppName("demo").setMaster("local[*]")
val sc = new SparkContext(conf)
//设置控制台日志输出级别
sc.setLogLevel("WARN")
//加载数据
val rdd= sc.parallelize(List(("a",4),("c",2),("b",1)))
//默认是升序排序，指定false，转为倒叙输出
val rdd1: RDD[(String, Int)] = rdd.sortBy(_._2,false)
//收集结果，返回数组输出
println(rdd1.collect().toBuffer)
}

sortByKey([ascending],[numTasks])算子
在一个(K,V)的RDD上调用，K必须实现Ordered接口，返回一个按照key进行排序的(K,V)的RDD。

def main(args: Array[String]): Unit = {
val conf: SparkConf = new SparkConf().setAppName("demo").setMaster("local[*]")
val sc = new SparkContext(conf)
//设置控制台日志输出级别
sc.setLogLevel("WARN")
//加载数据
val rdd = sc.parallelize(Array((3,"aa"),(6,"cc"),(2,"bb"),(1,"dd")))
//默认按照key进行升序输出，加false，转为倒叙输出
val result: RDD[(Int, String)] = rdd.sortByKey(false)
//收集结果，返回数组输出
println(result.collect().toBuffer)
}

groupByKey算子
groupByKey(numPartition):[K, Iterable[V]]
按照key来进行分组，numPartition指的是分组之后的分区个数。
这是一个宽依赖操作，但是需要注意一点的是，groupByKey相比较reduceByKey而言，没有本地预聚合操作，显然其效率并没有reduceByKey效率高，在使用的时候如果可以，尽量使用reduceByKey等去代替groupByKey。

def groupByOps(sc: SparkContext): Unit = {
case class Student(id: Int, name:String, province: String)
val stuRDD = sc.parallelize(List(
Student(1, "张三", "安徽"),
Student(2, "李梦", "山东"),
Student(3, "王五", "甘肃"),
Student(4, "周七", "甘肃"),
Student(5, "Lucy", "黑吉辽"),
Student(10086, "魏八", "黑吉辽")
))
//val province2Infos: RDD[(String, Iterable[Student]] = stuRDD.groupBy(stu => stu.province)
val province2Infos: RDD[(String, Iterable[Student])] = stuRDD.map(stu => (stu.province, stu)).groupByKey()province2Infos.foreach{case (province, stus) => {
println(s"省份：${province}, 学生信息：${stus.mkString(", ")}, 人数：${stus.size}")
}}
}
}

reduceByKey算子
reduceByKey((A1, A2) => A3)
前提不是对全量的数据集进行reduce操作，而是对每一个key所对应的所有的value进行reduce操作。

def reduceByKeyOps(sc: SparkContext): Unit = {
case class Student(id: Int, name:String, province: String)
val stuRDD = sc.parallelize(List(
Student(1, "张三", "安徽"),
Student(2, "李梦", "山东"),
Student(3, "王五", "甘肃"),
Student(4, "周七", "甘肃"),
Student(5, "Lucy", "黑吉辽"),
Student(10086, "魏八", "黑吉辽")
))
val ret = stuRDD.map(stu => (stu.province, 1)).reduceByKey((v1, v2) => v1 + v2)
ret.foreach{case (province, count) => {
println(s"province: ${province}, count: ${count}")
}}
}

foldByKey算子
foldByKey(zeroValue)((A1, A2) => A3)，其作用和reduceByKey一样，唯一的区别就是zeroValue初始化值不一样，相当于在scala集合操作中的reduce和fold的区别。

def foldByKeyOps(sc: SparkContext): Unit = {
case class Student(id: Int, name:String, province: String)
val stuRDD = sc.parallelize(List(
Student(1, "张三", "安徽"),
Student(3, "王五", "甘肃"),
Student(5, "Lucy", "黑吉辽"),
Student(2, "李梦", "山东"),
Student(4, "周七", "甘肃"),
Student(10086, "魏八", "黑吉辽")
), 2).mapPartitionsWithIndex((index, partition) => {
val list = partition.toList
println(s"-->stuRDD的分区编号为<${index}>中的数据为：${list.mkString("[", ", ", "]")}")
list.toIterator
})
val ret = stuRDD.map(stu => (stu.province, 1)).foldByKey(0)((v1, v2) => v1 + v2)
ret.foreach{case (province, count) => {
println(s"province: ${province}, count: ${count}")
}}
}

aggregateByKey算子
combineByKey和aggregateByKey的区别就相当于reduceByKey和foldByKey。

def abk2rbk(sc: SparkContext): Unit = {
val array = sc.parallelize(Array(
"hello you",
"hello me",
"hello you",
"hello you",
"hello me",
"hello you"
), 2)
val pairs = array.flatMap(_.split("\\s+")).map((_, 1))
val ret = pairs.aggregateByKey(0)(_+_, _+_)
ret.foreach{case (key, count) => {
println(s"key: ${key}, count: ${count}")
}}
}

action行动算子操作
1)foreach算子：foreach主要功能，就是用来遍历RDD中的每一条纪录，其实现就是将map或者flatMap中的返回值变为Unit即可，即foreach(A => Unit)。
在上述transformation操作学习过程中，多次使用到了foreach算子，所以这里就跳过学习了。
2)count算子：统计该rdd中元素的个数。
val count = rdd.count()
println(“rdd的count个数为：” + count)
3)collect算子：字面意思就是收集，拉取的意思，该算子的含义就是将分布在集群中的各个partition中的数据拉回到driver中，进行统一的处理；但是这个算子有很大的风险存在，第一，driver内存压力很大，第二数据在网络中大规模的传输，效率很低；所以一般不建议使用，如果非要用，请先执行filter。

val arr = rdd.filter(_._2 > 2).collect()
arr.foreach(println)

4)take&first：返回该rdd中的前N个元素,如果该rdd的数据是有序的，那么take(n)就是TopN;而first是take(n)中比较特殊的一个take(1)。

val arr:Array[(String, Int)] = rdd.take(2)
arr.foreach(println)val ret:(String, Int) = rdd.first()
println(ret)

5)takeOrdered(n)：返回前几个的排序。

val arr:Array[(String, Int)] = rdd.takeOrdered(2)
arr.foreach(println)

6)reduce算子：需要清楚的是，reduce是一个action操作，reduceByKey是一个transformation。reduce对一个rdd执行聚合操作，并返回结果，结果是一个值。

//例子1
val rdd: RDD[Int] = sc.parallelize(1 to 5,2)
//聚合
val result: Int = rdd.reduce(_+_)
//打印输出
println(result)
//例子2
val rdd2 = sc.makeRDD(Array(("a",1),("a",3),("c",3),("d",5)))
//聚合
val result1= rdd2.reduce((x,y)=>(x._1 + y._1,x._2 + y._2))
//打印输出
println(result1)
需要注意一点的是，聚合前后的数据类型保持一致。
7)countByKey算子：统计key出现的次数。
val rdd = sc.parallelize(List((1,3),(1,2),(1,4),(2,3),(3,6),(3,8)),3)
//统计相同key出现的次数
val result: collection.Map[Int, Long] = rdd.countByKey()
//打印输出
println(result)

8)saveAsXxx算子：

//saveAsTextFile
rdd.saveAsTextFile("file:/E:/data/out/")
//saveAsNewAPIHadoopFile
val path = "file:/E:/data/out1"
rr.saveAsNewAPIHadoopFile(path,
classOf[Text],
classOf[IntWritable],
classOf[TextOutputFormat[Text, IntWritable]])
9)foreachPartition算子：foreach写入写入数据库，但是极其不友好。
def saveInfoMySQL2(rdd: RDD[(String, Int)]): Unit = {
rdd.foreach{case (word, count) => {
Class.forName("com.mysql.jdbc.Driver")
val url = "jdbc:mysql://localhost:3306/test"
val connection = DriverManager.getConnection(url, "mark", "sorry")
val sql =
"""
|insert into wordcounts(word, `count`) Values(?, ?)
|""".stripMargin
val ps = connection.prepareStatement(sql)
ps.setString(1, word)
ps.setInt(2, count)
ps.execute()
ps.close()
connection.close()
}}
}
val rdd: RDD[(String, Int)] = sc.parallelize(List(("hadoop",2)))
save(rdd)

高效写入数据库：

def saveInfoMySQLByForeachPartition(rdd: RDD[(String, Int)]): Unit = {
rdd.foreachPartition(partition => {
//这是在partition内部，属于该partition的本地
Class.forName("com.mysql.jdbc.Driver")
val url = "jdbc:mysql://localhost:3306/test"
val connection = DriverManager.getConnection(url, "mark", "sorry")
val sql =
"""
|insert into wordcounts(word, `count`) Values(?, ?)
|""".stripMargin
val ps = connection.prepareStatement(sql)
partition.foreach{case (word, count) => {
ps.setString(1, word)
ps.setInt(2, count)
ps.execute()
}}
ps.close()
connection.close()
})
}