Spark基础学习笔记17：掌握RDD算子

Posted 2022-03-15 howard2005

文章目录

• 零、本讲学习目标
• 一、RDD算子
• 二、准备工作
• • （一）准备文件
  • • 1、准备本地系统文件
    • 2、准备HDFS系统文件
  • （二）启动Spark Shell
  • • 1、启动HDFS服务
    • 2、启动Spark服务
    • 3、启动Spark Shell
• 三、转化算子
• • （一）映射算子 - map()
  • • 1、映射算子功能
    • 2、映射算子案例
    • • 任务1、将rdd1每个元素翻倍得到rdd2
      • 任务2、将rdd1每个元素平方得到rdd2
  • （二）过滤算子 - filter()
  • • 1、过滤算子功能
    • 2、过滤算子案例
    • • 任务1、过滤出列表中的偶数
      • 任务2、过滤出文件中包含`spark`的行
  • （三）扁平映射算子 - flatMap()
  • • 1、扁平映射算子功能
    • 2、扁平映射算子案例
    • • 任务1、统计文件中单词个数
      • 任务2、统计不规则二维列表元素个数
  • （四）按键归约算子 - reduceByKey()
  • • 1、按键归约算子功能
    • 2、按键归约算子案例
    • • 任务：计算学生总分

零、本讲学习目标

1. 掌握转化算子的使用
2. 掌握行动算子的使用

一、RDD算子

• RDD被创建后是只读的，不允许修改。Spark提供了丰富的用于操作RDD的方法，这些方法被称为算子。一个创建完成的RDD只支持两种算子：转化（Transformation）算子和行动（Action）算子。

二、准备工作

（一）准备文件

1、准备本地系统文件

• 在/home目录里创建words.txt
  
  

2、准备HDFS系统文件

• 将words.txt上传到HDFS系统的/park目录里
  
• 说明：/park是在上一讲我们创建的目录

（二）启动Spark Shell

1、启动HDFS服务

• 执行命令：start-dfs.sh
  

2、启动Spark服务

• 进入Spark的sbin目录执行命令：./start-all.sh
  

3、启动Spark Shell

• 执行名命令： spark-shell --master spark://master:7077
  

三、转化算子

• 转化算子负责对RDD中的数据进行计算并转化为新的RDD。Spark中的所有转化算子都是惰性的，因为它们不会立即计算结果，而只是记住对某个RDD的具体操作过程，直到遇到行动算子才会与行动算子一起执行。

（一）映射算子 - map()

1、映射算子功能

• map()是一种转化算子，它接收一个函数作为参数，并把这个函数应用于RDD的每个元素，最后将函数的返回结果作为结果RDD中对应元素的值。

2、映射算子案例

• 预备工作：创建一个RDD - rdd1
  

任务1、将rdd1每个元素翻倍得到rdd2

• 对rdd1应用map()算子，将rdd1中的每个元素平方并返回一个名为rdd2的新RDD
  
• 上述代码中，向算子map()传入了一个函数x = > x * 2。其中，x为函数的参数名称，也可以使用其他字符，例如a => a * 2。Spark会将RDD中的每个元素传入该函数的参数中。
• rdd1和rdd2中实际上没有任何数据，因为parallelize()和map()都为转化算子，调用转化算子不会立即计算结果。若需要查看计算结果，则可使用行动算子collect()。
• 执行rdd2.collect()进行计算，并将结果以数组的形式收集到当前Driver。因为RDD的元素为分布式的，数据可能分布在不同的节点上。
  
• 其实，向map()算子传入函数的参数可以用下划线_来代替，并且可以简化成_ * 2

把 x => func(x) 简化为 func _ 或 func 的过程称为eta-conversion
把 func 或 func _ 展开为 x => func(x) 的过程为eta-expansion

• 上述使用map()算子的运行过程如下图所示
  

任务2、将rdd1每个元素平方得到rdd2

• 方法一、采用普通函数作为参数传给map()算子
  
• 方法二、采用下划线表达式作为参数传给map()算子
• 刚才翻倍用的是map(_ * 2)，很自然地想到平方应该是map(_ * _)
  
