SparkCore学习笔记

一. RDD概述

1.1 什么是RDD

	RDD 叫做弹性分布式数据集，是Spark中最基本的数据抽象。它代表的是一个弹性的、不可变、可分区、里面的元素可并行计算的集合。

1.1.1WordCount工作流程

	首先在yarn模式下将生成的WordCount的Jar包导入，然后启动bin/spark-submit --class WordCount --master yarn --deploy-mode cluster WordCount.jar /input /output   然后向yarn提交submit，在Container容器里开启Driver，然后再开启Executor，textFile从HDFS上读取数据，在Executor上调用FlatMap方法，接收lineRDD传过来的数据，FlatMap处理以后调用map方法，然后调用reduceByKey方法，reduceBykey方法调用的时候会进行shuffler，对数据进行打散、混洗、聚合，同时，在这个阶段数据会落盘。然后调用collect,收集的数据最后传回Driver.

注意

1.所有的RDD算子相关操作都在Executor端执行，RDD算子之外的操作都在Driver端执行
2.Spark中，只有遇到action等行动算子，才会执行RDD的计算（即延迟计算）

RDD几大特点

1) 弹性存储的弹性：内存和磁盘自动切换容错的弹性：数据丢失可以自动恢复计算的弹性：计算出错重试机制分片的弹性：可根据需要重新分片
2） 分布式数据存储在大数据集群的不同节点上
3） 数据集，不存储数据RDD封装了计算逻辑，并不保存数据
4） 数据抽象RDD是一个抽象类，需要子类具体实现
5）不可变RDD封装了计算逻辑，是不可以改变的，想要改变只能产生新的RDD，在新的RDD里面封装计算逻辑
6）可分区，并行计算

1.2 RDD的五大特性

二. RDD编程

2.1 RDD的创建

在Spark中创建RDD的创建方式可以分为三种：从集合中创建RDD、从外部存储创建RDD、从其他RDD创建。

2.1.1 从集合中创建

