3.2.6 spark体系之分布式计算-spark-core之离线计算-Spark的RDD和算子概念

it2024-12-11 17

1.RDD

1.1 概念

1.2 RDD的五大特性

1.3 RDD的理解图

2.Spark任务执行原理（Standalone集群）

3.Spark代码流程

4.Transformations转换算子

4.1 概念

4.2 Transformation类算子

4.2.1 filter算子的使用

4.2.2 sample算子的使用

5.Action行动算子

5.1 概念

5.2 Action类算子

6.控制算子-RDD的持久化（都是懒执行方式）

6.1 概念

6.2 cache

6.3 persist

6.4 checkpoint

6.4.1 checkpoint 的执行原理：

6.4.2 checkpoint 使用方式

1.RDD

1.1 概念

RDD全称叫做弹性分布式数据集(Resilient Distributed Datasets)，它是一种分布式的内存抽象（计算抽象），表示一个只读的记录分区的集合，它只能通过其他RDD转换而创建，为此，RDD支持丰富的转换操作(如map, join, filter, groupBy等)，通过这种转换操作，新的RDD则包含了如何从其他RDDs衍生所必需的信息，所以说RDDs之间是有依赖关系的。基于RDDs之间的依赖，RDDs会形成一个有向无环图DAG，该DAG描述了整个流式计算的流程，实际执行的时候，RDD是通过血缘关系(Lineage)一气呵成的，即使出现数据分区丢失，也可以通过血缘关系重建分区，总结起来，基于RDD的流式计算任务可描述为：从稳定的物理存储(如分布式文件系统)中加载记录，记录被传入由一组确定性操作构成的DAG，然后写回稳定存储。另外RDD还可以将数据集缓存到内存中，使得在多个操作之间可以重用数据集，基于这个特点可以很方便地构建迭代型应用(图计算、机器学习等)或者交互式数据分析应用。可以说Spark最初也就是实现RDD的一个分布式系统，后面通过不断发展壮大成为现在较为完善的大数据生态系统，简单来讲，Spark-RDD的关系类似于Hadoop-MapReduce关系。

1.2 RDD的五大特性

RDD是由一系列的partition组成的，partition数量由读取文件的block决定。函数（算子）是作用在每一个partition（split）上的。RDD之间有一系列的依赖关系。分区器是作用在K,V格式的RDD上。RDD提供一系列最佳的计算位置。

1.3 RDD的理解图

注意：

sc.textFile方法底层封装的是读取MR读取文件的方式，读取文件之前先split，默认split大小是一个block大小(128M)。RDD实际上不存储数据，这里方便理解，暂时理解为存储数据。什么是K,V格式的RDD? 如果RDD里面存储的数据都是二元组对象，那么这个RDD我们就叫做K,V格式的RDD。哪里体现RDD的弹性（容错）？partition数量可多可少，体现了RDD的弹性。RDD之间依赖关系，可以基于上一个RDD重新计算出RDD。哪里体现RDD的分布式？RDD是由Partition组成，partition是分布在不同节点上的。RDD提供计算最佳位置，体现了数据本地化。体现了大数据中“计算移动数据不移动”的理念。

补充：javaIO

2.Spark任务执行原理（Standalone集群）

以上图中有四个机器节点，Driver和Worker是启动在节点上的进程，运行在JVM中的进程。

Driver与集群节点之间有频繁的通信。Driver负责任务(tasks)的分发和结果的回收。任务的调度。如果task的计算结果非常大就不要回收了。会造成oom。Worker是Standalone资源调度框架里面资源管理的从节点。也是JVM进程。Master是Standalone资源调度框架里面资源管理的主节点。也是JVM进程。

3.Spark代码流程

创建SparkConf对象，可以设置Application name；可以设置运行模式及资源需求。创建SparkContext对象基于Spark的上下文创建一个RDD，对RDD进行处理。应用程序中要有Action类算子来触发Transformation类算子执行。关闭Spark上下文对象SparkContext。

4.Transformations转换算子

4.1 概念

Transformations类算子是一类算子（函数）叫做转换算子，如map,flatMap,reduceByKey等。Transformations算子是延迟执行，也叫懒加载执行，需要行动算子的触发。

4.2 Transformation类算子

filter：过滤符合条件的记录数，true保留，false过滤掉。map：将一个RDD中的每个数据项，通过map中的函数映射变为一个新的元素。特点：输入一条，输出一条数据。flatMap：先map后flat。与map类似，每个输入项可以映射为0到多个输出项。sample：随机抽样算子，根据传进去的小数按比例进行又放回或者无放回的抽样。

reduceByKey：将相同的Key根据相应的逻辑进行处理。

sortByKey/sortBy：作用在K,V格式的RDD上，对key进行升序或者降序排序。

4.2.1 filter算子的使用

