Spark---转换算子、行动算子、持久化算子

一、转换算子和行动算子

1、Transformations转换算子

1)、概念

Transformations类算子是一类算子(函数)叫做转换算子,如map、flatMap、reduceByKey等。Transformations算子是延迟执行,也叫懒加载执行。

2)、Transformation类算子

filter :过滤符合条件的记录数,true保留,false过滤掉

map:将一个RDD中的每个数据项,通过map中的函数映射变为一个新的元素。特点:输入一条,输出一条数据。

flatMap:先map后flat。与map类似,每个输入项可以映射为0到多个输出项。

sample:随机抽样算子,根据传进去的小数按比例进行又放回或者无放回的抽样。

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

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

2、Action行动算子

1)、概念:

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

2)、Action类算子

count:返回数据集中的元素数。会在结果计算完成后回收到Driver端。

take(n):返回一个包含数据集前n个元素的集合。

first:first=take(1),返回数据集中的第一个元素。

foreach:循环遍历数据集中的每个元素,运行相应的逻辑。

collect:将计算结果回收到Driver端。

3)、demo:动态统计出现次数最多的单词个数,过滤掉。

  • 一千万条数据量的文件,过滤掉出现次数多的记录,并且其余记录按照出现次数降序排序。

假设有一个records.txt文件

hello Spark
hello HDFS
hello hadoop
hello linux
hello Spark
hello Spark
hello Spark1
hello Spark
hello Spark
hello Spark2
hello Spark
hello Spark
hello Spark
hello Spark3
hello Spark
hello HDFS
hello hadoop
hello linux
hello Spark
hello Spark
hello Spark4
hello Spark
hello Spark
hello Spark5
hello Spark
hello Spark

代码处理:

package com.bjsxt.demo;import java.util.Arrays;
import java.util.List;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;
/*** 动态统计出现次数最多的单词个数,过滤掉。* @author root**/
public class Demo1 {public static void main(String[] args) {SparkConf conf = new SparkConf();conf.setMaster("local").setAppName("demo1");JavaSparkContext jsc = new JavaSparkContext(conf);JavaRDD<String> lines = jsc.textFile("./records.txt");JavaRDD<String> flatMap = lines.flatMap(new FlatMapFunction<String, String>() {/*** */private static final long serialVersionUID = 1L;@Overridepublic Iterable<String> call(String t) throws Exception {return Arrays.asList(t.split(" "));}});JavaPairRDD<String, Integer> mapToPair = flatMap.mapToPair(new PairFunction<String,String, Integer>() {/*** */private static final long serialVersionUID = 1L;@Overridepublic Tuple2<String, Integer> call(String t) throws Exception {return new Tuple2<String, Integer>(t, 1);}});JavaPairRDD<String, Integer> sample = mapToPair.sample(true, 0.5);final List<Tuple2<String, Integer>> take = sample.reduceByKey(new Function2<Integer,Integer,Integer>(){/*** */private static final long serialVersionUID = 1L;@Overridepublic Integer call(Integer v1, Integer v2) throws Exception {return v1+v2;}}).mapToPair(new PairFunction<Tuple2<String,Integer>, Integer, String>() {/*** */private static final long serialVersionUID = 1L;@Overridepublic Tuple2<Integer, String> call(Tuple2<String, Integer> t)throws Exception {return new Tuple2<Integer, String>(t._2, t._1);}}).sortByKey(false).mapToPair(new PairFunction<Tuple2<Integer,String>, String, Integer>() {/*** */private static final long serialVersionUID = 1L;@Overridepublic Tuple2<String, Integer> call(Tuple2<Integer, String> t)throws Exception {return new Tuple2<String, Integer>(t._2, t._1);}}).take(1);System.out.println("take--------"+take);JavaPairRDD<String, Integer> result = mapToPair.filter(new Function<Tuple2<String,Integer>, Boolean>() {/*** */private static final long serialVersionUID = 1L;@Overridepublic Boolean call(Tuple2<String, Integer> v1) throws Exception {return !v1._1.equals(take.get(0)._1);}}).reduceByKey(new Function2<Integer,Integer,Integer>(){/*** */private static final long serialVersionUID = 1L;@Overridepublic Integer call(Integer v1, Integer v2) throws Exception {return v1+v2;}}).mapToPair(new PairFunction<Tuple2<String,Integer>, Integer, String>() {/*** */private static final long serialVersionUID = 1L;@Overridepublic Tuple2<Integer, String> call(Tuple2<String, Integer> t)throws Exception {return new Tuple2<Integer, String>(t._2, t._1);}}).sortByKey(false).mapToPair(new PairFunction<Tuple2<Integer,String>, String, Integer>() {/*** */private static final long serialVersionUID = 1L;@Overridepublic Tuple2<String, Integer> call(Tuple2<Integer, String> t)throws Exception {return new Tuple2<String, Integer>(t._2, t._1);}});result.foreach(new VoidFunction<Tuple2<String,Integer>>() {/*** */private static final long serialVersionUID = 1L;@Overridepublic void call(Tuple2<String, Integer> t) throws Exception {System.out.println(t);}});jsc.stop();}
}

