Spark---RDD算子(单值类型转换算子)

这篇具有很好参考价值的文章主要介绍了Spark---RDD算子(单值类型转换算子)。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

1.RDD算子介绍

RDD算子是用于对RDD进行转换(Transformation)或行动(Action)操作的方法或函数。通俗来讲,RDD算子就是RDD中的函数或者方法,根据其功能,RDD算子可以分为两大类:
转换算子(Transformation): 转换算子用于从一个RDD生成一个新的RDD,但是原始RDD保持不变。常见的转换算子包括map、filter、flatMap等,它们通过对RDD的每个元素执行相应的操作来生成新的RDD。
行动算子(Action): 行动算子触发对RDD的实际计算,并返回计算结果或将结果写入外部存储系统。与转换算子不同,行动算子会导致Spark作业的执行。如collect方法。

2.转换算子

RDD 根据数据处理方式的不同将算子整体上分为:
Value 类型:对一个RDD进行操作或行动,生成一个新的RDD。
双 Value 类型:对两个RDD进行操作或行动,生成一个新的RDD。
Key-Value类型:对键值对进行操作,如reduceByKey((x, y),按照key对value进行合并。

2.1 Value类型

2.1.1 map

将处理的数据逐条进行映射转换,这里的转换可以是类型的转换,也可以是值的转换。

函数定义
def map[U: ClassTag](f: T => U): RDD[U]

代码实现:

    //建立与Spark框架的连接
    val rdd = new SparkConf().setMaster("local[*]").setAppName("RDD") //配置文件
    val sparkRdd = new SparkContext(rdd) //读取配置文件

    val mapRdd: RDD[Int] = sparkRdd.makeRDD(List(1, 2, 3, 4))
    //对mapRdd进行转换
    val mapRdd1 = mapRdd.map(num => num * 2)
    //对mapRdd1进行转换
    val mapRdd2 = mapRdd1.map(num => num + "->")

    mapRdd2.collect().foreach(print)

    sparkRdd.stop();//关闭连接

Spark---RDD算子(单值类型转换算子),spark,大数据,spark,javascript,服务器

2.1.2 mapPartitions

将待处理的数据以分区为单位发送到计算节点进行处理,这里的处理是指可以进行任意的处理,哪怕是过滤数据。

函数定义
def mapPartitions[U: ClassTag](
f: Iterator[T] => Iterator[U],
preservesPartitioning: Boolean = false): RDD[U]

val rdd = sc.parallelize(Seq(1, 2, 3, 4, 5, 6))  
  
// 使用 mapPartitions 将每个分区中的所有元素增加 1  
val mappedRDD = rdd.mapPartitions(iter => {  
  val increment = 1  
  iter.map(x => x + increment)  
})  
  
mappedRDD.collect().foreach(println)  // 输出: 2, 3, 4, 5, 6, 7

Map 算子是分区内一个数据一个数据的执行,类似于串行操作。而 mapPartitions 算子是以分区为单位进行批处理操作。

mapPartitions在处理数据的时候因为是批处理,相对于map来说处理效率较高,但是如果数据量较大的情况下使用mapPartitions可能会造成内存溢出,因为mapPartitions会将分区内的数据全部加载到内存中。此时更推荐使用map。

2.1.3 mapPartitionsWithIndex

将待处理的数据以分区为单位发送到计算节点进行处理,这里的处理是指可以进行任意的处理,哪怕是过滤数据,在处理时同时可以获取当前分区索引。

函数定义
def mapPartitionsWithIndex[U: ClassTag](
f: (Int, Iterator[T]) => Iterator[U],
preservesPartitioning: Boolean = false): RDD[U]

实现只保留第二个分区的数据

    val mapRdd: RDD[Int] = sparkRdd.makeRDD(List(1, 2, 3, 4),2)
    val newRdd: RDD[Int] = mapRdd.mapPartitionsWithIndex((index, iterator) => {
      if (index == 1) iterator
      else Nil.iterator
    })
    newRdd.collect().foreach(println)
2.1.4 flatMap

将处理的数据进行扁平化后再进行映射处理,所以算子也称之为扁平映射

       //建立与Spark框架的连接
    val rdd = new SparkConf().setMaster("local[*]").setAppName("RDD") //配置文件
    val sparkRdd = new SparkContext(rdd) //读取配置文件

    val rdd1: RDD[List[Int]] = sparkRdd.makeRDD(List(List(1, 2), List(3, 4)))
    val rdd2: RDD[String] = sparkRdd.makeRDD(List("Hello Java", "Hello Scala"), 2)

    val frdd1: RDD[Int] =rdd1.flatMap(list=>{list})
    val frdd2: RDD[String] =rdd2.flatMap(str=>str.split(" "))

    frdd1.collect().foreach(println)
    frdd2.collect().foreach(println)
    sparkRdd.stop();//关闭连接

Spark---RDD算子(单值类型转换算子),spark,大数据,spark,javascript,服务器

2.1.5 glom

将同一个分区的数据直接转换为相同类型的内存数组进行处理,分区不变,glom函数的作用就是将一组数据转换为数组。

函数定义
def glom(): RDD[Array[T]]

    /建立与Spark框架的连接
    val rdd = new SparkConf().setMaster("local[*]").setAppName("RDD") //配置文件
    val sparkRdd = new SparkContext(rdd) //读取配置文件

    val rdd1: RDD[Any] = sparkRdd.makeRDD(List(1,2,3,4),2)
    val value: RDD[Array[Any]] = rdd1.glom()
    value.collect().foreach(data=> println(data.mkString(",")))
    sparkRdd.stop();//关闭连接

Spark---RDD算子(单值类型转换算子),spark,大数据,spark,javascript,服务器

2.1.6 groupBy

将数据根据指定的规则进行分组, 分区默认不变,但是数据会被打乱重新组合,我们将这样的操作称之为 shuffle。 极限情况下,数据可能被分在同一个分区中

函数定义
def groupBy[K](f: T => K)(implicit kt: ClassTag[K]): RDD[(K, Iterable[T])]

	    //按照奇偶分组
    val rdd1: RDD[Int] = sparkRdd.makeRDD(List(1,2,3,4),2)
    val value = rdd1.groupBy(num => num % 2)
    value.collect().foreach(println)
    
    //将 List("Hello", "hive", "hbase", "Hadoop")根据单词首写字母进行分组。
    val rdd2: RDD[String] = sparkRdd.makeRDD(List("Hello", "hive", "hbase", "Hadoop"))
    val value1: RDD[(Char, Iterable[String])] = rdd2.groupBy(str => {
      str.charAt(0)
    })
    value1.collect().foreach(println)

Spark---RDD算子(单值类型转换算子),spark,大数据,spark,javascript,服务器

2.1.7 filter

将数据根据指定的规则进行筛选过滤,符合规则的数据保留,不符合规则的数据丢弃。当数据进行筛选过滤后,分区不变,但是分区内的数据可能不均衡,生产环境下,可能会出现数据倾斜。

函数定义
def filter(f: T => Boolean): RDD[T]

	//获取偶数
    val dataRDD = sparkRdd.makeRDD(List(
      1, 2, 3, 4
    ), 1)
    val value1 = dataRDD.filter(_ % 2 == 0)
2.1.8 sample

函数定义
def sample(
withReplacement: Boolean,
fraction: Double,
seed: Long = Utils.random.nextLong): RDD[T]

根据指定的规则从数据集中抽取数据

参数具体意义:
1.抽取数据不放回
 withReplacement: Boolean, 该参数表示抽取不放回,此时采用伯努利算法(false)
 fraction: Double,该参数表示抽取的几率,范围在[0,1]之间,0:全不取;1:全取;
 seed: Long = Utils.random.nextLong): RDD[T] 该参数表示随机数种子

