淘宝网站开发,用python做网站,单页面网站模板,wordpress判断使用不同模板在之前的文章中#xff0c;我们学习了如何在IDEA中导入jars包#xff0c;并做了一道例题#xff0c;了解了RDD。想了解的朋友可以查看这篇文章。同时#xff0c;希望我的文章能帮助到你#xff0c;如果觉得我的文章写的不错#xff0c;请留下你宝贵的点赞#xff0c;谢谢…在之前的文章中我们学习了如何在IDEA中导入jars包并做了一道例题了解了RDD。想了解的朋友可以查看这篇文章。同时希望我的文章能帮助到你如果觉得我的文章写的不错请留下你宝贵的点赞谢谢。
Spark-Scala语言实战6-CSDN博客文章浏览阅读695次点赞15次收藏24次。今天我会给大家带来如何在IDEA中导入jars包以及使用SparkRDD并正确使用它们同时也会给大家讲解一道实训题例。希望在本篇文章中大家有所收获。也欢迎朋友们到评论区下一起交流学习共同进步。https://blog.csdn.net/qq_49513817/article/details/137121524?spm1001.2014.3001.5502
今天开始的文章我会带给大家如何在spark的中使用我们的RDD方法今天学习RDD方法中的map,sortby,collect三种方法。
目录
一、知识回顾
二、RDD方法
1.map
2.sortby
3.collect
拓展-RDD和DStream
1.RDD和DStream的区别
2.RDD和DStream的联系 一、知识回顾
导入jars包的过程在上一篇文章中以及讲解的很清楚了图文一步一步带着做。
主要就是进入Libraries 添加java然后选择spark的jars文件夹即可 如果还有不懂的朋友可以直接评论问我。
在就是文件的这几行代码
import org.apache.spark.{SparkConf, SparkContext}val confnew SparkConf().setMaster(local).setAppName(123456)val scnew SparkContext(conf)
这是配置与方法记住它们的作用。 现在开始今天的学习吧 二、RDD方法
1.map
map()方法是一种基础的RDD转换操作可以对RDD中的每一个数据元素通过某种函数进行转换并返回新的RDD。map()方法是转换操作不会立即进行计算。转换操作是创建RDD的第二种方法通过转换已有RDD生成新的RDD。因为RDD是一个不可变的集合所以如果对RDD数据进行了某种转换那么会生成一个新的RDD。
例
import org.apache.spark.{SparkConf, SparkContext} // 定义一个名为p1的Scala对象
object p1 { // 定义main方法作为程序的入口点 def main(args: Array[String]): Unit { // 创建一个Spark配置对象并设置运行模式为local本地模式应用程序名称为p2 val conf new SparkConf().setMaster(local).setAppName(p2) // 使用Spark配置对象创建一个SparkContext对象SparkContext是Spark功能的入口点 val sc new SparkContext(conf) // 创建一个包含整数的列表并使用parallelize方法将其转换为RDD val ppp sc.parallelize(List(1, 2, 3, 4, 5)) // 使用map操作将RDD中的每个元素乘以2并返回一个新的RDD val ppppp ppp.map(x x * 2) //oreach方法遍历并打印每个元素 ppppp.collect().foreach(println) }
} 可以看到我们输出的在原列表上*2达到了代码预期效果
2.sortby
sortBy()方法用于对标准RDD进行排序有3个可输入参数说明如下。第1个参数是一个函数f:(T) K左边是要被排序对象中的每一个元素右边返回的值是元素中要进行排序的值。第2个参数是ascending决定排序后RDD中的元素是升序的还是降序的默认是true即升序排序如果需要降序排序那么需要将参数的值设置为false。第3个参数是numPartitions决定排序后的RDD的分区个数默认排序后的分区个数和排序之前的分区个数相等即this.partitions.size。第一个参数是必须输入的而后面的两个参数可以不输入。
例
import org.apache.spark.{SparkConf, SparkContext} object p1 { def main(args: Array[String]): Unit { val conf new SparkConf().setMaster(local).setAppName(p2) // 使用配置好的conf对象创建一个SparkContext对象sc。 val sc new SparkContext(conf) // 使用SparkContext的parallelize方法将包含整数的序列转换成一个RDD。 // 这个RDD现在可以在Spark上并行处理。 val ppp sc.parallelize(Seq(5, 1, 9, 3, 7)) // 对ppp RDD中的元素进行排序。 // 使用sortBy方法并传递一个函数x x作为参数表示按照元素本身的值进行排序升序。 val pppp ppp.sortBy(x x) // 这将返回一个包含RDD所有元素的数组存储在ppppp中。 val ppppp pppp.collect() // 使用foreach方法遍历数组ppppp中的每个元素并使用println函数打印它们。 ppppp.foreach(println) }
} 看下输出可以看到我们的元素已经排序了
3.collect
collect()方法是一种行动操作可以将RDD中所有元素转换成数组并返回到Driver端适用于返回处理后的少量数据。因为需要从集群各个节点收集数据到本地经过网络传输并且加载到Driver内存中所以如果数据量比较大会给网络传输造成很大的压力。因此数据量较大时尽量不使用collect()方法否则可能导致Driver端出现内存溢出问题。
例
import org.apache.spark.{SparkConf, SparkContext}object p1 {def main(args: Array[String]): Unit {val confnew SparkConf().setMaster(local).setAppName(p2)val scnew SparkContext(conf)val pp sc.parallelize(Seq(1, 2, 3, 4, 5))val ppp pp.collect()ppp.foreach(println)}
} collect的作用是将RDD中的数据收集到驱动程序中所以这里运行看不出区别。
拓展-RDD和DStream
在上一篇文章中我们了解到了RDD那么DStream是什么呢我们先来了解一下
DStream离散流是Spark Streaming提供的一种高级抽象代表了一个持续不断的数据流。DStream的内部实际上是一系列持续不断产生的RDD每个RDD包含特定时间间隔的数据。DStream的创建可以通过输入数据源如Kafka、Flume或者通过对其他DStream应用高阶函数如map、reduce、join、window来实现。
1.RDD和DStream的区别
RDDDStream定义弹性分布式数据集是Spark中最基本的数据处理模型。离散流是Spark Streaming提供的一种高级抽象代表一个持续不断的数据流。数据结构静态的、不可变的数据集可以划分为多个分区。动态的、连续的数据流内部由一系列RDD组成。数据处理方式批处理适用于静态数据的处理和分析。流处理适用于实时数据流的处理和分析。时间维度无特定的时间维度主要关注数据的分区和处理。具有时间维度每个RDD代表一段时间内的数据。操作方式对整个RDD进行操作结果生成新的RDD。对DStream进行操作结果生成新的DStream底层转换为RDD操作。应用场景大规模数据的批处理任务如机器学习、数据挖掘等。实时数据流处理任务如日志分析、实时监控等。容错性具有容错性数据丢失可以自动恢复。继承了RDD的容错性特点。与Spark的关系Spark的核心组件用于构建各种数据处理和分析任务。Spark Streaming的核心组件用于处理实时数据流。
2.RDD和DStream的联系
RDDDStream基础构建单元RDD是Spark的基本数据处理单元。DStream基于RDD构建每个时间间隔内的数据对应一个RDD。计算模型RDD支持分布式计算模型数据被划分为多个分区进行并行处理。DStream继承了RDD的计算模型对流数据进行分布式处理。容错性RDD具有容错性可以自动恢复丢失的数据。DStream同样具有容错性因为它基于RDD构建。操作方式RDD提供了一系列转换操作如map、reduce和动作操作如collect、save。DStream也提供了类似的操作这些操作最终会转换为底层RDD的操作。数据处理能力RDD适用于批处理任务可以对大规模数据集进行处理和分析。DStream适用于实时流处理任务可以对连续的数据流进行实时分析和处理。底层实现DStream内部实际上是由一系列RDD组成的每个RDD代表一段时间内的数据。DStream的操作最终会转换为RDD的操作利用RDD的分布式计算能力。扩展性RDD可以通过自定义操作进行扩展支持更多的数据处理场景。DStream同样可以通过自定义操作和转换函数进行扩展以满足特定的实时处理需求