1）从集合中创建RDD，Spark主要提供了两种函数：parallelize和makeRDD

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object createrdd01_array {def main(args: Array[String]): Unit = {//1.创建SparkConf并设置App名称val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")//2.创建SparkContext，该对象是提交Spark App的入口val sc: SparkContext = new SparkContext(conf)//3.使用parallelize()创建rddval rdd: RDD[Int] = sc.parallelize(Array(1, 2, 3, 4, 5, 6, 7, 8))rdd.collect().foreach(println)//4.使用makeRDD()创建rddval rdd1: RDD[Int] = sc.makeRDD(Array(1, 2, 3, 4, 5, 6, 7, 8))rdd1.collect().foreach(println)sc.stop()}
}

注意：makeRDD有两种重构方法，重构方法一如下，makeRDD和parallelize功能一样。

def makeRDD[T: ClassTag](seq: Seq[T],numSlices: Int = defaultParallelism): RDD[T] = withScope {parallelize(seq, numSlices)
}

2）makeRDD的重构方法二，增加了位置信息

注意：只需要知道makeRDD不完全等于parallelize即可。

def makeRDD[T: ClassTag](seq: Seq[(T, Seq[String])]): RDD[T] = withScope {assertNotStopped()val indexToPrefs = seq.zipWithIndex.map(t => (t._2, t._1._2)).toMapnew ParallelCollectionRDD[T](this, seq.map(_._1), math.max(seq.size, 1), indexToPrefs)
}

在大部分情况下，我们采用makeRDD方法！

2.1.2 从外部存储系统的数据集创建

由外部存储系统的数据集创建RDD包括：本地的文件系统，还有所有Hadoop支持的数据集，比如HDFS、HBase等。

1）数据准备

在新建的SparkCoreTest项目名称上右键=>新建input文件夹=>在input文件夹上右键=>分别新建1.txt和2.txt。每个文件里面准备一些word单词。

2）创建RDD

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object createrdd02_file {def main(args: Array[String]): Unit = {//1.创建SparkConf并设置App名称val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")//2.创建SparkContext，该对象是提交Spark App的入口val sc: SparkContext = new SparkContext(conf)//3.读取文件。如果是集群路径：hdfs://hadoop102:9000/inputval lineWordRdd: RDD[String] = sc.textFile("input")//4.打印lineWordRdd.foreach(println)//5.关闭sc.stop()}
}

2.1.3 从其他RDD创建

主要是通过一个RDD运算完后，再产生新的RDD。

2.2 分区规则

2.2.1 默认分区源码（RDD数据从集合中创建）

1）默认分区数源码解读

2）代码

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object partition01_default {def main(args: Array[String]): Unit = {//1.创建SparkConf并设置App名称val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")//2.创建SparkContext，该对象是提交Spark App的入口val sc: SparkContext = new SparkContext(conf)val rdd: RDD[Int] = sc.makeRDD(Array(1,2,3,4))//3. 输出数据，产生了8个分区rdd.saveAsTextFile("output")//4.关闭连接sc.stop()}
}

2.2.2 分区源码（RDD数据从集合中创建）

1）分区测试（RDD数据从集合中创建）

object partition02_Array {def main(args: Array[String]): Unit = {val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkCoreTest")val sc: SparkContext = new SparkContext(conf)//1）4个数据，设置4个分区，输出：0分区->1，1分区->2，2分区->3，3分区->4//val rdd: RDD[Int] = sc.makeRDD(Array(1, 2, 3, 4), 4)//2）4个数据，设置3个分区，输出：0分区->1，1分区->2，2分区->3,4//val rdd: RDD[Int] = sc.makeRDD(Array(1, 2, 3, 4), 3)//3）5个数据，设置3个分区，输出：0分区->1，1分区->2、3，2分区->4、5val rdd: RDD[Int] = sc.makeRDD(Array(1, 2, 3, 4, 5), 3)rdd.saveAsTextFile("output")sc.stop()}
}

2）分区源码

分区的开始位置 = 分区号 * 数据总长度/分区总数

分区的结束位置 =（分区号 + 1） 数据总长度/分区总数*

2.2.3 默认分区源码（RDD数据从文件中读取后创建）

1）分区测试

object partition03_file_default {def main(args: Array[String]): Unit = {val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkCoreTest")val sc: SparkContext = new SparkContext(conf)//1）默认分区的数量：默认取值为当前核数和2的最小值//val rdd: RDD[String] = sc.textFile("input")rdd.saveAsTextFile("output")sc.stop()}
}

2）分区源码

2.2.4 分区源码（RDD数据从文件中读取后创建）

1）分区测试

object partition04_file {def main(args: Array[String]): Unit = {val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkCoreTest")val sc: SparkContext = new SparkContext(conf)//1）输入数据1-4，每行一个数字；输出：0=>{1、2} 1=>{3} 2=>{4} 3=>{空}//val rdd: RDD[String] = sc.textFile("input/3.txt", 3)rdd.saveAsTextFile("output")sc.stop()}
}

2）源码解析

注意：getSplits文件返回的是切片规划，真正读取是在compute方法中创建LineRecordReader读取的，有两个关键变量： start = split.getStart() end = start + split.getLength

2.3 Transformation转换算子

RDD整体上分为Value类型、双Value类型和Key-Value类型

2.3.1 Value类型

2.3.1.1 map()映射

map算子

代码实现

object value01_map {def main(args: Array[String]): Unit = {//1.创建SparkConf并设置App名称val conf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")//2.创建SparkContext，该对象是提交Spark App的入口val sc = new SparkContext(conf)//3具体业务逻辑// 3.1 创建一个RDDval rdd: RDD[Int] = sc.makeRDD(1 to 4, 2)// 3.2 调用map方法，每个元素乘以2val mapRdd: RDD[Int] = rdd.map(_ * 2)// 3.3 打印修改后的RDD中数据mapRdd.collect().foreach(println)//4.关闭连接sc.stop()}
}

2.3.1.2 mapPartitions()以分区为单位执行Map

mapPartitions算子

代码实现

object value02_mapPartitions {def main(args: Array[String]): Unit = {//1.创建SparkConf并设置App名称val conf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")//2.创建SparkContext，该对象是提交Spark App的入口val sc = new SparkContext(conf)//3具体业务逻辑// 3.1 创建一个RDDval rdd: RDD[Int] = sc.makeRDD(1 to 4, 2)// 3.2 调用mapPartitions方法，每个元素乘以2val rdd1 = rdd.mapPartitions(x=>x.map(_*2))// 3.3 打印修改后的RDD中数据rdd1.collect().foreach(println)//4.关闭连接sc.stop()}
}

2.3.1.3 map()和mapPartitions()区别

2.3.1.4 mapPartitionsWithIndex()带分区号

1）函数签名：

​ def mapPartitionsWithIndex[U: ClassTag](

​ f: (Int, Iterator[T]) => Iterator[U], // Int表示分区编号

​ preservesPartitioning: Boolean = false): RDD[U]

2）功能说明：类似于mapPartitions，比mapPartitions多一个整数参数表示分区号

3）需求说明：创建一个RDD，使每个元素跟所在分区号形成一个元组，组成一个新的RDD

代码实现

object value03_mapPartitionsWithIndex {def main(args: Array[String]): Unit = {//1.创建SparkConf并设置App名称val conf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")//2.创建SparkContext，该对象是提交Spark App的入口val sc = new SparkContext(conf)//3具体业务逻辑// 3.1 创建一个RDDval rdd: RDD[Int] = sc.makeRDD(1 to 4, 2)// 3.2 创建一个RDD，使每个元素跟所在分区号形成一个元组，组成一个新的RDDval indexRdd = rdd.mapPartitionsWithIndex( (index,items)=>{items.map( (index,_) )} )// 3.3 打印修改后的RDD中数据indexRdd.collect().foreach(println)//4.关闭连接sc.stop()}
}

2.3.1.5 flatMap()扁平化

1）函数签名：def flatMap[U: ClassTag](f: T => TraversableOnce[U]): RDD[U]

2）功能说明

与map操作类似，将RDD中的每一个元素通过应用f函数依次转换为新的元素，并封装到RDD中。

区别：在flatMap操作中，f函数的返回值是一个集合，并且会将每一个该集合中的元素拆分出来放到新的RDD中。

3）需求说明：创建一个集合，集合里面存储的还是子集合，把所有子集合中数据取出放入到一个大的集合中。

代码实现

object value04_flatMap {def main(args: Array[String]): Unit = {//1.创建SparkConf并设置App名称val conf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")//2.创建SparkContext，该对象是提交Spark App的入口val sc = new SparkContext(conf)//3具体业务逻辑// 3.1 创建一个RDDval listRDD=sc.makeRDD(List(List(1,2),List(3,4),List(5,6),List(7)), 2)// 3.2 把所有子集合中数据取出放入到一个大的集合中listRDD.flatMap(list=>list).collect.foreach(println)//4.关闭连接sc.stop()}
}

2.3.1.6 glom()分区转换数组

1）函数签名：def glom(): RDD[Array[T]]

2）功能说明

​ 该操作将RDD中每一个分区变成一个数组，并放置在新的RDD中，数组中元素的类型与原分区中元素类型一致

3）需求说明：创建一个2个分区的RDD，并将每个分区的数据放到一个数组，求出每个分区的最大值

代码实现

object value05_glom {def main(args: Array[String]): Unit = {//1.创建SparkConf并设置App名称val conf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")//2.创建SparkContext，该对象是提交Spark App的入口val sc = new SparkContext(conf)//3具体业务逻辑// 3.1 创建一个RDDval rdd = sc.makeRDD(1 to 4, 2)// 3.2 求出每个分区的最大值  0->1,2   1->3,4val maxRdd: RDD[Int] = rdd.glom().map(_.max)// 3.3 求出所有分区的最大值的和 2 + 4println(maxRdd.collect().sum)//4.关闭连接sc.stop()}
}

2.3.1.7 groupBy()分组

groupBy算子

代码实现

object value06_groupby {def main(args: Array[String]): Unit = {//1.创建SparkConf并设置App名称val conf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")//2.创建SparkContext，该对象是提交Spark App的入口val sc = new SparkContext(conf)//3具体业务逻辑// 3.1 创建一个RDDval rdd = sc.makeRDD(1 to 4, 2)// 3.2 将每个分区的数据放到一个数组并收集到Driver端打印rdd.groupBy(_ % 2).collect().foreach(println)// 3.3 创建一个RDDval rdd1: RDD[String] = sc.makeRDD(List("hello","hive","hadoop","spark","scala"))// 3.4 按照首字母第一个单词相同分组rdd1.groupBy(str=>str.substring(0,1)).collect().foreach(println)sc.stop()}
}

groupBy会存在shuffle过程
shuffle：将不同的分区数据进行打乱重组的过程
shuffle一定会落盘。可以在local模式下执行程序，通过4040看效果。

2.3.1.8 GroupBy之WordCount

代码实现

object value07_groupby_wordcount {def main(args: Array[String]): Unit = {//1.创建SparkConf并设置App名称val conf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")//2.创建SparkContext，该对象是提交Spark App的入口val sc = new SparkContext(conf)//3具体业务逻辑// 3.1 创建一个RDDval strList: List[String] = List("Hello Scala", "Hello Spark", "Hello World")val rdd = sc.makeRDD(strList)// 3.2 将字符串拆分成一个一个的单词val wordRdd: RDD[String] = rdd.flatMap(str=>str.split(" "))// 3.3 将单词结果进行转换：word=>(word,1)val wordToOneRdd: RDD[(String, Int)] = wordRdd.map(word=>(word, 1))// 3.4 将转换结构后的数据分组val groupRdd: RDD[(String, Iterable[(String, Int)])] = wordToOneRdd.groupBy(t=>t._1)// 3.5 将分组后的数据进行结构的转换val wordToSum: RDD[(String, Int)] = groupRdd.map {case (word, list) => {(word, list.size)}}wordToSum.collect().foreach(println)sc.stop()}
}

2.3.1.9 filter()过滤

1）函数签名： def filter(f: T => Boolean): RDD[T]

2）功能说明

接收一个返回值为布尔类型的函数作为参数。当某个RDD调用filter方法时，会对该RDD中每一个元素应用f函数，如果返回值类型为true，则该元素会被添加到新的RDD中。

3）需求说明：创建一个RDD，过滤出对2取余等于0的数据

代码实现

object value08_filter {def main(args: Array[String]): Unit = {//1.创建SparkConf并设置App名称val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")//2.创建SparkContext，该对象是提交Spark App的入口val sc: SparkContext = new SparkContext(conf)//3.创建一个RDDval rdd: RDD[Int] = sc.makeRDD(Array(1, 2, 3, 4),2)//3.1 过滤出符合条件的数据val filterRdd: RDD[Int] = rdd.filter(_ % 2 == 0)//3.2 收集并打印数据filterRdd.collect().foreach(println)//4 关闭连接sc.stop()}
}

2.3.1.10 sample()采样

1）函数签名：

​ def sample(

​ withReplacement: Boolean,

​ fraction: Double,

​ seed: Long = Utils.random.nextLong): RDD[T]

// withReplacement： true为有放回的抽样，false为无放回的抽样；

// fraction表示：以指定的随机种子随机抽样出数量为fraction的数据；

// seed表示：指定随机数生成器种子。

2）功能说明

从大量的数据中采样

3）需求说明：创建一个RDD（1-10），从中选择放回和不放回抽样

代码实现

object value09_sample {def main(args: Array[String]): Unit = {//1.创建SparkConf并设置App名称val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")//2.创建SparkContext，该对象是提交Spark App的入口val sc: SparkContext = new SparkContext(conf)//3.1 创建一个RDDval dataRDD: RDD[Int] = sc.makeRDD(List(1,2,3,4,5,6))// 抽取数据不放回（伯努利算法）// 伯努利算法：又叫0、1分布。例如扔硬币，要么正面，要么反面。// 具体实现：根据种子和随机算法算出一个数和第二个参数设置几率比较，小于第二个参数要，大于不要// 第一个参数：抽取的数据是否放回，false：不放回// 第二个参数：抽取的几率，范围在[0,1]之间,0：全不取；1：全取；// 第三个参数：随机数种子val sampleRDD: RDD[Int] = dataRDD.sample(false, 0.5)sampleRDD.collect().foreach(println)println("----------------------")// 抽取数据放回（泊松算法）// 第一个参数：抽取的数据是否放回，true：放回；false：不放回// 第二个参数：重复数据的几率，范围大于等于0.表示每一个元素被期望抽取到的次数// 第三个参数：随机数种子val sampleRDD1: RDD[Int] = dataRDD.sample(true, 2)sampleRDD1.collect().foreach(println)//4.关闭连接sc.stop()}
}

java随机数测试

public class TestRandom {public static void main(String[] args) {// 随机算法相同，种子相同，那么随机数就相同//Random r1 = new Random(100);// 不输入参数，种子取的当前时间的纳秒值，所以随机结果就不相同了Random r1 = new Random();for (int i = 0; i < 5; i++) {System.out.println(r1.nextInt(10));}System.out.println("--------------");//Random r2 = new Random(100);Random r2 = new Random();for (int i = 0; i < 5; i++) {System.out.println(r2.nextInt(10));}}
}/**种子相同时的输出结果：
5
0
4
8
1
--------------
5
0
4
8
18*/

2.3.1.11 distinct()去重

distinct算子

代码实现

object value10_distinct {def main(args: Array[String]): Unit = {//1.创建SparkConf并设置App名称val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")//2.创建SparkContext，该对象是提交Spark App的入口val sc: SparkContext = new SparkContext(conf)//3具体业务逻辑// 3.1 创建一个RDDval distinctRdd: RDD[Int] = sc.makeRDD(List(1,2,1,5,2,9,6,1))// 3.2 打印去重后生成的新RDDdistinctRdd.distinct().collect().foreach(println)// 3.3 对RDD采用多个Task去重，提高并发度distinctRdd.distinct(2).collect().foreach(println)//4.关闭连接sc.stop()}
}