2.抽取数据放回
 withReplacement: Boolean, 该参数表示抽取放回,此时采用泊松算法(true)
 fraction: Double,该参数表示重复数据的几率,范围大于等于 0.表示每一个元素被期望抽取到的次数
 seed: Long = Utils.random.nextLong): RDD[T] 该参数表示随机数种子
2.1.9 distinct

将数据集中重复的数据去重

def distinct()(implicit ord: Ordering[T] = null): RDD[T]
def distinct(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]

    val dataRDD = sparkRdd.makeRDD(List(
      1, 2, 3, 4, 1, 2
    ), 6)
    val value = dataRDD.distinct()

Spark---RDD算子(单值类型转换算子),spark,大数据,spark,javascript,服务器

2.1.10 coalesce

根据数据量缩减分区,用于大数据集过滤后,提高小数据集的执行效率当 spark 程序中,存在过多的小任务的时候,可以通过 coalesce 方法,收缩合并分区,减少分区的个数,减小任务调度成本

def coalesce(numPartitions: Int, shuffle: Boolean = false,
partitionCoalescer: Option[PartitionCoalescer] = Option.empty)
(implicit ord: Ordering[T] = null)
: RDD[T]

    //初始Rdd采用6个分区
    val dataRDD = sparkRdd.makeRDD(List(
      1, 2, 3, 4, 1, 2
    ), 6)
    //将分区数量缩减至2个
    val value = dataRDD.coalesce(2)

