[Spark精进]必须掌握的4个RDD算子之map算子

简介: [Spark精进]必须掌握的4个RDD算子之map算子

序章

第一个map. 以元素为粒度的数据转换

我们先来说说 map 算子的用法:给定映射函数 f,map(f) 以元素为粒度对 RDD 做数据转换。其中 f 可以是带有明确签名的带名函数,也可以是匿名函数,它的形参类型必须与 RDD 的元素类型保持一致,而输出类型则任由开发者自行决定。

我们使用如下代码,把包含单词的 RDD 转换成元素为(Key,Value)对的 RDD,后者统称为 Paired RDD。

// 把普通RDD转换为Paired RDD
val cleanWordRDD: RDD[String] = _ 
val kvRDD: RDD[(String, Int)] = cleanWordRDD.map(word => (word, 1))

在上面的代码实现中,传递给 map 算子的形参,即:word => (word,1),就是我们上面说的映射函数 f。只不过,这里 f 是以匿名函数的方式进行定义的,其中左侧的 word 表示匿名函数 f 的输入形参,而右侧的(word,1)则代表函数 f 的输出结果。

如果我们把匿名函数变成带名函数的话,可能你会看的更清楚一些。这里我用一段代码重新定义了带名函数 f。

// 把RDD元素转换为(Key,Value)的形式
// 定义映射函数f
def f(word: String): (String, Int) = {
return (word, 1)
}
val cleanWordRDD: RDD[String] = _ 
val kvRDD: RDD[(String, Int)] = cleanWordRDD.map(f)

可以看到,我们使用 Scala 的 def 语法,明确定义了带名映射函数 f,它的计算逻辑与刚刚的匿名函数是一致的。在做 RDD 数据转换的时候,我们只需把函数 f 传递给 map 算子即可。不管 f 是匿名函数,还是带名函数,map 算子的转换逻辑都是一样的,你不妨把以上两种实现方式分别敲入到 spark-shell,去验证执行结果的一致性。

到这里为止,我们就掌握了 map 算子的基本用法。现在你就可以定义任意复杂的映射函数 f,然后在 RDD 之上通过调用 map(f) 去翻着花样地做各种各样的数据转换。

比如,通过定义如下的映射函数 f,我们就可以改写 Word Count 的计数逻辑,也就是把“Spark”这个单词的统计计数权重提高一倍:

// 把RDD元素转换为(Key,Value)的形式
// 定义映射函数f
def f(word: String): (String, Int) = {
if (word.equals("Spark")) { return (word, 2) }
return (word, 1)
}
val cleanWordRDD: RDD[String] = _ // 请参考第一讲获取完整代码
val kvRDD: RDD[(String, Int)] = cleanWordRDD.map(f)

尽管 map 算子足够灵活,允许开发者自由定义转换逻辑。不过,就像我们刚刚说的,map(f) 是以元素为粒度对 RDD 做数据转换的,在某些计算场景下,这个特点会严重影响执行效率。为什么这么说呢?我们来看一个具体的例子。

比方说,我们把 Word Count 的计数需求,从原来的对单词计数,改为对单词的哈希值计数,在这种情况下,我们的代码实现需要做哪些改动呢?我来示范一下:

// 把普通RDD转换为Paired RDD
import java.security.MessageDigest
val cleanWordRDD: RDD[String] = _ // 请参考第一讲获取完整代码
val kvRDD: RDD[(String, Int)] = cleanWordRDD.map{ word =>
  // 获取MD5对象实例
  val md5 = MessageDigest.getInstance("MD5")
  // 使用MD5计算哈希值
  val hash = md5.digest(word.getBytes).mkString
  // 返回哈希值与数字1的Pair
  (hash, 1)
}

由于 map(f) 是以元素为单元做转换的,那么对于 RDD 中的每一条数据记录,我们都需要实例化一个 MessageDigest 对象来计算这个元素的哈希值。

