Spark的核心是建立在同一的抽象弹性分布式数据集(Resilient Distributed Datasets,RDD)之上的,这使得Spark的各个组件可以无缝的进行集成,能够在同一个应用程序中完成大数据处理
1.RDD基本概念
RDD是spark提供的最重要的抽象概念,它是一种有容错机制的特殊数据集合,可以分布在集群的节点上,以函数式操作集合的方式进行各种并行操作
可以将RDD理解为一个分布式对象集合,本质上是一个只读的分区记录集合。每个RDD可以分成多个分区,每个分区就是一个数据集片段。一个RDD的不同分区可以保存到集群中的不同节点上,从而可以在集群中的不同节点上进行并行计算。
弹性:
存储的弹性:内存与磁盘的自动切换
容错的弹性:数据丢失可以自动恢复
计算的弹性:计算出错重试机制
分片的弹性:可以根据需要重新分片
分布式:
数据存储在大数据集群不同节点上
数据集:
RDD封装了计算逻辑,并不保存数据
数据抽象:
RDD是一个抽象类,需要子类具体实现
不可变:
RDD封装了计算逻辑,是不可以改变的,想要改变,只能产生新的RDD,在新的RDD里面封装计算逻辑(有向无环图DAG)
可分区:
实现并行计算
2.RDD的核心属性
分区列表
RDD数据结构中存在分区列表,用于执行任务时并行计算,是实现分布式计算的重要属性
分区计算函数
Spark在计算时,是使用分区函数对每一个分区进行计算
RDD之间的依赖关系
RDD是计算模型的封装,当需求中需要将多个计算模型进行组合是,就需要将多个RDD建立依赖关系
分区器(可选)
当数据为KV类型数据时,可以通过设定分区器自定义数据的分区
首选位置(可选)
计算数据时,可以根据计算节点的状态选择不同的节点位置进行计算
3.RDD工作原理
1)启动Yarn集群环境
2)Spark通过申请资源创建调度节点和计算节点
3)Spark框架根据需求将计算逻辑根据分区划分成不同的任务
4)调度节点将任务根据计算节点状态发送到对应的计算节点进行计算
从以上流程可以看出RD在整个流程中主要用于讲逻辑进行封装,并生成Task发送给Executor节点执行计算
4.RDD创建
在Spark中创建RDD的方式可以分为四种:
1)从集合(内存)中创建RDD
从集合中创建RDD,Spark主要提供了两个方法:parallelize和makeRDD
val conf = new SparkConf().setMaster("local[*]").setAppName("test1")val sc = new SparkContext(conf)val rdd1 = sc.parallelize(List(1, 2, 3, 4))val rdd2 = sc.makeRDD(List(1, 2, 3, 4))println(rdd1.collect().mkString(","))println(rdd2.collect().mkString(","))sc.stop()
从底层代码实现来讲,makeRDD方法其实就是parallelize方法
/** Distribute a local Scala collection to form an RDD.** This method is identical to `parallelize`.* @param seq Scala collection to distribute* @param numSlices number of partitions to divide the collection into* @return RDD representing distributed collection*/def makeRDD[T: ClassTag](seq: Seq[T],numSlices: Int = defaultParallelism): RDD[T] = withScope {parallelize(seq, numSlices)}
2)从外部存储(文件)创建RDD
由外部存储系统的数据集创建RDD包括:本地的文件系统,所有Hadoop支持的数据集,比如HDFS、HBase等
val conf = new SparkConf().setMaster("local[*]").setAppName("test1")val sc = new SparkContext(conf)val rdd = sc.textFile("file:///D:\\Study\\13_spark\\cha01\\file\\tweets.txt")rdd.collect().foreach(println)sc.stop()
3)从其他RDD创建
主要是通过一个RDD运算完后,在产生新的RDD
4)直接创建RDD(new)
使用new的方式直接构造RDD一般由Spark框架自身使用
5.RDD算子
Value类型
1)map
作用:返回一个新的RDD,该RDD由每一个输入元素经过func函数转换后组成
需求:创建一个RDD,使每个元素*2组成新的RDD
object SparkRdd {def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建RDDval listRdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5))//4、每个数变成原来的2倍val rsRdd: RDD[Int] = listRdd.map(_ * 2)//5、打印rsRdd.foreach(println(_))}}
2)mapPartitions
类似于map,但独立地在RDD的每一个分片上运行,因此在类型为T的RDD上运行时,func的函数类型必须是Iterator[T] => Iterator[U]。假设有N个元素,有M个分区,那么map的函数的将被调用N次,而mapPartitions被调用M次,一个函数一次处理所有分区。
需求:创建一个RDD,使每个元素*2组成新的RDD
object SparkRdd {def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建RDDval listRdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5))//4、每个数变成原来的2倍val rsRdd: RDD[Int] = listRdd.mapPartitions(i=>{i.map(_*2)})//5、打印rsRdd.foreach(println)}}
3)mapPartitionsWithIndex(func)
类似于mapPartitions,但func带有一个整数参数表示分片的索引值,因此在类型为T的RDD上运行时,func的函数类型必须是(Int, Interator[T]) => Iterator[U];
需求:创建一个RDD,使每个元素跟所在分区形成一个元组组成一个新的RDD
object SparkRdd {def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建2个分区的RDDval listRdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5), 2)//4、每个数变成原来的2倍val rsRdd: RDD[(Int, Int)] = listRdd.mapPartitionsWithIndex((index, item) => {item.map((index, _))})//5、打印rsRdd.foreach(println)}}
4)flatMap(func)
对集合中每个元素进行操作然后再扁平化。
需求:把集合中元素按照’ '提取出来
object SparkRdd {def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建2个分区的RDDval listRdd: RDD[String] = sc.makeRDD(List("Hello World","Hello Spark"))//4、扁平化val rsRdd: RDD[String] = listRdd.flatMap(_.split(" "))//5、打印rsRdd.foreach(println)}}
5)glom
作用:将每一个分区形成一个数组,形成新的RDD类型时RDD[Array[T]]
需求:创建一个3个分区的RDD,并将每个分区的数据放到一个数组
object SparkRdd {def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建3个分区的RDDval listRdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4, 5, 6, 7, 8), 3)//4、glomval rsRdd: RDD[Array[Int]] = listRdd.glom()//5、打印rsRdd.foreach(item => {println(item.mkString(","))})}}
6)groupBy
分组,按照传入函数的返回值进行分组。将相同的key对应的值放入一个迭代器。
需求:创建一个RDD,按照元素模以2的值进行分组。
def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建RDDval listRdd: RDD[Int] = sc.makeRDD(1 to 20)//4、分组val rsRdd: RDD[(Int, Iterable[Int])] = listRdd.groupBy(_%2)//5、打印rsRdd.foreach(t=>println(t._1,t._2))}
7)filter
作用:过滤。返回一个新的RDD,该RDD由经过func函数计算后返回值为true的输入元素组成。
需求:创建一个RDD ,过滤出和2取余为0的数
object SparkRdd {def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建RDDval listRdd: RDD[Int] = sc.makeRDD(1 to 10)//4、过滤val rsRdd: RDD[Int] = listRdd.filter(_%2==0)//5、打印rsRdd.foreach(println)}}
8)distinct
去重分区
object SparkRdd {def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建RDDval listRdd: RDD[Int] = sc.makeRDD(List(1,2,1,2,1,2))//4、用2个分区保存结果val rsRdd: RDD[Int] = listRdd.distinct(2)//5、打印rsRdd.foreach(println)}}
9)repartition(numPartitions)
重分区,根据分区数,重新通过网络随机洗牌所有数据。
需求:创建一个4个分区的RDD,对其重新分区
object SparkRdd {def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建4个分区的Rddval listRdd: RDD[Int] = sc.makeRDD(1 to 10, 4)println("repartition before:" + listRdd.partitions.size)//4、重新分区val rsRdd: RDD[Int] = listRdd.repartition(2)//5、打印println("repartition after:" + rsRdd.partitions.size)}}
10)sortBy(func,[ascending], [numTasks])
作用;使用func先对数据进行处理,按照处理后的数据比较结果排序,默认为正序。
需求:创建一个RDD,按照不同的规则进行排序
object SparkRdd {def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建4个分区的Rddval listRdd: RDD[Int] = sc.makeRDD(1 to 10)//4、升序val sortAsc: Array[Int] = listRdd.sortBy(x => x).collect()println(sortAsc.mkString(","))//5、降序val sortDesc: Array[Int] = listRdd.sortBy(x => x, false).collect()println(sortDesc.mkString(","))}}
双Value类型交互
1) union(otherDataset)
作用:对源RDD和参数RDD求并集后返回一个新的RDD
需求:创建两个RDD,求并集
object SparkRdd {def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建2个listval list1: RDD[Int] = sc.makeRDD(1 to 2)val list2: RDD[Int] = sc.makeRDD(2 to 4)//4、求并级val rsRdd: Array[Int] = list1.union(list2).collect()//5、打印rsRdd.foreach(println)}}
2)subtract (otherDataset)
差集,计算差的一种函数,去除两个RDD中相同的元素,不同的RDD将保留下来,创建两个RDD,求第一个RDD与第二个RDD的差集
object SparkRdd {def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建2个listval list1: RDD[Int] = sc.makeRDD(1 to 10)val list2: RDD[Int] = sc.makeRDD(5 to 10)//4、求并级val rsRdd: Array[Int] = list1.subtract(list2).collect()//5、打印rsRdd.foreach(println)}}
3)intersection(otherDataset)
交集,对源RDD和参数RDD求交集后返回一个新的RDD
object SparkRdd {def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建2个listval list1: RDD[Int] = sc.makeRDD(1 to 10)val list2: RDD[Int] = sc.makeRDD(5 to 10)//4、交集val rsRdd: Array[Int] = list1.intersection(list2).collect()//5、打印rsRdd.foreach(println)}}
Key-Value类型
1)groupByKey
groupByKey也是对每个key进行操作,但只生成一个sequence。
需求:wordcount
object SparkRdd {def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建listval list: RDD[String] = sc.makeRDD(List("Hello World","Hello Scala","Spark Spark Spark"))//4、wordcountval mapRdd: RDD[(String, Int)] = list.flatMap(_.split(" ")).map((_,1))//5、groupByKeyval groupByKeyRdd: RDD[(String, Iterable[Int])] = mapRdd.groupByKey()groupByKeyRdd.foreach(println)//6、聚合val wcRdd: RDD[(String, Int)] = groupByKeyRdd.map(t=>(t._1,t._2.size))wcRdd.foreach(println)}}
2)reduceByKey
在一个(K,V)的RDD上调用,返回一个(K,V)的RDD,使用指定的reduce函数,将相同key的值聚合到一起,reduce任务的个数可以通过第二个可选的参数来设置。
需求:wordcount
object SparkRdd {def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建listval list: RDD[String] = sc.makeRDD(List("Hello World","Hello Scala","Spark Spark Spark"))//4、扁平化val mapRdd: RDD[(String, Int)] = list.flatMap(_.split(" ")).map((_,1))//5、wordcountval reduceMapRdd: RDD[(String, Int)] = mapRdd.reduceByKey(_+_)//6、打印reduceMapRdd.foreach(println)}}
3)sortByKey([ascending], [numTasks])
作用:在一个(K,V)的RDD上调用,K必须实现Ordered接口,返回一个按照key进行排序的(K,V)的RDD
创建一个pairRDD,按照key的正序和倒序进行排序
object SparkRdd {def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建listval listRdd: RDD[(Int, Int)] = sc.makeRDD(List((1, 3), (1, 2), (1, 4), (2, 3), (3, 6), (3, 8)))//4、升序val sortAsc: Array[(Int, Int)] = listRdd.sortByKey(true).collect()println("升序:" + sortAsc.mkString(","))//5、降序val sortDesc: Array[(Int, Int)] = listRdd.sortByKey(false).collect()println("降序:" + sortDesc.mkString(","))}}
4)join(otherDataset, [numTasks])
在类型为(K,V)和(K,W)的RDD上调用,返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD
object SparkRdd {def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)val r1: RDD[(Int, String)] = sc.makeRDD(Array((1, "a"), (2, "b"), (3, "c")))val r2: RDD[(Int, String)] = sc.makeRDD(Array((1, "a"), (2, "b"), (3, "c")))val rsRdd: Array[(Int, (String, String))] = r1.join(r2).collect()rsRdd.foreach(println)}}
Action算子
1)reduce,作用:通过func函数聚集RDD中的所有元素,先聚合分区内数据,再聚合分区间数据。
需求:创建一个RDD,将所有元素聚合得到结果。
object SparkRdd {def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建RDDval rdd: RDD[Int] = sc.makeRDD(1 to 5)//4、reduceval rs: Int = rdd.reduce(_+_)//5、结果println(rs)}}
2) collect()
在驱动程序中,以数组的形式返回数据集的所有元素。
创建一个RDD,并将RDD内容收集到Driver端打印
object SparkRdd {def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建RDDval lisrRdd: RDD[Int] = sc.makeRDD(List(1,2,3,4,5))val ints: Array[Int] = lisrRdd.collect()println(ints.mkString(","))}}
3) count()
返回RDD中元素的个数
object SparkRdd {def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建RDDval rdd: RDD[Int] = sc.makeRDD(1 to 10)val rddCount: Long = rdd.count()println(rddCount)}}
4)first()
返回RDD中的第一个元素
object SparkRdd {def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建RDDval rdd: RDD[Int] = sc.makeRDD(1 to 10)//4、返回RDD中第一个元素val firstNum: Int = rdd.first()println(firstNum)}}
5) take(n)
返回一个由RDD的前n个元素组成的数组
object SparkRdd {def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建RDDval rdd: RDD[Int] = sc.makeRDD(1 to 10)val top3: Array[Int] = rdd.take(3)top3.foreach(println)}}
6) takeOrdered(n)
返回该RDD排序后的前n个元素组成的数组
def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建RDDval rdd: RDD[Int] = sc.makeRDD(List(5, 3, 2, 1, 7, 6))val top3: Array[Int] = rdd.takeOrdered(3)println(top3.mkString(","))}
7)countByKey()
针对(K,V)类型的RDD,返回一个(K,Int)的map,表示每一个key对应的元素个数。
object SparkRdd {def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建RDDval rdd: RDD[(Int, Int)] = sc.makeRDD(List((1,3),(1,2),(1,4),(2,3),(3,6),(3,8)))//4、统计keyval countKey: collection.Map[Int, Long] = rdd.countByKey()//5、打印println(countKey)}}
8)foreach
打印
object SparkRdd {def main(args: Array[String]): Unit = {//1、创建本地spark配置文件val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")//2、创建Spark上下文对象val sc = new SparkContext(config)//3、创建RDDval rdd: RDD[Int] = sc.makeRDD(List(1,2,3,4,5))rdd.foreach(println)}}