在coalesce中默认不开启shuffle,在进行分区缩减的时候,数据不会被打散。
Spark---RDD算子(单值类型转换算子),spark,大数据,spark,javascript,服务器

2.1.11 repartition

def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]

repartition内部其实执行的是 coalesce 操作,参数 shuffle 的默认值为 true。无论是将分区数多的RDD 转换为分区数少的 RDD,还是将分区数少的 RDD 转换为分区数多的 RDD,repartition操作都可以完成,因为无论如何都会经 shuffle 过程。
Spark---RDD算子(单值类型转换算子),spark,大数据,spark,javascript,服务器

	//将分区数量从2个提升至4个
    val dataRDD = sparkRdd.makeRDD(List(
      1, 2, 3, 4, 1, 2
    ), 2)
    val dataRDD1 = dataRDD.repartition(4)
2.1.12 sortBy

该操作用于排序数据。在排序之前,可以将数据通过 f 函数进行处理,之后按照 f 函数处理的结果进行排序,默认为升序排列。排序后新产生的 RDD 的分区数与原 RDD 的分区数一致。中间存在 shuffle 的过程

def sortBy[K](
f: (T) => K, 该参数表述用于处理的函数
ascending: Boolean = true, 该参数表示是否升序排序
numPartitions: Int = this.partitions.length) 该参数表示设置分区数量
(implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]文章来源地址https://www.toymoban.com/news/detail-812505.html

    val dataRDD = sparkRdd.makeRDD(List(
      1, 2, 3, 4, 1, 2
    ), 2)
    //按照初始数据降序排列
    val dataRDD1 = dataRDD.sortBy(num => num, false, 4)

