由于RDD的不可修改的特性,导致RDD的操作与正常面向对象的操作不同,RDD的操作基本分为3大类:transformation,action,contoller
1. Transformation
Transformation是通过转化针对已有的RDD创建出新的RDD
map(func):对调用map的RDD数据集中的每个element都使用func,然后返回一个新的RDD,这个返回的数据集是分布式的数据集
filter(func): 对调用filter的RDD数据集中的每个元素都使用func,然后返回一个包含使func为true的元素构成的RDD
flatMap(func):和map差不多,但是flatMap生成的是多个结果
mapPartitions(func):和map很像,但是map是每个element,而mapPartitions是每个partition
mapPartitionsWithSplit(func):和mapPartitions很像,但是func作用的是其中一个split上,所以func中应该有index
sample(withReplacement,faction,seed):抽样
union(otherDataset):返回一个新的dataset,包含源dataset和给定dataset的元素的集合
distinct([numTasks]):返回一个新的dataset,这个dataset含有的是源dataset中的distinct的element
groupByKey(numTasks):返回(K,Seq[V]),也就是hadoop中reduce函数接受的key-valuelist
reduceByKey(func,[numTasks]):就是用一个给定的reducefunc再作用在groupByKey产生的(K,Seq[V]),比如求和,求平均数
sortByKey([ascending],[numTasks]):按照key来进行排序,是升序还是降序,ascending是boolean类型
join(otherDataset,[numTasks]):当有两个KV的dataset(K,V)和(K,W),返回的是(K,(V,W))的dataset,numTasks为并发的任务数
cogroup(otherDataset,[numTasks]):当有两个KV的dataset(K,V)和(K,W),返回的是(K,Seq[V],Seq[W])的dataset,numTasks为并发的任务数
Transformation特性:
lazy优化:由于Tranformation的lazy特性,也就是创建不马上运行,对于框架来说,我有足够的时间查看到尽可能多的步骤,看到的步骤越多,优化的空间就越大。最简单的优化方式就是步骤合并,例如本来的做法是a=b*3;b=c*3;c=d*3;d=3,步骤合并后就是a=3*3*3*3。
2. Action
Action操作的目的是得到一个值,或者一个结果
reduce(func):说白了就是聚集,但是传入的函数是两个参数输入返回一个值,这个函数必须是满足交换律和结合律的
collect():一般在filter或者足够小的结果的时候,再用collect封装返回一个数组
count():返回的是dataset中的element的个数
first():返回的是dataset中的第一个元素
take(n):返回前n个elements,这个士driverprogram返回的
takeSample(withReplacement,num,seed):抽样返回一个dataset中的num个元素,随机种子seed
saveAsTextFile(path):把dataset写到一个textfile中,或者hdfs,或者hdfs支持的文件系统中,spark把每条记录都转换为一行记录,然后写到file中
saveAsSequenceFile(path):只能用在key-value对上,然后生成SequenceFile写到本地或者hadoop文件系统
countByKey():返回的是key对应的个数的一个map,作用于一个RDD
foreach(func):对dataset中的每个元素都使用func
3. Contoller
Contoller动作主要为持久化RDD,例如cache(),persist(),checkpoint();
具体内容在后续刊物中会讲解。
4. Spark WordCount动手实践
本小节通过IDEA具体逐步调试一个WordCount案例,让学员知道各步骤中RDD的具体类型,并为下一节逐步解析做铺垫
(1) 使用的wordCount代码如下:
- object WordCount {
- def main (args: Array[String]) {
- val conf = new SparkConf()//create SparkConf
- conf.setAppName("Wow,My First Spark App")//set app name
- conf.setMaster("local")//run local
- val sc =new SparkContext(conf)
- val lines =sc.textFile("C://Users//feng//IdeaProjects//WordCount//src//SparkText.txt")
- val words = lines.flatMap{ lines => lines.split(" ") }
- val pairs =words.map ( word => (word,1) )
- val reduce = pairs.reduceByKey(_+_)
- val sort_1 = reduce.map(pair=>(pair._2,pair._1))
- val sort_2 = sort_1.sortByKey(false)
- val sort_3=sort_2.map(pair=>(pair._2,pair._1))
- val filter=sort_3.filter(pair=>pair._2>2)
- filter.collect.foreach(wordNumberPair => println(wordNumberPair._1+" : "+wordNumberPair._2))
- sc.stop()
- }
- }
(1) 程序使用的SparkText.txt文件内容如下
hadoop hadoop hadoop
spark Flink spark
scala scala object
object spark scala
spark spark
Hadoop hadoop
(2) 程序WordCount调试结果:
通过IDEA的逐步调试,会在调试窗口显示每一行代码具体操作什么类型的RDD,此RDD通过什么依赖关系依赖于父RDD等重要信息(如图2-14所示),程序运行结果如图2-15所示。
图2-14调试过程图
图2-15wordCount结果
2.8.2 解析RDD生成的内部机制
本小节基于上小节程序的调试结果,逐条查看调试信息内容,并基于信息内容进行讲解,并在讲解中回顾并复习本章所有内容。
(1) line = sc.textFile()
本语句的作用在于从外部数据中读取数据,并生成MapPartitionsRDD。此处需要注意:
如图2-16所示,可以看出次MapPartitionsRDD的deps(dependency,依赖)为HadoopRDD,从这里可以发现其实textFile()过程包含两个步骤,第一步骤将文件内容转化为HadoopRDD(key-value形式,key为行号),第二步骤将HadoopRDD转化为MapPartitionsRDD(value形式,将key-value类型的key删去)
图2-16通过HadoopRDD获取数据
(2) words=line.flatMap()
此命令对于RDD采取transformation(转换)操作,作用在于将MapPartitionsRDD中的每一个记录进行以空格为标记的切分,并把每一个RDD的切分的结果放在一个MapPartitionRDD中
(3) pairs=words.map(word=>(word,1))
此命令对于RDD采取transformation(转换)操作,作用在于将MapPartitionsRDD中的每一个记录(例:spark(value类型))转换为key-value类型(例: (spark,1)),便于下一步reduceByKey操作
(4) reduce = pairs.reduceByKey(_+_)
此命令对于RDD采取action(动作)操作,作用在于通过shuffle将pairs中所有的记录按照key相同value相加的规则进行处理,并把结果放到一个shuffleRDD中。例((spark,1),(spark,1))变成((spark,2))。
同时需要注意一下两点:首先本步骤实质上分为两个步骤,第一步骤为local级别的reduce,对当前计算机所拥有的数据先进行reduce操作,生成MapPartitionsRDD;第二步骤为shuffle级别的reduce,基于第一步骤的结果,对结果进行shuffle-reduce操作,生成最终的shuffleRDD。其次 Action操作进行时,对此操作之前的所有转换操作进行执行,所以调试过程中会出现此前的除textFile操作的执行时间均非常短,说明RDD转换操作不直接进行运算。
(5) sort_1 = reduce.map(pair=>(pair._2,pair._1))
此命令对于RDD采取transformation(转换)操作,作用在于将shuffleRDD中的每一个记录的key和value互换,生成一个新的MapPartitionsRDD。例: (spark,2)变为(2,spark)
(6) sort_2 = sort_1.sortByKey(false)
此命令对于RDD采取action(动作)操作,作用在于将MapPartitionsRDD根据key进行排序,并生成shuffleRDD
(7) sort_3=sort_2.map(pair=>(pair._2,pair._1))
此命令对于RDD采取transformation(转换)操作,作用在于将shuffleRDD中的每一个记录的key和value互换,生成一个新的MapPartitionsRDD。例: (2,spark)变为(spark,2)
(8) filter=sort_3.filter(pair=>pair._2>2)
此命令对于RDD采取transformation(转换)操作,作用在于根据value值筛选MapPartitionsRDD中的数据,输出value大于2的记录
(9) 最后通过collect()方法将结果收集后,使用foreach()方法遍历数据并通过println()方法打印出所有数据。
注:本内容原型来自 IMP 课程笔记
相关推荐
spark rdd 实战 ,基本语法
知识点介绍、代码演示、逻辑分析、灵活举例、使用图形的方式详细演示代码的流程和细节、整合企业级实战案例,全面讲解并突出重点,让学习也变成一种快乐。 课程亮点 1,知识体系完备,阶段学习者都能学有所获。 2,...
章 电光石火间体验Spark 2.2开发实战... 2 1.1 通过RDD实战电影点评系统入门及源码阅读... 2 1.1.1 Spark核心概念图解... 2 1.1.2 通过RDD实战电影点评系统案例... 4 1.2 通过DataFrame和DataSet实战电影点评...
通过spark实现运营案例,使用spark的python接口实现。
10 实战解析spark运行原理和RDD解密
3、不仅适合小白学习实战练习,也可作为大作业、课程设计、毕设项目、初期项目立项演示等,欢迎下载,互相学习,共同进步! 电影评分数据汇总,(使用spark2.4+scala完成, 分析采用spark RDD的API. 数据集采用标准...
【资源说明】 1、该资源内项目代码都是经过测试运行成功,功能正常的情况下才上传的,请...3、不仅适合小白学习实战练习,也可作为大作业、课程设计、毕设项目、初期项目立项演示等,欢迎下载,互相学习,共同进步!
在“One Stack to rule them all”理念的指引下,Spark基于RDD成功地构建起了大数据处理的一体化解决方案,将MapReduce、Streaming、SQL、Machine Learning、Graph Processing等大数据计算模型统一到一个技术堆栈中...
在Spark 1.3之前,核心的新类型为RDD-schemaRDD,现改为DataFrame。spark 通过DataFrame操作大量的数据源,包括外部文件(如 json、avro、parquet、sequencefile 等等)、Hive、关系数据库、cassandra等
本教程从最基础的Spark介绍开始,介绍Spark的各种部署模式以及动手进行搭建,然后逐步介绍其中RDD的计算模型,创建和常用的操作,以及其中一些分布式计算,R...
, 在“One Stack to rule them all”理念的指引下,Spark基于RDD成功地构建起了大数据处理的一体化解决方案,将MapReduce、Streaming、SQL、Machine Learning、Graph Processing等大数据计算模型统一到一个技术堆栈...
0.1配置可参考:0.2有关spark说明:spark不兼容Python3.6安装注意版本可下载:anaconda4.21.1数据student.txt1.2代码1.3结果展示2.1函数解析2.1.1collect()RDD的特性在进行基本RDD“转换”运算时不会立即执行,结果...
Spark中的(弹性分布式数据集)简称RDD: Spark中的Transformation操作之Value数据类型的算子: Spark中的Transformation操作之Key-Value数据类型的算子: Spark中的Action操作: Transformation->map算子: ...
一、前置知识详解 Spark SQL重要是操作DataFrame,...二、Spark SQL读写数据代码实战 import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkC
第4章 Spark RDD与编程API实战 第5章 Spark运行模式深入解析 第6章 Spark内核解析 第7章 GraphX大规模图计算与图挖掘实战 第8章 Spark SQL原理与实战 第9章 Machine Learning on Spark 第10章 Tachyon文件系统 第11...
通过学习Spark,我了解了其基本概念和核心组件,如RDD(弹性分布式数据集)、Spark SQL、Spark Streaming和MLlib等。我学会了使用Spark编写分布式的数据处理程序,并通过调优技巧提高了程序的性能。在实践过程中,我...
Spark大数据处理 技术、应用与性能优化.pdf Spark大数据分析实战.pdf Spark快速大数据分析.pdf 自己动手做大数据系统.pdf
通过学习Spark,我了解了其基本概念和核心组件,如RDD(弹性分布式数据集)、Spark SQL、Spark Streaming和MLlib等。我学会了使用Spark编写分布式的数据处理程序,并通过调优技巧提高了程序的性能。在实践过程中,我...