文章目录

• 行动操作

• 
  

• Action算子概念

• Action函数

行动操作

在spark当中RDD的操作可以分为两种一种是转化操作(transformation另一种是行动操作(action)。在转化操作当中spark不会为我们计算结果而是会生成一个新的RDD节点记录下这个操作。只有在行动操作执行的时候spark才会从头开始计算整个计算。

而转化操作又可以进一步分为针对元素的转化操作以及针对集合的转化操作。和转化操作API不同的是行动操作API只能作用于一个RDD。

Action算子概念

Action 用来触发RDD的计算得到相关计算结果
Action触发Job。一个Spark程序(Driver程序)包含了多少 Action 算子那么
就有多少Job&
典型的Action算子: collect / count
collect() => sc.runJob() => … => dagScheduler.runJob() => 触发了Job

Action函数

以数据rdd{1,2,3,3}为例进行RDD转换如下

函数	举例	运行结果	解释说明
collect	rdd.collect()	{1,2,3,3}	将RDD数据转换成数组
count	rdd.count()	4	对RDD中元素进行统计
take(num)	rdd.take(3)	{1,2,3}	返回RDD中从0到(num-1)下标的元素不进行排序结果中返回rdd下标为0,1,2是元素
top(num)	rdd.top(3)	{3,3,2}	对RDD数据按照默认降序或者指定的排列顺序返回num个元素
takeOrdered(num)	rdd.takeOrdered(3)	{1,2,3}	与top函数类似但是与top顺序相反按照升序返回num个元素
takeSample	rdd.takeSample(false,2)	随机两个元素组合的新RDD	随机生成一个数组包含num个元素的RDD例如{2,3}或者{1,3}
reduce	rdd.reduce((x,y)=>x+y)	9	根据映射函数f=x+y对RDD中的元素进行加法其他二元计算返回计算结果
fold(zero)(func)	rdd.fold(0)((x,y)=>x+y)	9	将每个元素累加求和与reduce一样但是需要提供初始化值
foreach	rdd.foreach(print)	{1,2,3,3}	foreach是对每个元素使用的指定函数例如print进行操作
countByValue	rdd.countByValue()	Map(1 -> 1, 2 -> 1, 3 -> 2)	各结果在RDD中出现的次数
aggregate(zeroValue)(seq0p,comb0p)	rdd.aggregate((0,0))((x,y) => (x._1 + y,x._2+1),(x,y) => (x._1 + y._1,x._2+y._2))	(9,4)	aggregate(zeroValue)(seq0p,comb0p) 和reduce类似但是通常返回不同的数据
first	rdd.first	1	first返回RDD中的第一个元素不排序

注意aggregate函数中

• 第一个函数的作用是把RDD中的元素合并起来放入累加器；

• 第二个函数的作用是考虑到每个节点是在本地进行累加的最后还需要提供一个函数将两个累加器进行两两和合并。

collect()函数
返回RDD中的所有元素

val testList = List(1, 2, 3, 3)val testRdd = sc.parallelize(testList)testRdd.collect().foreach(ele => print(s"$ele "))// 输出1 2 3 3

注意该函数会将所有元素放入驱动器进程中只有当整个数据集能在单台机器的内存中放得下时才能使用不适宜用在大规模数据集因此大多数情况下用于本地开发测试以下还会碰到很多这样不适合大数据量访问的函数

指定顺序排序

implicit val myOrd = implicitly[Ordering[Int]].reverse
rdd.top(1)rdd.top(2)

保存 rdd 到 hdfssaveAsTextFile

rdd.saveAsTextFile("hdfs://localhost:9000/user/hadoop/Output")