到了这里,关于Spark---RDD算子(单值类型转换算子)的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Spark【Spark SQL(二)RDD转换DataFrame、Spark SQL读写数据库 】

    Saprk 提供了两种方法来实现从 RDD 转换得到 DataFrame: 利用反射机制推断 RDD 模式 使用编程方式定义 RDD 模式 下面使用到的数据 people.txt :         在利用反射机制推断 RDD 模式的过程时,需要先定义一个 case 类,因为只有 case 类才能被 Spark 隐式地转换为DataFrame对象。 注意

    2024年02月09日
    浏览(38)
  • Spark源码解析(一):RDD之Transfrom算子

    RDD 代表的是分布式数据形态,因此,RDD 到 RDD 之间的转换,本质上是数据形态上的转换(Transformations) 在 RDD 的编程模型中,一共有两种算子,Transformations 类算子和 Actions 类算子。开发者需要使用 Transformations 类算子,定义并描述数据形态的转换过程,然后调用 Actions 类算子

    2024年01月20日
    浏览(32)
  • 2023_Spark_实验十:RDD基础算子操作

    Ø练习 1: Ø 练习 2: Ø 练习 3: Ø 练习 4: Ø 练习 5: groupByKey groupByKey会将RDD[key,value]按照相同的key进行分组,形成RDD[key,iterable[value]]的形式,有点类似于sql中的groupby,例如类似于mysql中的group_contact cogroup groupByKey是对单个RDD的数据进行分组,还可以使用一个叫作cogroup()的函

    2024年02月08日
    浏览(34)
  • 2023_Spark_实验十一:RDD高级算子操作

    coalesce : 总所周知,spark的rdd编程中有两个算子repartition和coalesce。公开的资料上定义为,两者都是对spark分区数进行调整的算子。         repartition会经过shuffle,其实际上就是调用的coalesce(shuffle=true)。         coalesce,默认shuffle=false,不会经过shuffle。         当

    2024年02月08日
    浏览(25)
  • Spark核心RDD详解(设计与运行原理,分区,创建,转换,行动与持久化)

    在实际应用中,存在许多迭代式算法(比如机器学习、图算法等)和交互式数据挖掘工具,这些应用场景的共同之处是,不同计算阶段之间会重用中间结果,即一个阶段的输出结果会作为下一个阶段的输入。但是,目前的MapReduce框架都是把中间结果写入到HDFS中,带来了大量的

    2024年02月04日
    浏览(36)
  • Spark大数据处理讲课笔记---Spark RDD典型案例

    利用RDD计算总分与平均分 利用RDD统计每日新增用户 利用RDD实现分组排行榜 针对成绩表,计算每个学生总分和平均分   读取成绩文件,生成lines;定义二元组成绩列表;遍历lines,填充二元组成绩列表;基于二元组成绩列表创建RDD;对rdd按键归约得到rdd1,计算总分;将rdd1映射

    2024年02月06日
    浏览(37)
  • Spark RDD编程 文件数据读写

    从本地文件系统读取数据,可以采用textFile()方法,可以为textFile()方法提供一个本地文件或目录地址,如果是一个文件地址,它会加载该文件,如果是一个目录地址,它会加载该目录下的所有文件的数据。 示例:读取一个本地文件word.txt val textFile中的textFile是变量名称,sc.t

    2024年02月05日
    浏览(27)
  • 大数据 - Spark系列《六》- RDD详解

    Spark系列文章: 大数据 - Spark系列《一》- 从Hadoop到Spark:大数据计算引擎的演进-CSDN博客 大数据 - Spark系列《二》- 关于Spark在Idea中的一些常用配置-CSDN博客 大数据 - Spark系列《三》- 加载各种数据源创建RDD-CSDN博客 大数据 - Spark系列《四》- Spark分布式运行原理-CSDN博客 大数据

    2024年02月20日
    浏览(32)
  • Spark大数据分析与实战笔记(第三章 Spark RDD 弹性分布式数据集-02)

    人生很长,不必慌张。你未长大,我要担当。 传统的MapReduce虽然具有自动容错、平衡负载和可拓展性的优点,但是其最大缺点是采用非循环式的数据流模型,使得在迭代计算式要进行大量的磁盘IO操作。Spark中的RDD可以很好的解决这一缺点。 RDD是Spark提供的最重要的抽象概念

    2024年02月22日
    浏览(45)
  • 大数据开发之Spark(RDD弹性分布式数据集)

    rdd(resilient distributed dataset)叫做弹性分布式数据集,是spark中最基本的数据抽象。 代码中是一个抽象类,它代表一个弹性的、不可变、可分区、里面的元素可并行计算的集合。 1.1.1 rdd类比工厂生产 1.1.2 wordcount工作流程 1、一组分区(partition),即是数据集的基本组成单位,

    2024年01月24日
    浏览(42)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包