国内使用vue做的网站,做网站用vs,上海工程建设造价信息网站,姜堰网站定制该文章主要为完成实训任务#xff0c;详细实现过程及结果见【http://t.csdn.cn/FArNP】 文章目录 一、准备工作1.1 准备文件1. 准备本地系统文件2. 把文件上传到 1.2 启动Spark Shell1. 启动HDFS服务2. 启动Spark服务3. 启动Spark Shell 二、掌握转换算子2.1 映射算子 - map()… 该文章主要为完成实训任务详细实现过程及结果见【http://t.csdn.cn/FArNP】 文章目录 一、准备工作1.1 准备文件1. 准备本地系统文件2. 把文件上传到 1.2 启动Spark Shell1. 启动HDFS服务2. 启动Spark服务3. 启动Spark Shell 二、掌握转换算子2.1 映射算子 - map()1. 映射算子功能2. 映射算子案例任务1、将rdd1每个元素翻倍得到rdd2任务2、将rdd1每个元素平方得到rdd2任务3、利用映射算子打印菱形 2.2 过滤算子 - filter()1. 过滤算子功能2. 过滤算子案例任务1、过滤出列表中的偶数任务2、过滤出文件中包含spark的行任务3、利用过滤算子输出[2000, 2500]之间的全部闰年任务4、利用过滤算子输出[10, 100]之间的全部素数 2.3 扁平映射算子 - flatMap()1. 扁平映射算子功能2. 扁平映射算子案例任务1、统计文件中单词个数方法一、利用Scala来实现方法二、利用Spark RDD来实现 2.4 按键归约算子 - reduceByKey()1. 按键归约算子功能2. 按键归约算子案例任务1、在Spark Shell里计算学生总分任务2、在IDEA里计算学生总分 2.5 合并算子 - union()1. 合并算子功能2. 合并算子案例 2.6 排序算子 - sortBy()1. 排序算子功能2. 排序算子案例 2.7 按键排序算子 - sortByKey()1. 按键排序算子功能2. 按键排序算子案例 2.8 连接算子1. 内连接算子 - join()2. 左外连接算子 - leftOuterJoin()3. 右外连接算子 - rightOuterJoin()4. 全外连接算子 - fullOuterJoin() 2.9 交集算子 - intersection()2.10 去重算子 - distinct()1. 去重算子案例2. IP地址去重案例 十一组合分组算子 - cogroup() 一、准备工作
1.1 准备文件
1. 准备本地系统文件
在\home目录里创建words.txt
2. 把文件上传到
将words.txt上传到HDFS系统的/park目录里 查看文件内容
1.2 启动Spark Shell
1. 启动HDFS服务
执行命令start-dfs.sh
2. 启动Spark服务
执行命令start-all.sh
3. 启动Spark Shell
执行名命令 spark-shell --master spark://master:7077
二、掌握转换算子
2.1 映射算子 - 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 查看结果
任务2、将rdd1每个元素平方得到rdd2
方法一、采用普通函数作为参数传给map()算子 方法二、采用下划线表达式作为参数传给map()算子
任务3、利用映射算子打印菱形
1Spark Shell里实现 右半菱形 加上前导空格左半菱形 前导空格折半显示菱形 2在IDEA里创建项目实现 创建Maven项目 将java目录改成scala目录 在pom.xml文件里添加相关依赖和设置源程序目录
?xml version1.0 encodingUTF-8?
project xmlnshttp://maven.apache.org/POM/4.0.0xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersiongroupIdcn.kox.rdd/groupIdartifactIdSparkRDDDemo/artifactIdversion1.0-SNAPSHOT/versiondependenciesdependencygroupIdorg.scala-lang/groupIdartifactIdscala-library/artifactIdversion2.12.15/version/dependencydependencygroupIdorg.apache.spark/groupIdartifactIdspark-core_2.12/artifactIdversion3.1.3/version/dependency/dependenciesbuildsourceDirectorysrc/main/scala/sourceDirectory/buildpropertiesmaven.compiler.source8/maven.compiler.sourcemaven.compiler.target8/maven.compiler.targetproject.build.sourceEncodingUTF-8/project.build.sourceEncoding/properties/project添加日志属性文件
log4j.rootLoggerERROR, stdout, logfile
log4j.appender.stdoutorg.apache.log4j.ConsoleAppender
log4j.appender.stdout.layoutorg.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern%d %p [%c] - %m%n
log4j.appender.logfileorg.apache.log4j.FileAppender
log4j.appender.logfile.Filetarget/rdd.log
log4j.appender.logfile.layoutorg.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern%d %p [%c] - %m%n
创建hdfs-site.xml文件允许客户端访问集群数据节点
?xml version1.0 encodingUTF-8?
configurationpropertydescriptiononly config in clients/descriptionnamedfs.client.use.datanode.hostname/namevaluetrue/value/property
/configuration
创建cn.kox.rdd.day01包 在cn.kox.rdd.day01包里创建Example01单例对象
package cn.kox.rdd.day01import org.apache.spark.{SparkConf, SparkContext}import scala.collection.mutable.ListBuffer
import scala.io.StdIn/*** ClassName: Example01* Author: Kox* Data: 2023/6/12* Sketch:*/
object Example01 {def main(args: Array[String]): Unit {// 创建Spark配置对象val conf new SparkConf().setAppName(PrintDiamond) // 设置应用名称.setMaster(local[*]) // 设置主节点位置本地调试// 基于Spark配置对象创建Spark容器val sc new SparkContext(conf)// 输入一个奇数print(输入一个奇数)val n StdIn.readInt()//判断n的奇偶性if (n % 2 0) {println(温馨提示你输入的不是奇数)return }// 创建一个可变列表val list new ListBuffer[Int]()// 给列表赋值(1 to n by 2).foreach(list.append(_))(n - 2 to 1 by -2).foreach(list.append(_))// 基于列表创建rddval rdd sc.makeRDD(list)// 对rdd进行映射操作val rdd1 rdd.map(i * ((n - i) / 2) * * i)// 输出rdd1结果rdd1.collect.foreach(println)}
}运行程序查看结果
2.2 过滤算子 - filter()
1. 过滤算子功能
filter(func)通过函数func对源RDD的每个元素进行过滤并返回一个新RDD一般而言新RDD元素个数会少于原RDD。
2. 过滤算子案例
任务1、过滤出列表中的偶数
整数Integer奇数odd number 偶数even number基于列表创建RDD然后利用过滤算子得到偶数构成的新RDD方法一、将匿名函数传给过滤算子 方法二、用神奇占位符改写传入过滤算子的匿名函数
任务2、过滤出文件中包含spark的行
查看源文件/park/words.txt内容 执行命令 val lines sc.textFile(/park/words.txt)读取文件 /park/words.txt生成RDD - lines 执行命令val sparkLines lines.filter(_.contains(spark))过滤包含spark的行生成RDD - sparkLines 执行命令sparkLines.collect查看sparkLines内容可以采用遍历算子分行输出内容 输出长度超过20的行
任务3、利用过滤算子输出[2000, 2500]之间的全部闰年
传统做法利用循环结构嵌套选择结构来实现 要求每行输出10个数 采用过滤算子来实现 要求每行输出10个数
任务4、利用过滤算子输出[10, 100]之间的全部素数
过滤算子filter(n !(n % 2 0 || n % 3 0 || n % 5 0 || n % 7 0))
2.3 扁平映射算子 - flatMap()
1. 扁平映射算子功能
flatMap()算子与map()算子类似但是每个传入给函数func的RDD元素会返回0到多个元素最终会将返回的所有元素合并到一个RDD。
2. 扁平映射算子案例
任务1、统计文件中单词个数
读取文件生成RDD - rdd1查看其内容和元素个数对于rdd1按空格拆分做映射生成新RDD - rdd2 对于rdd1按空格拆分做扁平映射生成新RDD - rdd3有一个降维处理的效果 统计结果文件里有25个单词
方法一、利用Scala来实现
利用列表的flatten函数在cn.kox.rdd.day01包里创建Example02单例对象
package cn.kox.rdd.day01import org.apache.spark.{SparkConf, SparkContext}/*** ClassName: Example02* Author: Kox* Data: 2023/6/12* Sketch:*/
object Example02 {def main(args: Array[String]): Unit {// 创建不规则二维列表val mat List(List(7, 8, 1, 5),List(10, 4, 9),List(7, 2, 8, 1, 4),List(21, 4, 7, -4))// 输出二维列表println(mat)// 将二维列表扁平化为一维列表val arr mat.flatten// 输出一维列表println(arr)// 输出元素个数println(元素个数 arr.size)}
}
运行程序查看结果
方法二、利用Spark RDD来实现
利用flatMap算子在cn.kox.rdd.day01包里创建Example03单例对象
package cn.kox.rdd.day01import org.apache.spark.{SparkConf, SparkContext}
/*** ClassName: Example03* Author: Kox* Data: 2023/6/12* Sketch:*/
object Example03 {def main(args: Array[String]): Unit {// 创建Spark配置对象val conf new SparkConf().setAppName(PrintDiamond) // 设置应用名称.setMaster(local[*]) // 设置主节点位置本地调试// 基于Spark配置对象创建Spark容器val sc new SparkContext(conf)// 创建不规则二维列表val mat List(List(7, 8, 1, 5),List(10, 4, 9),List(7, 2, 8, 1, 4),List(21, 4, 7, -4))// 基于二维列表创建rdd1val rdd1 sc.makeRDD(mat)// 输出rdd1rdd1.collect.foreach(x print(x ))println()// 进行扁平化映射val rdd2 rdd1.flatMap(x x.toString.substring(5, x.toString.length - 1).split(, ))// 输出rdd2rdd2.collect.foreach(x print(x ))println()// 输出元素个数println(元素个数 rdd2.count)}
}
运行程序查看结果
2.4 按键归约算子 - reduceByKey()
1. 按键归约算子功能
reduceByKey()算子的作用对像是元素为(key,value)形式Scala元组的RDD使用该算子可以将相同key的元素聚集到一起最终把所有相同key的元素合并成一个元素。该元素的key不变value可以聚合成一个列表或者进行求和等操作。最终返回的RDD的元素类型和原有类型保持一致。
2. 按键归约算子案例
任务1、在Spark Shell里计算学生总分 创建成绩列表scores基于成绩列表创建rdd1对rdd1按键归约得到rdd2然后查看rdd2内容agg: aggregation 聚合值cur: current 当前值
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里计算学生总分 第一种方式读取二元组成绩列表
在cn.kox.rdd.day02包里创建CalculateScoreSum01单例对象
package cn.kox.rdd.day02import org.apache.spark.{SparkConf, SparkContext}/*** ClassName: CalculateScoreSum01* Author: Kox* Data: 2023/6/12* Sketch:*/
object CalculateScoreSum01 {def main(args: Array[String]): Unit {// 创建Spark配置对象val conf new SparkConf().setAppName(PrintDiamond) // 设置应用名称.setMaster(local[*]) // 设置主节点位置本地调试// 基于Spark配置对象创建Spark容器val sc new SparkContext(conf)// 创建二元组成绩列表val scores List((张钦林, 78), (张钦林, 90), (张钦林, 76),(陈燕文, 95), (陈燕文, 88), (陈燕文, 98),(卢志刚, 78), (卢志刚, 80), (卢志刚, 60))// 基于二元组成绩列表创建RDDval rdd1 sc.makeRDD(scores)// 对成绩RDD进行按键归约处理val rdd2 rdd1.reduceByKey(_ _)// 输出归约处理结果rdd2.collect.foreach(println)}
}
运行程序查看结果 第二种方式读取四元组成绩列表在cn.kox.rdd.day02包里创建CalculateScoreSum02单例对象
package cn.kox.rdd.day02import org.apache.spark.{SparkConf, SparkContext}import scala.collection.mutable.ListBuffer
/*** ClassName: CalculateScoreSum02* Author: Kox* Data: 2023/6/12* Sketch:*/
object CalculateScoreSum02 {def main(args: Array[String]): Unit {// 创建Spark配置对象val conf new SparkConf().setAppName(PrintDiamond) // 设置应用名称.setMaster(local[*]) // 设置主节点位置本地调试// 基于Spark配置对象创建Spark容器val sc new SparkContext(conf)// 创建四元组成绩列表val scores List((张钦林, 78, 90, 76),(陈燕文, 95, 88, 98),(卢志刚, 78, 80, 60))// 将四元组成绩列表转化成二元组成绩列表val newScores new ListBuffer[(String, Int)]()// 通过遍历算子遍历四元组成绩列表scores.foreach(score {newScores.append(Tuple2(score._1, score._2))newScores.append(Tuple2(score._1, score._3))newScores.append(Tuple2(score._1, score._4))})// 基于二元组成绩列表创建RDDval rdd1 sc.makeRDD(newScores)// 对成绩RDD进行按键归约处理val rdd2 rdd1.reduceByKey(_ _)// 输出归约处理结果rdd2.collect.foreach(println)}
}
运行程序查看结果
2.5 合并算子 - union()
1. 合并算子功能
union()算子将两个RDD合并为一个新的RDD主要用于对不同的数据来源进行合并两个RDD中的数据类型要保持一致。
2. 合并算子案例
创建两个RDD合并成一个新RDD 课堂练习将两个二元组成绩表合并在集合运算里并集符号∪ \cup∪并集运算A ∪ B A \cup BA∪B在集合运算里交集符号∩ \cap∩交集运算A ∩ B A \cap BA∩B在集合运算里补集运算A ˉ \bar A
2.6 排序算子 - sortBy()
1. 排序算子功能
sortBy()算子将RDD中的元素按照某个规则进行排序。该算子的第一个参数为排序函数第二个参数是一个布尔值指定升序默认或降序。若需要降序排列则需将第二个参数置为false。
2. 排序算子案例
一个数组中存放了三个元组将该数组转为RDD集合然后对该RDD按照每个元素中的第二个值进行降序排列。 sortBy(xx._2,false)中的x代表rdd1中的每个元素。由于rdd1的每个元素是一个元组因此使用x._2取得每个元素的第二个值。当然sortBy(xx.2,false)也可以直接简化为sortBy(._2,false)。
2.7 按键排序算子 - sortByKey()
1. 按键排序算子功能
sortByKey()算子将(key, value)形式的RDD按照key进行排序。默认升序若需降序排列则可以传入参数false。
2. 按键排序算子案例
将三个二元组构成的RDD按键先降序排列然后升序排列 其实用排序算子也是可以搞定的
2.8 连接算子
1. 内连接算子 - join()
join()算子将两个(key, value)形式的RDD根据key进行连接操作相当于数据库的内连接Inner Join只返回两个RDD都匹配的内容。将rdd1与rdd2进行内连接
2. 左外连接算子 - leftOuterJoin()
leftOuterJoin()算子与数据库的左外连接类似以左边的RDD为基准例如rdd1.leftOuterJoin(rdd2)以rdd1为基准左边RDD的记录一定会存在。例如rdd1的元素以(k,v)表示rdd2的元素以(k, w)表示进行左外连接时将以rdd1为基准rdd2中的k与rdd1的k相同的元素将连接到一起生成的结果形式为(k, (v, Some(w))。rdd1中其余的元素仍然是结果的一部分元素形式为(k,(v, None)。Some和None都属于Option类型Option类型用于表示一个值是可选的有值或无值)。若确定有值则使用Some值表示该值若确定无值则使用None表示该值。rdd1与rdd2进行左外连接
3. 右外连接算子 - rightOuterJoin()
rightOuterJoin()算子的使用方法与leftOuterJoin()算子相反其与数据库的右外连接类似以右边的RDD为基准例如rdd1.rightOuterJoin(rdd2)以rdd2为基准右边RDD的记录一定会存在。rdd1与rdd2进行右外连接
4. 全外连接算子 - fullOuterJoin()
fullOuterJoin()算子与数据库的全外连接类似相当于对两个RDD取并集两个RDD的记录都会存在。值不存在的取None。rdd1与rdd2进行全外连接
2.9 交集算子 - intersection()
intersection()算子对两个RDD进行交集操作返回一个新的RDD。要求两个算子类型要一致。rdd1与rdd2进行交集操作满足交换律 A ∩ B ≠ ϕ
2.10 去重算子 - distinct()
distinct()算子对RDD中的数据进行去重操作返回一个新的RDD。有点类似与集合的不允许重复元素。
1. 去重算子案例
去掉rdd中重复的元素
2. IP地址去重案例
在项目根目录创建ips.txt文件
192.168.234.21
192.168.234.22
192.168.234.21
192.168.234.21
192.168.234.23
192.168.234.21
192.168.234.21
192.168.234.21
192.168.234.25
192.168.234.21
192.168.234.21
192.168.234.26
192.168.234.21
192.168.234.27
192.168.234.21
192.168.234.27
192.168.234.21
192.168.234.29
192.168.234.21
192.168.234.26
192.168.234.21
192.168.234.25
192.168.234.25
192.168.234.21
192.168.234.22
192.168.234.21
在cn.kox.rdd.day03包里创建DistinctIPs单例对象
package cn.kox.rdd.day03import org.apache.spark.{SparkConf, SparkContext}/*** ClassName: DistinctIPs* Author: Kox* Data: 2023/6/12* Sketch:*/
object DistinctIPs {def main(args: Array[String]): Unit {// 创建Spark配置对象val conf new SparkConf().setAppName(DistinctIPs ) // 设置应用名称.setMaster(local[*]) // 设置主节点位置本地调试// 基于Spark配置对象创建Spark容器val sc new SparkContext(conf)// 读取本地IP地址文件得到RDDval ips sc.textFile(D:\\Major\\BigData\\Spark\\SparkLesson2023U\\SparkRDDDemo\\ips.txt)// rdd去重再输出ips.distinct.collect.foreach(println)}
}
运行程序查看结果 修改代码保存去重结果到本地目录
十一组合分组算子 - cogroup()
cogroup()算子对两个(key, value)形式的RDD根据key进行组合相当于根据key进行并集操作。例如rdd1的元素以(k, v)表示rdd2的元素以(k, w)表示执行rdd1.cogroup(rdd2)生成的结果形式为(k, (Iterable, Iterable))。rdd1与rdd2进行组合分组操作