3、Spark代码流程

1)、创建SparkConf对象

可以设置Application name。

可以设置运行模式。

可以设置Spark application的资源需求。

2)、创建SparkContext对象

3)、基于Spark的上下文创建一个RDD,对RDD进行处理。

4)、应用程序中要有Action类算子来触发Transformation类算子执行。

5)、关闭Spark上下文对象SparkContext。

二、Spark持久化算子

1、控制算子

1)、概念

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

2)、cache

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

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

测试cache文件:

测试代码:

1.SparkConf conf = new SparkConf();
2.conf.setMaster("local").setAppName("CacheTest");
3.JavaSparkContext jsc = new JavaSparkContext(conf);
4.JavaRDD<String> lines = jsc.textFile("persistData.txt");
5.
6.lines = lines.cache();
7.long startTime = System.currentTimeMillis();
8.long count = lines.count();
9.long endTime = System.currentTimeMillis();
10.System.out.println("共"+count+ "条数据,"+"初始化时间+cache时间+计算时间="+ 
11.(endTime-startTime));
12.
13.long countStartTime = System.currentTimeMillis();
14.long countrResult = lines.count();
15.long countEndTime = System.currentTimeMillis();
16.System.out.println("共"+countrResult+ "条数据,"+"计算时间="+ (countEndTime-
17.countStartTime));
18.
19.jsc.stop();

persist:

可以指定持久化的级别。最常用的是MEMORY_ONLY和MEMORY_AND_DISK。”_2“表示有副本数。

持久化级别如下:

2、cache和persist的注意事项

1)、cache和persist都是懒执行,必须有一个action类算子触发执行。

2)、cache和persist算子的返回值可以赋值给一个变量,在其他job中直接使用这个变量就是使用持久化的数据了。持久化的单位是partition。

3)、cache和persist算子后不能立即紧跟action算子。

4)、cache和persist算子持久化的数据当applilcation执行完成之后会被清除。

错误:rdd.cache().count() 返回的不是持久化的RDD,而是一个数值了。

3、checkpoint

checkpoint将RDD持久化到磁盘,还可以切断RDD之间的依赖关系。checkpoint目录数据当application执行完之后不会被清除。
  • persist(StorageLevel.DISK_ONLY)与Checkpoint的区别?

1)、checkpoint需要指定额外的目录存储数据,checkpoint数据是由外部的存储系统管理,不是Spark框架管理,当application完成之后,不会被清空。

2)、cache() 和persist() 持久化的数据是由Spark框架管理,当application完成之后,会被清空。

3)、checkpoint多用于保存状态。

  • checkpoint 的执行原理:

1)、当RDD的job执行完毕后,会从finalRDD从后往前回溯。

2)、当回溯到某一个RDD调用了checkpoint方法,会对当前的RDD做一个标记。

3)、Spark框架会自动启动一个新的job,重新计算这个RDD的数据,将数据持久化到HDFS上。

  • 优化:对RDD执行checkpoint之前,最好对这个RDD先执行cache,这样新启动的job只需要将内存中的数据拷贝到HDFS上就可以,省去了重新计算这一步。
  • 使用:
1.SparkConf conf = new SparkConf();
2.conf.setMaster("local").setAppName("checkpoint");
3.JavaSparkContext sc = new JavaSparkContext(conf);
4.sc.setCheckpointDir("./checkpoint");
5.JavaRDD<Integer> parallelize = sc.parallelize(Arrays.asList(1,2,3));
6.parallelize.checkpoint();
7.parallelize.count();
8.sc.stop();

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.hqwc.cn/news/205544.html

如若内容造成侵权/违法违规/事实不符,请联系编程知识网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

Java Stream中的API你都用过了吗?