• 报错，(_ * _)经过eta-expansion变成普通函数，不是我们预期的x => x * x，而是(x$1, x$2) => (x$1 * x$2)，不是一元函数，而是二元函数，系统立马就蒙逼了，不晓得该怎么取两个参数来进行乘法运算。
• 难道就不能用下划线参数了吗？当然可以，但是必须保证下划线表达式里下划线只出现1次。引入幂函数scala.math.pow就可以搞定。
  
• 但是有点瑕疵，rdd2的元素变成了双精度实数，得转化成整数
  

（二）过滤算子 - filter()

1、过滤算子功能

• 通过函数func对源RDD的每个元素进行过滤，并返回一个新的RDD。

2、过滤算子案例

任务1、过滤出列表中的偶数

• 基于列表创建RDD，然后利用过滤算子得到偶数构成的新RDD
  

任务2、过滤出文件中包含spark的行

• 执行命令： val lines= sc.textFile("hdfs://master:9000/park/words.txt")，读取文件/park/words.txt生成RDD - lines
  
• 执行命令：val sparkLines = lines.filter(_.contains("spark"))，过滤包含spark的行生成RDD - sparkLines
  
• 执行命令：sparkLines.collect()，查看sparkLines内容
  

（三）扁平映射算子 - flatMap()

1、扁平映射算子功能

• flatMap()算子与map()算子类似，但是每个传入给函数func的RDD元素会返回0到多个元素，最终会将返回的所有元素合并到一个RDD。

2、扁平映射算子案例

任务1、统计文件中单词个数

• 读取文件，生成RDD - rdd1
  
• 按空格拆分，做扁平映射，生成新RDD - rdd2
  
• 大家可以看到，经过扁平映射，生成的RDD是一个单词构成一个元素，原先的RDD是五行单词构成五个元素
• 执行命令：rdd2.count()，即可知单词个数
  
• 扁平映射算子的运行过程，rdd1的5个元素经过扁平映射变成了rdd2的20个元素
  

任务2、统计不规则二维列表元素个数

$\left[\begin{array}{ccccc}7& 8& 1& 5& \\ 10& 4& 9& & \\ 7& 2& 8& 1& 4\\ 21& 4& 7& -4& \end{array}\right]$

• 方法一、利用Scala来实现
  

val m = List(List(7, 8, 1, 5),
             List(10, 4, 9), 
             List(7, 2, 8, 1, 4),
             List(21, 4, 7, -4))

val arr = m.flatten

arr.size

• 方法二、利用Spark来实现
  

val rdd1 = sc.makeRDD(List(List(7, 8, 1, 5),
                                List(10, 4, 9), 
                                List(7, 2, 8, 1, 4),
                                List(21, 4, 7, -4)))

val rdd2 = rdd1.flatMap(x => x.toString.substring(5, x.toString.length - 1).split(", "))

rdd2.collect()

scala> rdd2.count()

（四）按键归约算子 - reduceByKey()

1、按键归约算子功能

• reduceByKey()算子的作用对像是元素为(key,value)形式（Scala元组）的RDD，使用该算子可以将相同key的元素聚集到一起，最终把所有相同key的元素合并成一个元素。该元素的key不变，value可以聚合成一个列表或者进行求和等操作。最终返回的RDD的元素类型和原有类型保持一致。

2、按键归约算子案例

任务：计算学生总分

• 成绩表

姓名	语文	数学	英语
张钦林	78	90	76
陈燕文	95	88	98
卢志刚	78	80	60

• 创建成绩列表scores，基于成绩列表创建rdd1，对rdd1按键归约得到rdd2，然后查看rdd2内容
  

val scores = List(("张钦林", 78), ("张钦林", 90), ("张钦林", 76),
                  ("陈燕文", 95), ("陈燕文", 88), ("陈燕文", 98),
                  ("卢志刚", 78),     ("卢志刚", 80), ("卢志刚", 60))

val rdd1 = sc.parallelize(scores)

val rdd2 = rdd1.reduceByKey((x, y) => x + y)

rdd2.collect()

• 不仅可以在Spark Shell里完成任务，也可以编写Scala程序生成jar提交到Spark服务器运行。有兴趣的同学不妨参看《Spark案例：两种方式计算学生总分》