package com.bjsxt.spark; import java.util.Arrays; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.FlatMapFunction; import org.apache.spark.api.java.function.Function; import org.apache.spark.api.java.function.Function2; import org.apache.spark.api.java.function.PairFunction; import org.apache.spark.api.java.function.VoidFunction; import scala.Tuple2; public class JavaSparkWordCount2 { public static void main(String[] args) { SparkConf conf = new SparkConf(); conf.setMaster("local"); conf.setAppName("JavaSparkWordCount"); JavaSparkContext sc = new JavaSparkContext(conf); JavaRDD<String> lines = sc.textFile("./words.txt"); // 进行过滤，这个是Transformation类算子，懒加载 // true留下，false的数据会被过滤掉 // new Function<String, Boolean>表示进去一行数据，出去一个bool JavaRDD<String> filter = lines.filter(new Function<String, Boolean>() { private static final long serialVersionUID = 1L; @Override public Boolean call(String line) throws Exception {// 输入是每一行的数据 // 相当于是打印出所有的hello hadoop return line.equals("hello hadoop"); } }); // 遍历，行动算子，可以触发 filter.foreach(new VoidFunction<String>() { private static final long serialVersionUID = 1L; @Override public void call(String s) throws Exception { // TODO Auto-generated method stub System.out.println(s); } }); sc.stop(); } }

其他过滤实例：

我们分析一下上面的代码：

上面的代码是有问题的，当代码执行看到第一个count的时候，count是一个行动算子，往前回溯，找到errors这个RDD，再往前是lines这个RDD，然后按照顺序执行Transformations转换算子。但是当代码执行看到第二个count的时候，还会重复这一个过程，相当于每一次执行job都会向磁盘读取数据，速度是非常慢的。我们的代码对errors进行了重复使用，我们可以将errors这个RDD保存在磁盘上。

4.2.2 sample算子的使用

4.2.3

5.Action行动算子

5.1 概念

Action类算子也是一类算子（函数）叫做行动算子，如foreach,collect，count等。Transformations类算子是延迟执行，Action类算子是触发执行。一个application应用程序中有几个Action类算子执行，就有几个job运行。

5.2 Action类算子

count 返回数据集中的元素数。会在结果计算完成后回收到Driver端。take(n) 返回一个包含数据集前n个元素的集合。firstfirst=take(1), 返回数据集中的第一个元素。foreach 循环遍历数据集中的每个元素，运行相应的逻辑。collect 将计算结果回收到Driver端。

一个完整的Spark程序称为一个application应用程序，application是由job组成的，job就是我们理解的任务，job数量由Action行动算子决定，1个行动算子决定一个job任务。

Transformation类算子都是从RDD到RDD类型，Action类算子都是由RDD到非RDD类型。

package com.bjsxt.spark; import java.util.List; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; public class JavaSparkWordCount3 { public static void main(String[] args) { SparkConf conf = new SparkConf(); conf.setMaster("local"); conf.setAppName("JavaSparkWordCount"); JavaSparkContext sc = new JavaSparkContext(conf); JavaRDD<String> lines = sc.textFile("./words.txt"); // take取出 List<String> take = lines.take(5); for (String s : take) { System.out.println(s); } // first第一行数据 String first_data = lines.first(); System.out.println(first_data); // collect 返回list List<String> collect = lines.collect(); for (String s : collect) { System.out.println(s); } sc.stop(); } }

6.控制算子-RDD的持久化（都是懒执行方式）

6.1 概念

控制算子有三种，cache,persist,checkpoint，以上算子都可以将RDD持久化，三种持久化的单位都是partition。三种算子都是懒执行的。必须有一个action类算子触发执行。checkpoint算子不仅能将RDD持久化到磁盘，还能切断RDD之间的依赖关系。

6.2 cache

默认将RDD的数据持久化到内存中，cache是懒执行。

注意：chche () = persist()=persist(StorageLevel.Memory_Only)

测试cache文件：

6.3 persist

可以指定持久化的级别。最常用的是MEMORY_ONLY和MEMORY_AND_DISK（内存不够了，剩下的放磁盘）。”_2”表示有副本数。

持久化级别如下：

cache和persist的注意事项：

cache和persist都是懒执行，必须有一个action类算子触发执行。cache和persist算子的返回值可以赋值给一个变量，在其他job中直接使用这个变量就是使用持久化的数据了。持久化的单位是partition。cache和persist算子后不能立即紧跟action算子。错误：rdd.cache().count() 返回的不是持久化的RDD，而是一个数值了。

6.4 checkpoint

checkpoint将RDD持久化到磁盘，还可以切断RDD之间的依赖关系，也是一个懒执行方式。

6.4.1 checkpoint 的执行原理：

当RDD的job执行完毕后，会从finalRDD从后往前回溯。当回溯到某一个RDD调用了checkpoint方法，会对当前的RDD做一个标记，继续回溯到源头。Spark框架会自动启动一个新的job，重新计算这个RDD的数据，将数据持久化到HDFS上（checkpoint目录中）。

优化：对RDD执行checkpoint之前，最好对这个RDD先执行cache，这样新启动的job只需要将内存中的数据拷贝到HDFS上就可以，省去了重新计算这一步。

6.4.2 checkpoint 使用方式

SparkConf conf = new SparkConf(); conf.setMaster("local").setAppName("checkpoint"); JavaSparkContext sc = new JavaSparkContext(conf); sc.setCheckpointDir("./checkpoint"); JavaRDD<Integer> parallelize = sc.parallelize(Arrays.asList(1,2,3)); parallelize.checkpoint(); parallelize.count(); sc.stop();

敲代码的乔帮主认证博客专家数据分析大数据工程师 java软件开发走在梦想的大道上，跌跌撞撞，时而悲伤，时而难过失落，是什么又让我扬起头继续向前呢？是心中的那一朵彼岸之花，牵使着你我之间的约定，成为我心中那一股永不可灭的骨气，终究是信仰。

最新回复(0)