在工业级生产系统中,一个 RDD 动辄包含上百万甚至是上亿级别的数据记录,如果处理每条记录都需要事先创建 MessageDigest,那么实例化对象的开销就会聚沙成塔,不知不觉地成为影响执行效率的罪魁祸首。

那么问题来了,有没有什么办法,能够让 Spark 在更粗的数据粒度上去处理数据呢?还真有,mapPartitions 和 mapPartitionsWithIndex 这对“孪生兄弟”就是用来解决类似的问题。相比 mapPartitions,mapPartitionsWithIndex 仅仅多出了一个数据分区索引,因此接下来我们把重点放在 mapPartitions 上面。

点击跳转到下一讲


目录
打赏
0
0
0
0
111
分享
相关文章
【赵渝强老师】Spark RDD的缓存机制
Spark RDD通过`persist`或`cache`方法可将计算结果缓存,但并非立即生效,而是在触发action时才缓存到内存中供重用。`cache`方法实际调用了`persist(StorageLevel.MEMORY_ONLY)`。RDD缓存可能因内存不足被删除,建议结合检查点机制保证容错。示例中,读取大文件并多次调用`count`,使用缓存后执行效率显著提升,最后一次计算仅耗时98ms。
【赵渝强老师】Spark RDD的缓存机制
【赵渝强老师】Spark中的RDD
RDD(弹性分布式数据集)是Spark的核心数据模型,支持分布式并行计算。RDD由分区组成,每个分区由Spark Worker节点处理,具备自动容错、位置感知调度和缓存机制等特性。通过创建RDD,可以指定分区数量,并实现计算函数、依赖关系、分区器和优先位置列表等功能。视频讲解和示例代码进一步详细介绍了RDD的组成和特性。
【赵渝强老师】Spark RDD的依赖关系和任务阶段
Spark RDD之间的依赖关系分为窄依赖和宽依赖。窄依赖指父RDD的每个分区最多被一个子RDD分区使用,如map、filter操作;宽依赖则指父RDD的每个分区被多个子RDD分区使用,如分组和某些join操作。窄依赖任务可在同一阶段完成,而宽依赖因Shuffle的存在需划分不同阶段执行。借助Spark Web Console可查看任务的DAG图及阶段划分。
60 15
大数据-99 Spark 集群 Spark Streaming DStream 文件数据流、Socket、RDD队列流
大数据-99 Spark 集群 Spark Streaming DStream 文件数据流、Socket、RDD队列流
55 0
大数据-94 Spark 集群 SQL DataFrame & DataSet & RDD 创建与相互转换 SparkSQL
大数据-94 Spark 集群 SQL DataFrame & DataSet & RDD 创建与相互转换 SparkSQL
139 0
|
8月前
|
Dart之集合详解(List、Set、Map)
Dart之集合详解(List、Set、Map)
ES6的Set和Map你都知道吗?一文了解集合和字典在前端中的应用
该文章详细介绍了ES6中Set和Map数据结构的特性和使用方法,并探讨了它们在前端开发中的具体应用,包括如何利用这些数据结构来解决常见的编程问题。
ES6的Set和Map你都知道吗?一文了解集合和字典在前端中的应用
java集合框架复习----(4)Map、List、set
这篇文章是Java集合框架的复习总结,重点介绍了Map集合的特点和HashMap的使用,以及Collections工具类的使用示例,同时回顾了List、Set和Map集合的概念和特点,以及Collection工具类的作用。
java集合框架复习----(4)Map、List、set
【Java集合类面试二十二】、Map和Set有什么区别?
该CSDN博客文章讨论了Map和Set的区别,但提供的内容摘要并未直接解释这两种集合类型的差异。通常,Map是一种键值对集合,提供通过键快速检索值的能力,而Set是一个不允许重复元素的集合。
AI助理

你好,我是AI助理

可以解答问题、推荐解决方案等