Spark【RDD编程（二）RDD编程基础】

前言

接上午的那一篇，下午我们学习剩下的RDD编程，RDD操作中的剩下的转换操作和行动操作，最好把剩下的RDD编程都学完。

Spark【RDD编程（一）RDD编程基础】

RDD 转换操作

6、distinct

对 RDD 集合内部的元素进行去重，然后把去重后的其他元素放到一个新的 RDD 集合内。

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object RDDTransForm {def main(args: Array[String]): Unit = {// 创建SparkContext对象val conf = new SparkConf()conf.setAppName("spark core rdd transform").setMaster("local")val sc = new SparkContext(conf)// 通过并行集合创建RDD对象val arr = Array("Spark","Flink","Spark","Storm")val rdd1: RDD[String] = sc.parallelize(arr)val rdd2: RDD[String] = rdd1.distinct()rdd2.foreach(println)//关闭SparkContextsc.stop()}
}

运行输出：

Flink
Spark
Storm

可以看到，重复的元素"Spark"被去除掉。 

7、union

对 两个 RDD 集合进行并集运算，并返回新的 RDD集合，虽然是并集运算，但整个过程不会把重复的元素去除掉。 
// 通过并行集合创建RDD对象val arr1 = Array("Spark","Flink","Storm")val arr2 = Array("Spark","Flink","Hadoop")val rdd1: RDD[String] = sc.parallelize(arr1)val rdd2: RDD[String] = sc.parallelize(arr2)val rdd3: RDD[String] = rdd1.union(rdd2)rdd3.foreach(println)

运行结果：

Spark
Flink
Storm
Spark
Flink
Hadoop
可以看到，重复的元素"Spark"和"Flink"没有被去除。

8、intersection

对两个RDD 集合进行交集运算。

// 通过并行集合创建RDD对象val arr1 = Array("Spark","Flink","Storm")val arr2 = Array("Spark","Flink","Hadoop")val rdd1: RDD[String] = sc.parallelize(arr1)val rdd2: RDD[String] = sc.parallelize(arr2)val rdd3: RDD[String] = rdd1.intersection(rdd2)rdd3.foreach(println)

运行结果：

Spark
Flink

"Spark"和"Flink"是两个RDD集合都有的。 

9、subtract

对两个RDD 集合进行差集运算，并返回新的RDD 集合。

rdd1.substract(rdd2) 返回的是 rdd1有而rdd2中没有的元素，并不会把rdd2中有rdd1中没有的元素也包进来。

// 通过并行集合创建RDD对象val arr1 = Array("Spark","Flink","Storm")val arr2 = Array("Spark","Flink","Hadoop")val rdd1: RDD[String] = sc.parallelize(arr1)val rdd2: RDD[String] = sc.parallelize(arr2)val rdd3: RDD[String] = rdd1.subtract(rdd2)rdd3.foreach(println)

运算结果：

Storm

"Storm"是rdd1中有的二rdd2中没有的，并不会返回"Hadoop"。 

10、zip

把两个 RDD 集合中的元素以键值对的形式进行合并，所以需要确保两个RDD 集合的元素个数必须是相同的。

// 通过并行集合创建RDD对象val arr1 = Array("Spark","Flink","Storm")val arr2 = Array(1,3,5)val rdd1: RDD[String] = sc.parallelize(arr1)val rdd2: RDD[Int] = sc.parallelize(arr2)val rdd3: RDD[(String,Int)] = rdd1.zip(rdd2)rdd3.foreach(println)

运行结果：

(Spark,1)
(Flink,3)
(Storm,5)

RDD 行动操作

RDD 的行动操作是真正触发计算的操作，计算过程十分简单。

1、count

返回 RDD 集合中的元素数量。

2、collect

以数组的形式返回 RDD 集合中所有元素。

3、first

返回 RDD 集合中的第一个元素。

4、take(n)

返回 RDD 集合中前n个元素。

5、reduce(func)

以规则函数func对RDD集合中的元素进行循环处理，比如将所有元素加到一起或乘起来。

6、foreach

对RDD 集合进行遍历，输出RDD集合中所有元素。

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object RDDAction {def main(args: Array[String]): Unit = {// 创建SparkContext对象val conf = new SparkConf()conf.setAppName("spark core rdd transform").setMaster("local")val sc = new SparkContext(conf)//通过并行集合创建 RDD 对象val arr: Array[Int] = Array(1,2,3,4,5)val rdd: RDD[Int] = sc.parallelize(arr)val size: Long = rdd.count()val nums: Array[Int] = rdd.collect()val value: Int = rdd.first()val res: Array[Int] = rdd.take(3)val sum: Int = rdd.reduce((v1, v2) => v1 + v2)println("size = " + size)println("The all elements are ")nums.foreach(println)println("The first element in rdd is " + value)println("The first three elements are ")res.foreach(println)println("sum is " + sum)rdd.foreach(print)//关闭SparkContextsc.stop()}}

运行结果：

size = 5
The all elements are 
1
2
3
4
5
The first element in rdd is 1
The first three elements are 
1
2
3
sum is 15
12345
Process finished with exit code 0

文本长度计算案例

计算 data 目录下的文件字节数（文本总长度）。

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object FileLength {def main(args: Array[String]): Unit = {val conf = new SparkConf()conf.setAppName("spark core rdd transform").setMaster("local")val sc = new SparkContext(conf)val rdd1: RDD[String] = sc.textFile("data")val rdd2: RDD[Int] = rdd1.map(line => line.length)val fileLength: Int = rdd2.reduce((len1, len2) => len1 + len2)println("File length is " + fileLength)sc.stop()}
}

持久化

在Spark 中，RDD采用惰性机制，每次遇到行动操作，就会从头到尾开始执行计算，这对于迭代计算代价是很大的，因为迭代计算经常需要多次重复使用相同的一组数据。

  • 使用cache() 方法将需要持久化的RDD对象持久化进缓存中
  • 使用unpersist() 方法将持久化rdd从缓存中释放出来
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object RDDCache {def main(args: Array[String]): Unit = {val conf = new SparkConf()conf.setAppName("spark core rdd transform").setMaster("local")val sc = new SparkContext(conf)val list = List("Hadoop","Spark","Hive","Flink")val rdd: RDD[String] = sc.parallelize(list)rdd.cache()println(rdd.count())  //第一次行动操作println(rdd.collect.mkString(",")) //第二次行动操作rdd.unpersist() //把这个持久化的rdd从缓存中移除，释放内存空间sc.stop()}
}

分区

分区的作用

        RDD 是弹性分布式数据集，通过 RDD 都很大，会被分成多个分区，分别保存在不同的节点上。进行分区的好处：  

  1. 增加并行度。一个RDD不分区直接进行计算的话，不能充分利用分布式集群的计算优势；如果对RDD集合进行分区，由于一个文件保存在分布式系统中不同的机器节点上，可以就近利用本分区的机器进行计算，从而实现多个分区多节点同时计算，并行度更高。
  2. 减少通信开销。通过数据分区，对于一些特定的操作（如join、reduceByKey、groupByKey、leftOuterJoin等），可以大幅度降低网络传输。

分区的原则

        使分区数量尽量等于集群中CPU核心数目。可以通过设置配置文件中的 spark.default.parallelism 这个参数的值，来配置默认的分区数目。

设置分区的个数 

1、创建 RDD对象时指定分区的数量

1.1、通过本地文件系统或HDFS加载

sc.textFile(path,partitionNum)

1.2、通过并行集合加载 

 对于通过并行集合来创建的RDD 对象，如果没有在参数中指定分区数量，默认分区数目为 min(defaultParallelism,2) ，其中defaultParallelism就是配置文件中的spark.default.parallelism。如果是从HDFS中读取文件，则分区数目为文件分片的数目。

2、使用repartition()方法重新设置分区个数

val rdd2 = rdd1.repartition(1)    //重新设置分区为1

自定义分区函数

继承 org.apache.spark.Partitioner 这个类，并实现下面3个方法：

  1. numPartitions: Int ，用于返回创建出来的分区数。
  2. getPartition(key: Any)，用于返回给定键的分区编号(0~paratitionNum-1)。
  3. equals()，Java中判断相等想的标准方法。

注意：Spark 的分区函数针对的是（key，value）类型的RDD，也就是说，RDD中的每个元素都是（key,value）类型的，然后函数根据 key 对RDD 元素进行分区。所以，当要对一些非（key，value）类型的 RDD 进行自定义分区时，需要首先把 RDD 元素转换为（key，value）类型，然后再使用分区函数。

案例

将奇数和偶数分开写到不同的文件中去。

import org.apache.spark.rdd.RDD
import org.apache.spark.{Partitioner, SparkConf, SparkContext}class MyPartitioner(numParts: Int = 2) extends Partitioner{//覆盖默认的分区数目override def numPartitions: Int = numParts//覆盖默认的分区规则override def getPartition(key: Any): Int = {if (key.toString.toInt%2==0) 1 else 0}
}
object MyPartitioner{def main(args: Array[String]): Unit = {val conf = new SparkConf()conf.setAppName("partitioner").setMaster("local")val sc: SparkContext = new SparkContext(conf)val data: Array[Int] = (1 to 100).toArrayval rdd: RDD[Int] = sc.parallelize(data,5)val savePath:String = System.getProperty("user.dir")+"/data/rdd/out"rdd.map((_,1)).partitionBy(new MyPartitioner()).map(_._1).saveAsTextFile(savePath)sc.stop()}
}

我们在代码中创建RDD 对象的时候，我们指定了分区默认的数量为 5，然后我们使用我们自定义的分区，观察会不会覆盖掉默认的分区数量： 

运行结果：

我们可以看到，除了校验文件，一共生成了两个文件，其中一个保存了1~100的所有奇数，一个保存了1~100的所有偶数； 

综合案例

在上一篇博客中，我们已经做过WordCount了，但是明显篇幅比较长，这里我们简化后只需要两行代码：

    //使用本地文件作为数据加载创建RDD 对象val rdd: RDD[String] = sc.textFile("data/word.txt")//RDD("Hadoop is good","Spark is better","Spark is fast")val res_rdd: RDD[(String,Int)] = rdd.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)//flatMap://RDD(Array("Hadoop is good"),Array("Spark is better"),Array("Spark is fast"))//RDD("Hadoop","is",good","Spark","is","better","Spark","is","fast"))

运行结果：

(Spark,2)
(is,3)
(fast,1)
(good,1)
(better,1)
(Hadoop,1)

总结

至此，我们RDD基础编程部分就结束了，但是RDD编程还没有结束，接下来我会继续学习键值对RDD、数据读写，最后总结性低做一个大的综合案例。

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

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

相关文章

C语言：截断+整型提升+算数转换练习

C语言：截断+整型提升+算数转换练习

详情关于整型提升、算数转换与截断见文章： 《C语言：整型提升》 《C语言：算数转换》 一、代码一 int main() { char a -1; signed char b -1; unsigned char c -1; printf("%d %d %d", a, b, c); return 0; } 求…
阅读更多...
ToBeWritten之VSOC安全运营

ToBeWritten之VSOC安全运营

也许每个人出生的时候都以为这世界都是为他一个人而存在的，当他发现自己错的时候，他便开始长大 少走了弯路，也就错过了风景，无论如何，感谢经历 转移发布平台通知：将不再在CSDN博客发布新文章，敬…
阅读更多...
华为OD：敏感字段加密

华为OD：敏感字段加密

题目描述： 给定一个由多个命令字组成的命令字符串： 1、字符串长度小于等于127字节,只包含大小写字母,数字,下划线和偶数个双引号； 2、命令字之间以一个或多个下划线_进行分割； 3、可以通过两个双引号”"来标识包含下划线…
阅读更多...
【实战】十一、看板页面及任务组页面开发（六） —— React17+React Hook+TS4 最佳实践，仿 Jira 企业级项目（二十八）

【实战】十一、看板页面及任务组页面开发（六） —— React17+React Hook+TS4 最佳实践，仿 Jira 企业级项目（二十八）

文章目录 一、项目起航：项目初始化与配置二、React 与 Hook 应用：实现项目列表三、TS 应用：JS神助攻 - 强类型四、JWT、用户认证与异步请求五、CSS 其实很简单 - 用 CSS-in-JS 添加样式六、用户体验优化 - 加载中和错误状态处理七、Hook&…
阅读更多...
pytest笔记： pytest单元测试框架

pytest笔记： pytest单元测试框架

第一步：安装 和查看版本 pycharm settings 查看 第二步： 编写test_example.py def inc(x):return x1 def test_answer():assert inc(4) 5 第三步：在当前路径下执行pytest 命令 PS E:\data\web测试\Selenium3自动化测试实战——基于Pyth…
阅读更多...
应用程序管理工具

应用程序管理工具

应用程序管理是 DevOps 的重要组成部分。它可以定义为在所有阶段监控和管理软件应用程序的可用性、运行状况、性能和功能的过程，包括规划、设计、构建、测试、部署、维护和更新。这意味着应用程序从概念到停止都受到监控。 应用程序管理的重要性 管理应用程序可确…
阅读更多...
db2迁移至oracle

db2迁移至oracle

1.思路 （1）用java连接数据库（2）把DB2数据导出为通用的格式如csv，json等（3）导入其他数据库，比如oracle，mongodb。这个方法自由发挥的空间比较大。朋友说他会用springboot…
阅读更多...
《2023年网信人才培训-网络安全从业人员能力素养提升培训》第一期成功举办

《2023年网信人才培训-网络安全从业人员能力素养提升培训》第一期成功举办

随着网络强国和数字中国建设的步伐加快，建设规模宏大、结构合理、素质优良的人才队伍成为一项重要工作。知了汇智作为数字产教融合基地，通过与高校、企业等多方合作，建立了完整的网络安全人才培养生态链。凭借自身技术优势和丰富的产业资源&a…
阅读更多...
新建工程——第一个S32DS工程

新建工程——第一个S32DS工程

之前的"测试开发板"章节 测试开发板——第一个AutoSAR程序，使用了一个 demo 工程，不管是裸机程序还是 AutoSAR 程序，那都是别人已经创建好的工程。本节来介绍如何来创建自己的工程，本节介绍如何创建一个 S32DS 的工程，点亮开发板上的 LED 我们从官方提供的例程…
阅读更多...
为什么删除Windows 11上的Bloatware可以帮助加快你的电脑速度

为什么删除Windows 11上的Bloatware可以帮助加快你的电脑速度

如果你感觉你的电脑迟钝，彻底清除软件会有所帮助，而且这个过程对Windows用户来说越来越容易。 微软正在使删除以前难以删除的其他预装Windows应用程序成为可能。专家表示，这项新功能可能会改变用户的游戏规则。 科技公司Infatica的主管Vlad…
阅读更多...
测试人:“躺平?不可能的“, 盘点测试人在职场的优势

测试人:“躺平?不可能的“, 盘点测试人在职场的优势

之前有这么一个段子：有人喜欢创造世界，他们做了程序员；有人喜欢拯救世界，他们做了测试员！近几年，测试工程师在企业究竟是怎么样的发展？随着企业对于用户体验的满意度越来越重视，更加…
阅读更多...
两个线程同步执行：解决乱箭穿心（STL/Windows/Linux）

两个线程同步执行：解决乱箭穿心（STL/Windows/Linux）

C自学精简教程 目录(必读) C并发编程入门 目录 多线程同步 线程之间同步是指线程等待其他线程执行完某个动作之后再执行（本文情况）。 线程同步还可以是像十字路口的红绿灯一样，只允许一个方向的车同行，其他方向的车等待。 本…
阅读更多...
C#，《小白学程序》第八课：列表（List）应用之二“编制高铁列车时刻表”

C#，《小白学程序》第八课：列表（List）应用之二“编制高铁列车时刻表”

1 文本格式 /// <summary> /// 车站信息类 class /// </summary> public class Station { /// <summary> /// 编号 /// </summary> public int Id { get; set; } 0; /// <summary> /// 车站名 /// </summary&g…
阅读更多...
JavaScript设计模式（四）——策略模式、代理模式、观察者模式

JavaScript设计模式（四）——策略模式、代理模式、观察者模式

个人简介 &#x1f440;个人主页&#xff1a; 前端杂货铺 &#x1f64b;‍♂️学习方向&#xff1a; 主攻前端方向&#xff0c;正逐渐往全干发展 &#x1f4c3;个人状态&#xff1a; 研发工程师&#xff0c;现效力于中国工业软件事业 &#x1f680;人生格言&#xff1a; 积跬步…
阅读更多...
Python3 条件控制

Python3 条件控制

Python3 条件控制 Python 条件语句是通过一条或多条语句的执行结果&#xff08;True 或者 False&#xff09;来决定执行的代码块。 可以通过下图来简单了解条件语句的执行过程: 代码执行过程&#xff1a; if 语句 Python中if语句的一般形式如下所示&#xff1a; if conditi…
阅读更多...
Spring Cloud Foundry上使用通配符模式匹配进行的安全绕过漏洞 CVE-2023-20873

Spring Cloud Foundry上使用通配符模式匹配进行的安全绕过漏洞 CVE-2023-20873

文章目录 0.前言1.参考文档2.基础介绍描述如果满足以下任一条件&#xff0c;应用程序就不会有太大风险&#xff1a;受影响的Spring产品和版本 3.解决方案3.1. 升级版本3.2. 替代方案 0.前言 背景&#xff1a;公司项目扫描到 Spring Cloud Foundry上使用通配符模式匹配进行的安全…
阅读更多...
将符号分隔的文本文件txt转换为excel的实现

将符号分隔的文本文件txt转换为excel的实现

文本文件如下&#xff1a; 现在不好处理&#xff0c;打算将其转换为excel&#xff0c;其中通过冒号分割&#xff1a;line.split(":") main方法如下&#xff1a; public static void main(String[] args) {String textFilePath "D:\\zoom\\期刊\\J_Medline\\J_…
阅读更多...
肖sir__linux详解__001

肖sir__linux详解__001

linux详解: 1、ifconfig 查看ip地址 2、6版本&#xff1a;防火墙的命令&#xff1a; service iptables status 查看防火墙状态 service iptables statrt 开启防火墙 service iptables stop 关闭防火墙 service iptables restart 重启防火墙状态 7版本&#xff1a; systemctl s…
阅读更多...
【Golang】函数篇

【Golang】函数篇

1、golang函数基本定义与使用 func 函数名 (形参列表) (返回值类型列表) {函数体return 返回值列表 }其中func用于表明这是一个函数&#xff0c;剩下的东西与其他语言的函数基本一致&#xff0c;在定义与使用的时候注意函数名、参数、返回值书写的位置即可。下面使用一个例子…
阅读更多...
【超详细~KVM】KVM概述、安装及简单操作-------从小白到大神之路之学习运维第91天

【超详细~KVM】KVM概述、安装及简单操作-------从小白到大神之路之学习运维第91天

第四阶段提升 时 间&#xff1a;2023年8月30日 参加人&#xff1a;全班人员 内 容&#xff1a; KVM概述、安装及简单操作 目录 一、KVM 概述 二、KVM工作原理 三、KVM应用场景 四、centos7 下安装部署 五、新建虚拟机步骤 1、创建存储池并创建存储卷 2、点击号创建…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023