1.RDD详解
RDD的概念,注意事项已经在Spark的总体概述中说过,就不再提了,可以自行查看
https://blog.csdn.net/jiezou12138/article/details/88928265
1.1RDD的创建
得到rdd有三种方式
- 集合并行化
- 读取外部文件系统
- 其他方式
集合并行化
通过SparkContext
- java
sc.parallelize(xx,numpartition)
sc.parallelizePairs(seq[Tuple2<k,v>]) 将数据转换成K,V格式的RDD
- spark
sc.makeRDD(xx,numpartition)
sc.parallelize(xx,numpartition)
读取外部文件
- java:
sc.textFile(xxx,minNumPartitions)
- spark:
sc.textFile(xxx,minNumPartitions)
其他方式
读取数据库等等其他的操作。也可以生成RDD。
RDD可以通过transformation由其他的RDD转换而来的。
1.2 transformation算子和action算子
常用的Transformation:
转换 |
含义 |
map(func) |
返回一个新的RDD,该RDD由每一个输入元素经过func函数转换后组成 |
filter(func) |
返回一个新的RDD,该RDD由经过func函数计算后返回值为true的输入元素组成 |
flatMap(func) |
类似于map,但是每一个输入元素可以被映射为0或多个输出元素(所以func应该返回一个序列,而不是单一元素) |
mapPartitions(func) |
类似于map,但独立地在RDD的每一个分片上运行,因此在类型为T的RDD上运行时,func的函数类型必须是Iterator[T] => Iterator[U] |
mapPartitionsWithIndex(func) |
类似于mapPartitions,但func带有一个整数参数表示分片的索引值,因此在类型为T的RDD上运行时,func的函数类型必须是 (Int, Interator[T]) => Iterator[U] |
sample(withReplacement, fraction, seed) |
根据fraction指定的比例对数据进行采样,可以选择是否使用随机数进行替换,seed用于指定随机数生成器种子 |
union(otherDataset) |
对源RDD和参数RDD求并集后返回一个新的RDD |
intersection(otherDataset) |
对源RDD和参数RDD求交集后返回一个新的RDD |
distinct([numTasks])) |
对源RDD进行去重后返回一个新的RDD |
groupByKey([numTasks]) |
在一个(K,V)的RDD上调用,返回一个(K, Iterator[V])的RDD |
reduceByKey(func, [numTasks]) |
在一个(K,V)的RDD上调用,返回一个(K,V)的RDD,使用指定的reduce函数,将相同key的值聚合到一起,与groupByKey类似,reduce任务的个数可以通过第二个可选的参数来设置 |
aggregateByKey(zeroValue)(seqOp, combOp, [numTasks]) |
先按分区聚合 再总的聚合 每次要跟初始值交流 例如:aggregateByKey(0)(_+_,_+_) 对k/y的RDD进行操作 |
sortByKey([ascending], [numTasks]) |
在一个(K,V)的RDD上调用,K必须实现Ordered接口,返回一个按照key进行排序的(K,V)的RDD |
sortBy(func,[ascending], [numTasks]) |
与sortByKey类似,但是更灵活 第一个参数是根据什么排序 第二个是怎么排序 false倒序 第三个排序后分区数 默认与原RDD一样 |
join(otherDataset, [numTasks]) |
在类型为(K,V)和(K,W)的RDD上调用,返回一个相同key对应的所有元素对在一起的(K,(V,W))的RDD 相当于内连接(求交集) |
cogroup(otherDataset, [numTasks]) |
在类型为(K,V)和(K,W)的RDD上调用,返回一个(K,(Iterable<V>,Iterable<W>))类型的RDD |
cartesian(otherDataset) |
两个RDD的笛卡尔积 的成很多个K/V |
pipe(command, [envVars]) |
调用外部程序 |
coalesce(numPartitions) |
重新分区 第一个参数是要分多少区,第二个参数是否shuffle 默认false 少分区变多分区 true 多分区变少分区 false |
repartition(numPartitions) |
重新分区 必须shuffle 参数是要分多少区 少变多 |
repartitionAndSortWithinPartitions(partitioner) |
重新分区+排序 比先分区再排序效率高 对K/V的RDD进行操作 |
foldByKey(zeroValue)(seqOp) |
该函数用于K/V做折叠,合并处理 ,与aggregate类似 第一个括号的参数应用于每个V值 第二括号函数是聚合例如:_+_ |
combineByKey |
合并相同的key的值 rdd1.combineByKey(x => x, (a: Int, b: Int) => a + b, (m: Int, n: Int) => m + n) |
partitionBy(partitioner) |
对RDD进行分区 partitioner是分区器 例如new HashPartition(2 |
cache |
RDD缓存,可以避免重复计算从而减少时间,区别:cache内部调用了persist算子,cache默认就一个缓存级别MEMORY-ONLY ,而persist则可以选择缓存级别 |
persist |
|
|
|
Subtract(rdd) |
返回前rdd元素不在后rdd的rdd |
leftOuterJoin |
leftOuterJoin类似于SQL中的左外关联left outer join,返回结果以前面的RDD为主,关联不上的记录为空。只能用于两个RDD之间的关联,如果要多个RDD关联,多关联几次即可。 |
rightOuterJoin |
rightOuterJoin类似于SQL中的有外关联right outer join,返回结果以参数中的RDD为主,关联不上的记录为空。只能用于两个RDD之间的关联,如果要多个RDD关联,多关联几次即可 |
subtractByKey |
substractByKey和基本转换操作中的subtract类似只不过这里是针对K的,返回在主RDD中出现,并且不在otherRDD中出现的元素 |
常用的Action:
动作 |
含义 |
reduce(func) |
通过func函数聚集RDD中的所有元素,这个功能必须是课交换且可并联的 |
collect() |
在驱动程序中,以数组的形式返回数据集的所有元素 |
count() |
返回RDD的元素个数 |
first() |
返回RDD的第一个元素(类似于take(1)) |
take(n) |
返回一个由数据集的前n个元素组成的数组 |
takeSample(withReplacement,num, [seed]) |
返回一个数组,该数组由从数据集中随机采样的num个元素组成,可以选择是否用随机数替换不足的部分,seed用于指定随机数生成器种子 |
takeOrdered(n, [ordering]) |
|
saveAsTextFile(path) |
将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统,对于每个元素,Spark将会调用toString方法,将它装换为文件中的文本 |
saveAsSequenceFile(path) |
将数据集中的元素以Hadoop sequencefile的格式保存到指定的目录下,可以使HDFS或者其他Hadoop支持的文件系统。 |
saveAsObjectFile(path) |
|
countByKey() |
针对(K,V)类型的RDD,返回一个(K,Int)的map,表示每一个key对应的元素个数。 |
foreach(func) |
在数据集的每一个元素上,运行函数func进行更新。 |
aggregate |
先对分区进行操作,在总体操作 |
reduceByKeyLocally |
|
lookup |
|
top |
|
fold |
|
foreachPartition |
具体详细的可参考该博客 https://www.cnblogs.com/qingyunzong/p/8922135.html
1.3 RDD的宽依赖和窄依赖
RDD之间有一系列的依赖关系,依赖关系又分为窄依赖和宽依赖
窄依赖
父RDD与子RDD partition之间的关系是一对一
父RDD与子RDD partition之间的关系是多对一
宽依赖(shuffle)
父RDD与子RDD partition之间的关系是一对多
宽窄依赖图理解
依赖和容错
窄依赖,错误恢复快; 宽依赖,错误恢复慢
如果是窄依赖,只需要拿父rdd的对应分区的数据,+ 函数的业务逻辑 = 新的结果数据,恢复快
宽依赖: 必须拿父RDD中的所有的分区数据 + 业务逻辑 = 新的结果数据, 恢复慢
Driver: 会记录 rdd的依赖 ,rdd的操作的函数 等 executor端挂掉了。Driver会把运行在挂掉机器上的任务,重新分发到其他的executor中去运行。
2. Stage
2.1DAG的生成
DAG(Directed Acyclic Graph)叫做有向无环图,指任意一条边有方向,且不存在环路的图。
比如这个就是一个DAG
图的构成: 点 + 边
点: RDD
边: Dependency
那么这个图,在哪一个步骤生成的?
当触发action算子的时候,DAG图就生成了。
DAG的起始边界:
起点在哪里? 第一个rdd
终点在哪里? 触发action。
2.2stage
那怎么从DAG到stage的呢?
DAG -> 切分stage 有什么规则?
stage切割规则
切割规则:从后往前,遇到宽依赖就切割stage。
在spark中,会根据RDD之间的依赖关系将DAG图(有向无环图)划分为不同的阶段,对于窄依赖,由于partition依赖关系的确定性,partition的转换处理就可以在同一个线程里完成,窄依赖就被spark划分到同一个stage中,而对于宽依赖,只能等父RDD shuffle处理完成后,下一个stage才能开始接下来的计算。
因此spark划分stage的整体思路是:从后往前推,遇到宽依赖就断开,划分为一个stage;遇到窄依赖就将这个RDD加入该stage中。因此在图2中RDD C,RDD D,RDD E,RDDF被构建在一个stage中,RDD A被构建在一个单独的Stage中,而RDD B和RDD G又被构建在同一个stage中。
stage计算模式
pipeline管道计算模式,pipeline只是一种计算思想,模式。
- 数据一直在管道里面什么时候数据会落地
- 对RDD进行持久化。
- shuffle write的时候。
- Stage的task并行度是由stage的最后一个RDD的分区数来决定的 。
- 如何改变RDD的分区数?
例如:reduceByKey(XXX,3),GroupByKey(4)
stage是由一组并行的task组成。
测试验证pipeline计算模式
val conf = new SparkConf().setMaster("local").setAppName("pipeline");
val sc = new SparkContext(conf)
val rdd = sc.parallelize(Array(1,2,3,4))
val rdd1 = rdd.map { x => {
println("map--------"+x)
x
}}
val rdd2 = rdd1.filter { x => {
println("fliter********"+x)
true
} }
rdd2.collect()
sc.stop()