公众号「架构成长指南」&#xff0c;专注于生产实践、云原生、分布式系统、大数据技术分享。 在本教程中&#xff0c;您将通过大量示例来学习 Java 8 Stream API。 Java 在 Java 8 中提供了一个新的附加包&#xff0c;称为 java.util.stream。该包由类、接口和枚举组成&#x…

Maven中常用命令以及idea中使用maven指南

文章目录 Maven 常用命令compiletestcleanpackageinstallMaven 指令的生命周期maven 的概念模型 idea 开发maven 项目idea 的maven 配置idea 中创建一个maven 的web 工程在pom.xml 文件添加坐标坐标的来源方式依赖范围编写servlet maven 工程运行调试 Maven 常用命令 compile …

腾讯云服务器99元一年?假的,阿里云是99元

腾讯云服务器99元一年是真的吗&#xff1f;假的&#xff0c;不用99元&#xff0c;只要88元即可购买一台2核2G3M带宽的轻量应用服务器&#xff0c;99元太多了&#xff0c;88元就够了&#xff0c;腾讯云百科活动 txybk.com/go/txy 活动打开如下图&#xff1a; 腾讯云服务器价格 腾…

C#中的var究竟是强类型还是弱类型?

前言 在C#中&#xff0c;var关键字是用来声明变量类型的&#xff0c;它是C# 3.0推出的新特征&#xff0c;它允许编译器根据初始化表达式推断变量类型&#xff0c;有点跟javascript类似&#xff0c;而javascript中的var是弱类型。它让C#变量声明更加简洁&#xff0c;但也导致了…

【Pytorch】Visualization of Fature Maps(2)

学习参考来自 使用CNN在MNIST上实现简单的攻击样本https://github.com/wmn7/ML_Practice/blob/master/2019_06_03/CNN_MNIST%E5%8F%AF%E8%A7%86%E5%8C%96.ipynb 文章目录 在 MNIST 上实现简单的攻击样本1 训练一个数字分类网络2 控制输出的概率, 看输入是什么3 让正确的图片分…

麻雀搜索优化算法MATLAB实现,SSA-BP网络

对于麻雀搜索算法的介绍&#xff0c;网上已经有不少资料了&#xff0c;这边公布SSA的matlab实现 下面展示SSA算法的核心代码以及详细注解 % 麻雀搜索算法函数定义 % 输入&#xff1a;种群大小(pop)&#xff0c;最大迭代次数(Max_iter)&#xff0c;搜索空间下界(lb)&#xff0c…

CSS实现空心的“尖角”

大家好&#xff0c;我是南宫&#xff0c;来分享一个昨天解决的问题。 我记得之前刷面试题的时候&#xff0c;CSS面试题里面赫然有一题是“如何用CSS实现三角形”&#xff0c;我觉得这个问题确实很经典&#xff0c;我上的前端培训班当初就讲过。 大概思路如下&#xff1a; 先…

想分析全国用电及煤气、液化石油气供应利用情况,这部分数据对你有帮助!

随着经济的发展和人民生活水平的提高&#xff0c;能源的需求量越来越大。其中&#xff0c;电力和煤气、液化石油气等能源的供应利用情况与我们的日常生活息息相关。 今天我们根据《中国城市统计年鉴》统计的中国地级及以上城市的煤气及液化石油气供应及利用情况的指标&#xff…

洗内裤的小洗衣机买啥牌子的?性价比婴儿洗衣机推荐

在近些年来&#xff0c;人们对生活和健康的要求越来越高&#xff0c;所以内衣洗衣机也走进了人们的视线&#xff0c;许多研究显示&#xff0c;单纯的手洗是不能彻底消除体内的细菌的&#xff0c;而机洗则可以有效地消除大部分的细菌&#xff0c;但是机洗内衣裤对洗衣机的卫生要…

MySQL数据库入门到大牛_基础_11_数据处理之增删改

本章将会介绍DML中的增删改查操作&#xff0c;增删改泛泛来讲是针对表中数据的修改。 文章目录 1. 插入数据1.1 实际问题1.2 方式1&#xff1a;VALUES的方式添加1.3 方式2&#xff1a;将查询结果插入到表中 2. 更新数据3. 删除数据4. 小结5. MySQL8新特性&#xff1a;计算列6. …

如何正确复制CSDN文章到自己的博客

1.csdn 文章页面&#xff0c;按f12打开浏览器开发者工具 2.按ctrl f 找 "article_content" 3.在该元素源代码上右键 “Copy”->“Copy element” 4.新建一个txt文件,把你粘贴的东西复制进去,然后再把文件名的后缀改为html,然后打开html文件,把里面的内容ctrlA全部…