资讯

精准传达 • 有效沟通

从品牌网站建设到网络营销策划,从策略到执行的一站式服务

Spark常用的action算子-创新互联

action算子简介

Action类算子也是一类算子(函数)叫做行动算子,如foreach,collect,count等。Transformations类算子是延迟执行,Action类算子是触发执行。一个application应用程序(就是我们编写的一个应用程序)中有几个Action类算子执行,就有几个job运行。

创新互联是一家集网站建设,武平企业网站建设,武平品牌网站建设,网站定制,武平网站建设报价,网络营销,网络优化,武平网站推广为一体的创新建站企业,帮助传统企业提升企业形象加强企业竞争力。可充分满足这一群体相比中小企业更为丰富、高端、多元的互联网需求。同时我们时刻保持专业、时尚、前沿,时刻以成就客户成长自我,坚持不断学习、思考、沉淀、净化自己,让我们为更多的企业打造出实用型网站。

1.reduce

通过函数func聚集数据集中的所有元素,这个函数必须是关联性的,确保可以被正确的并发执行

scala> val rdd1 = sc.makeRDD(1 to 10)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[3] at makeRDD at :24

scala> rdd1.reduce(_+_)
res3: Int = 55

2.collect

在driver的程序中,以数组的形式,返回数据集的所有元素,这通常会在使用filter或者其它操作后,返回一个足够小的数据子集再使用

scala> var rdd1 = sc.makeRDD(1 to 10)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[2] at makeRDD at :24

scala> rdd1.collect
res2: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)

3.count

返回数据集的元素个数

scala> val rdd1 = sc.makeRDD(1 to 10)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[4] at makeRDD at :24

scala> rdd1.count
res4: Long = 10

4.first

返回数据集的第一个元素(类似于take(1))

scala> val rdd1 = sc.makeRDD(1 to 10)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[5] at makeRDD at :24

scala> rdd1.first
res5: Int = 1

5.take

返回一个数组,由数据集的前n个元素组成。注意此操作目前并非并行执行的,而是driver程序所在机器

scala> val rdd1 = sc.makeRDD(1 to 10)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[7] at makeRDD at :24

scala> rdd1.take(3)
res6: Array[Int] = Array(1, 2, 3)

6.takeSample(withReplacement,num,seed)

withReplacement:结果中是否可重复
num:取多少个
seed:随机种子
返回一个数组,在数据集中随机采样num个元素组成,可以选择是否用随机数替换不足的部分,seed用于指定的随机数生成器种子
原理
takeSample()函数和sample函数是一个原理,但是不使用相对比例采样,而是按设定的采样个数进行采样,同时返回结果不再是RDD,而是相当于对采样后的数据进行collect(),返回结果的集合为单机的数组

scala> val rdd1 = sc.makeRDD(1 to 10)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[20] at makeRDD at :24

scala> rdd1.takeSample(true,4,10)
res19: Array[Int] = Array(10, 10, 2, 3)

7.takeOrdered

takeOrdered和top类似,只不过以和top相反的顺序返回元素。
top默认倒序,taskOrdered默认正序
top方法其实就是调用的taskOrdered,然后反转的结果

def top(num: Int)(implicit ord: Ordering[T]): Array[T] = withScope {
    takeOrdered(num)(ord.reverse)
  }
scala> val rdd1 = sc.makeRDD(1 to 10)
rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[23] at makeRDD at :24

scala> rdd1.top(5)
res22: Array[Int] = Array(10, 9, 8, 7, 6)

scala> rdd1.takeOrdered(5)
res23: Array[Int] = Array(1, 2, 3, 4, 5)

8.saveAsTextFile

saveAsTextFile用于将RDD以文本文件的格式存储到文件系统中

val conf = new SparkConf()
      .setAppName("saveFile")
      .setMaster("local[*]")

val sc = new SparkContext(conf)

val rdd1: RDD[Int] = sc.parallelize(1 to 10)

rdd1.repartition(1).saveAsTextFile("/tmp/fff")

9.saveAsSequenceFile

saveAsSequenceFile用于将RDD以SequenceFile的文件格式保存到HDFS上。使用方法和saveAsTextFile类似

10.saveAsObjectFile

saveAsObjectFile用于将RDD中的元素序列化成对象,存储到文件中。使用方法和saveAsTextFile类似

11.countByKey

对(K,V)类型的RDD有效,返回一个(K,Int)对的map,表示每一个可以对应的元素个数

scala> val rdd1 = sc.makeRDD(Array(("A",0),("A",2),("B",1),("B",2),("C",3)))
rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[3] at makeRDD at :24

scala> rdd1.countByKey
res1: scala.collection.Map[String,Long] = Map(B -> 2, A -> 2, C -> 1)

12.foreach

在数据集的每一个元素上,运行函数func,t通常用于更新一个累加器变量,或者和外部存储系统做交互

scala> val rdd1 = sc.makeRDD(Array(("A",0),("A",2),("B",1),("B",2),("C",3)))
rdd1: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[9] at makeRDD at :24

scala> rdd1.collect.foreach(println(_))
(A,0)
(A,2)
(B,1)
(B,2)
(C,3)

另外有需要云服务器可以了解下创新互联scvps.cn,海内外云服务器15元起步,三天无理由+7*72小时售后在线,公司持有idc许可证,提供“云服务器、裸金属服务器、高防服务器、香港服务器、美国服务器、虚拟主机、免备案服务器”等云主机租用服务以及企业上云的综合解决方案,具有“安全稳定、简单易用、服务可用性高、性价比高”等特点与优势,专为企业上云打造定制,能够满足用户丰富、多元化的应用场景需求。


网站栏目:Spark常用的action算子-创新互联
标题网址:http://www.cdkjz.cn/article/dcjigh.html
多年建站经验

多一份参考,总有益处

联系快上网,免费获得专属《策划方案》及报价

咨询相关问题或预约面谈,可以通过以下方式与我们联系

大客户专线   成都:13518219792   座机:028-86922220