美文网首页
spark常见行动算子

spark常见行动算子

作者: pkingdog | 来源:发表于2019-01-24 16:35 被阅读0次

    reduce(f:(T,T)=>T)

    1. 作用:通过func函数聚集RDD中的所有元素,先聚合分区内数据,再聚合分区间数据

    2.创建一个RDD针对里面的元素进行累加计算获取结果

    val rdd=sc.makeRDD(1 to 10)

    rdd.reduce(_ + _)

    count()

    1.作用:在驱动程序中,以数组的形式返回数据集的所有元素

    2. 需求:创建一个RDD,并将RDD内容收集到Driver端打印

    val rdd=sc.makeRDD(1 to 10) 

    rdd.count()

    结果:

    10

    first()

    1. 作用:返回RDD中的第一个元素

    2.创建一个RDD,获取里面的第一个元素调用first函数返回第一个元祖,底层调用的就是take()函数

    val rdd2=sc.makeRDD(Arrray(("qiuxiang",1),("tangbohu",3),("huangrong",3)))

    rdd2.first()

    结果:

    ("qiuxiang",1)

    take(num: Int)

    1. 作用:返回一个由RDD的前n个元素组成的数组,

    2.创建一个RDD,获取里面的第一个元素调用first函数返回第一个元祖

    val rdd2=sc.makeRDD(Arrray(("qiuxiang",1),("tangbohu",3),("huangrong",3)))

    rdd2.take(3)

    结果:

    ("qiuxiang",1),("tangbohu",3),("huangrong",3


    takeOrdered(num:Int)&& top(num:Int)

    1. 作用:返回该RDD排序后的前n个元素组成的数组,内部涉及排序默认是升序,top是降序

    2.创建一个RDD,获取里面的第一个元素调用takeOrdered算子

    val rdd=sc.makeRDD(Array(1,90,23,10,11,88,23,-1))

    rdd.takeOrdered(3)

    rdd.top(3)

    结果:

    Aarray(-1, 1, 10)

    Array(90, 88, 23)

    aggregat[U: ClassTag]((zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U))

    0.参数(zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U)

    1.作用:aggregate函数将每个分区里面的元素通过seqOp和初始值进行聚合(即第一个参数的作用),然后用combine函数将每个分区的结果和初始值(zeroValue)进行combine操作。这个函数最终返回的类型不需要和RDD中元素类型一致 为了更好的理解这个这个算子的作用我们下面进行一个简单的操作

    2.需求:创建一个RDD指定RDD由Array(num:Int),2个分区组成计算这个RDD元素的值

    val rdd=sc.makeRDD(Array(10,29,90,11,24,10),2)

    rdd.aggregate(0)((x:Int,y:Int)=>(x+y),(x:Int,y:Int)=>(x+y))

    结果 :174

    上面结果其实没有问题但是如果将初始值改成10的话

    rdd.aggregate(10)((x:Int,y:Int)=>(x+y),(x:Int,y:Int)=>(x+y))

    结果:204

    204的计算结果是是怎么算出来的呢?

    我们这里计算的有两个分区aggregat()算子,是先分区内计算,再分区之间计算,然后再分区数计算

    p0    (10,20,90)   p0分区内的数据计算结果是10+10+20+90=139

    p1     (11,24,10  )   p1分区内的数据计算结果是: 10+11+24+10=55

    p0    p1  两个分区 计算结果为:10

    最终结果是:139+55+10=204


    fold(zeroValue:T)(op: (T, T) =>T)

    1. 作用:折叠操作,aggregate的简化操作,seqop和combop一样

    2.val rdd=sc.makeRDD(Array(10,29,90,11,24,10),2)

    rdd.fold(0)(_+_)

    结果:174

    saveAsTextFile(path:String)

    1.作用:将数据集的元素以textfile的形式保存到HDFS文件系统或者其他支持的文件系统,对于每个元素,Spark将会调用toString方法,将它装换为文件中的文本

    saveAsSequenceFile(path:String)

    作用:将数据集中的元素以Hadoop sequencefile的格式保存到指定的目录下,可以使HDFS或者其他Hadoop支持的文件系统

    saveAsObjectFile(path:String)

    作用:用于将RDD中的元素序列化成对象,存储到文件中

    countByKey()

    1. 作用:针对(K,V)类型的RDD,返回一个(K,Int)的map,表示每一个key对应的元素个数。

    2.创建一个PairRDD,统计每种key的个数

    val rdd= sc.makRDD(List((1,3),(1,2),(1,4),(2,3),(3,6),(3,8)),3)

     rdd.countByKey

    结果:Map(3 -> 2, 1 -> 3, 2 -> 1)

    foreach[U](f:A =>U)

    1. 作用:在数据集的每一个元素上,运行函数func进行更新

    需求:创建一个RDD,对每个元素进行打印

    val rdd= sc.makRDD(List((1,3),(1,2),(1,4),(2,3),(3,6),(3,8)),3)

    rdd.foreach(t=>{print(t._1+"--"+t._2)})

    结果:1--3 1--2 1--4 2--3 3--6 3--8

    相关文章

      网友评论

          本文标题:spark常见行动算子

          本文链接:https://www.haomeiwen.com/subject/ihkyjqtx.html