文章目录
一、RDD算子
RDD被创建后是只读的,不允许修改。Spark提供了丰富的用于操作RDD的方法,这些方法被称为算子。一个创建完成的RDD只支持两种算子:转化(Transformation)算子和行动(Action)算子
二、准备工作
(一)准备文件
1、准备本地系统文件
在/home目录里创建wprds.txt
2、把文件上传到HDFS
将words.txt上传到HDFS系统的/park目录里
说明:/park是在上一讲我们创建的目录
查看文件内容
(二)启动Spark Shell
1、启动HDFS服务
执行命令:start-dfs.sh
如果采用的是Spark on YARN集群或者Spark HA集群,那么还得启动YARN服务
2、启动Spark服务
进入Spark的sbin目录执行命令:./start-all.sh
3、启动Spark Shell
执行名命令: spark-shell --master spark://master:7077
三、转化算子
(一)映射算子 - map()
1、映射算子功能
map()是一种转化算子,它接收一个函数作为参数,并把这个函数应用于RDD的每个元素,最后将函数的返回结果作为结果RDD中对应元素的值。
2、映射算子案例
预备工作:创建一个RDD - rdd1
执行命令:val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5, 6))
任务1、将rdd1每个元素翻倍得到rdd2
对rdd1应用map()算子,将rdd1中的每个元素平方并返回一个名为rdd2的新RDD
上述代码中,向算子map()传入了一个函数x = > x * 2。其中,x为函数的参数名称,也可以使用其他字符,例如a => a * 2。Spark会将RDD中的每个元素传入该函数的参数中。
其实,利用神奇占位符_可以写得更简洁
rdd1和rdd2中实际上没有任何数据,因为parallelize()和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、过滤算子功能
filter(func):通过函数func对源RDD的每个元素进行过滤,并返回一个新RDD,一般而言,新RDD元素个数会少于原RDD。
2、过滤算子案例
任务1、过滤出列表中的偶数
基于列表创建RDD,然后利用过滤算子得到偶数构成的新RDD
方法一、将匿名函数传给过滤算子
方法二、用神奇占位符改写传入过滤算子的匿名函数
任务2、过滤出文件中包含spark的行
查看源文件/park/words.txt内容
执行命令: 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内容,可以采用遍历算子,分行输出内容太
课堂练习
任务1、利用过滤算子输出[2000, 2500]之间的全部闰年
采用过滤算子来实现
(三)扁平映射算子 - flatMap()
1、扁平映射算子功能
flatMap()算子与map()算子类似,但是每个传入给函数func的RDD元素会返回0到多个元素,最终会将返回的所有元素合并到一个RDD。
2、扁平映射算子案例
任务1、统计文件中单词个数
对于rdd1按空格拆分,做映射,生成新RDD - rdd2
对于rdd1按空格拆分,做扁平映射,生成新RDD - rdd3
大家可以看到,经过扁平映射,生成的RDD是一个单词构成一个元素,而rdd1是6行单词构成6个元素
执行命令:rdd3.count(),即可知单词个数
任务2、统计不规则二维列表元素个数
方法一、利用Scala来实现
利用列表的flatten函数
方法二、利用Spark RDD来实现
利用flatMap算子
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(", "))
println("元素个数:" + rdd2.count)
查看rdd2的内容
scala> rdd2.collect
res26: Array[String] = Array(7, 8, 1, 5, 10, 4, 9, 7, 2, 8, 1, 4, 21, 4, 7, -4)
(三)扁平映射算子 - flatMap()
1、按键归约算子功能
reduceByKey()算子的作用对像是元素为(key,value)形式(Scala元组)的RDD,使用该算子可以将相同key的元素聚集到一起,最终把所有相同key的元素合并成一个元素。该元素的key不变,value可以聚合成一个列表或者进行求和等操作。最终返回的RDD的元素类型和原有类型保持一致
2、按键归约算子案例
任务1、在Spark Shell里计算学生总分
创建成绩列表scores,基于成绩列表创建rdd1,对rdd1按键归约得到rdd2,然后查看rdd2内容
val scores = List(("张钦林", 78), ("张钦林", 90), ("张钦林", 76),
("陈燕文", 95), ("陈燕文", 88), ("陈燕文", 98),
("卢志刚", 78), ("卢志刚", 80), ("卢志刚", 60))
val rdd1 = sc.makeRDD(scores)
val rdd2 = rdd1.reduceByKey((x, y) => x + y)
rdd2.collect.foreach(println)
任务2、在IDEA里计算学生总分
创建Maven项目-SparkRDDemo
将java目录改为scala
在pom.xml文件里添加依赖和构建插件
在资源文件夹里创建日志属性文件
log4j.rootLogger=ERROR, stdout, logfile
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/spark.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n
创建net.py.rdd包
第一种方式:读取二元组成绩列表
在net.py.rdd包里创建CalculateScoreSum单例对象
查看运行结果
第二种方式:读取四元组成绩列表
在net.py.rdd包里创建CalculateScoreSum02单例对象
查看运行结果
第三种情况:读取HDFS上的成绩文件
在master虚拟机的/home目录里创建成绩文件 - scores.txt
将成绩文件上传到HDFS的/input目录
在net.py.rdd包里创建CalculateScoreSum03单例对象
在Spark Shell里完成同样的任务