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.hqwc.cn/news/95706.html

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

相关文章

大语言模型之七- Llama-2单GPU微调SFT

大语言模型之七- Llama-2单GPU微调SFT

（T4 16G）模型预训练colab脚本在github主页面。详见Finetuning_LLama_2_0_on_Colab_with_1_GPU.ipynb 在上一篇博客提到两种改进预训练模型性能的方法Retrieval-Augmented Generation (RAG) 或者 finetuning。本篇博客过一下模型微调。 微调&#xff1a…
阅读更多...
文件夹中lib，dll含义

文件夹中lib，dll含义

.dll文件是动态链接库（Dynamic Link Library）的缩写，它包含了一组可执行的函数和数据，供程序调用。它可以被多个应用程序共享和重用，减少了代码的冗余。通过动态链接库，可以实现代码的模块化和提高代码的复…
阅读更多...
找到自制电子杂志的方法了，快来看看？

找到自制电子杂志的方法了，快来看看？

终于找到自制电子杂志的方法了，这真是令人兴奋啊！现在，我们可以利用这个方法来创造属于自己的电子杂志，将我们的想法和创意以独特的方式展现给世界。 1.需要一个电子杂志制作工具 市面上有许多专门用于制作电子杂志的工具&#x…
阅读更多...
编译问题：error: ‘printf’ was not declared in this scope

编译问题：error: ‘printf’ was not declared in this scope

这个错误提示意味着编译器在当前作用域内无法找到 printf 函数的声明。这通常是因为没有包含 <stdio.h> 头文件导致的。 解决方法是在程序中添加 #include <stdio.h> 这一行代码。这个头文件中包含了 printf 函数的声明&#xff0c;告诉编译器如何处理该函数。
阅读更多...
【LeetCode-中等题】994. 腐烂的橘子

【LeetCode-中等题】994. 腐烂的橘子

文章目录 题目方法一&#xff1a;bfs层序遍历 题目 该题值推荐用bfs&#xff0c;因为是一层一层的感染&#xff0c;而不是一条线走到底的那种&#xff0c;所以深度优先搜索不适合 方法一&#xff1a;bfs层序遍历 广度优先搜索&#xff0c;就是从起点出发&#xff0c;每次都尝…
阅读更多...
Jmeter如何设置中文版

Jmeter如何设置中文版

第一步&#xff1a;找到 apache-jmeter-5.4.3\bin目录下的 jmeter.properties 第二步:打开 三&#xff0c;ctrf 输入languageen&#xff0c;注释掉&#xff0c;增加以行修改如下 四&#xff0c;ctrs 保存修改内容&#xff0c;重新打开jmeter就可以了
阅读更多...
基于图像切割计算轨迹相似度

基于图像切割计算轨迹相似度

目录 背景思路与核心代码数值实验优缺点分析参考文献 背景 在前面2文&#xff0c;我们分别讨论了利用夹角余弦来计算轨迹相似度和利用缓冲原理来计算轨迹相似度两种方法&#xff0c;前者可以作为一个baseline提供参考&#xff0c;后者的计算更符合人们的感官和事实&#xff0c…
阅读更多...
ARTS打卡第三周之有序链表的合并、gdb中run命令、数制建议、WOOP思维心理学分享

ARTS打卡第三周之有序链表的合并、gdb中run命令、数制建议、WOOP思维心理学分享

Algorithm 题目&#xff1a;两个有序链表的合并 自己的分析见博客《合并两个有序链表》 Review 《run command》是我这周读的英文文章。 在gdb中&#xff0c;run命令在不设置断点的前提下&#xff0c;能够直接把程序运行完成&#xff1b;要是设置断点的话&#xff0c;可以直…
阅读更多...
c# modbus CRC计算器（查表法）

c# modbus CRC计算器（查表法）

一、简介&#xff1a; 本案例为crc计算器&#xff0c;通过查表法计算出结果 1.窗体后台源代码 using Crc; using System; using System.Collections.Generic; using System.ComponentModel; using System.Data; using System.Drawing; using System.Linq; using System.Text…
阅读更多...
项目 - 后端技术栈转型方案

项目 - 后端技术栈转型方案

前言 某开发项目的后端技术栈比较老了&#xff0c;现在想换到新的技术栈上。使用更好的模式、设计思想、更合理的架构等&#xff0c;为未来的需求迭代做铺垫。怎么办呢&#xff1f;假设系统目前在线上运行着的&#xff0c;直接整体换的话耗时太久&#xff0c;且中间还有新的需…
阅读更多...
Qt网络通信——获取本机网络信息

Qt网络通信——获取本机网络信息

查询一个主机的MAC地址或者IP地址是网络应用中常用到的功能&#xff0c;Qt提供了QHostInfo和QNetworkInterface 类可以用于此类信息的查询 1.QHostInfo 类&#xff08;显示和查找本地的信息&#xff09;是的主要函数 类别 函数原型作用公共函数QList <QHostAdress> addr…
阅读更多...
《论文阅读21》Equivariant Multi-View Networks

《论文阅读21》Equivariant Multi-View Networks

一、论文 研究领域&#xff1a;计算机视觉 | 多视角数据处理中实现等变性论文&#xff1a;Equivariant Multi-View Networks ICCV 2019 论文链接视频链接 二、论文简述 在计算机视觉中&#xff0c;模型在不同视角下对数据&#xff08;例如&#xff0c;点云、图像等&#xff0…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023