SparkCore的RDD

(1)RDD的介绍

   Spark Core 的RDD
 RDD(Resilient Distributed Dataset)叫做分布式数据集,是Spark中最基本的数据抽象,它代表一个不可变(RDD中的数据,不能增删改),可分区、元素可并行计算的集合。
 具有数据流的模型的特点,自动容错、位置感知性调度和可伸缩性。RDD允许用户在执行多个查询时显示的将工作集缓存在内存中。后续的查询能够重用工作集,这极大地提升了查询速度。
 RDD可以从 三方面理解:
   - 数据集:RDD是数据集合的抽象,是复杂物理介质上存在数据的一种逻辑视图。从外部看RDD的确可以被看待成经过封装,带扩展特性(如容错性)的数据集合。
   - 分布式:RDD的数据可能存储在多个节点的磁盘上或者内存中,也就是所谓的多级存储。
   - 弹性:虽然 RDD 内部存储的数据是只读的,但是,我们可以去修改(例如通 过 repartition 转换操作)并行计算计算单元的划分结构,也就是分区的数量。
 总之:RDD就是一个大集合,将所有的数据都加载到内存中,方便多次进行重用。它的数据可以在多个节点上,并且RDD可以保存在内存中,当如果某个阶段的RDD丢失,不需要重新计算,只需要提取上一次的RDD,在相应的计算即可。

成都创新互联公司是一家集网站设计、成都网站建设、网站页面设计、网站优化SEO优化为一体的专业网络公司,已为成都等多地近百家企业提供网站建设服务。追求良好的浏览体验,以探求精品塑造与理念升华,设计最适合用户的网站页面。 合作只是第一步,服务才是根本,我们始终坚持讲诚信,负责任的原则,为您进行细心、贴心、认真的服务,与众多客户在蓬勃发展的市场环境中,互促共生。

(2)RDD的属性

  Spark Core 的RDD

 1)A list of partitions(一组分片,数据集的基本单位)

  一个分区通常与一个任务向关联,分区的个数决定了并行的粒度。分区的个数可以在创建RDD的时候指定,如果不指定,那么默认的由节点的cores个数决定。最终每一个分区会被映射成为BlockManager 中的一个Block,而这个Block会被下一个task使用进行计算。

 2)A function for computing each split(算子)

  每一个RDD都会实现compute,用于分区进行计算

 3)A list of dependencies on other RDDs(RDD之间的依赖)

  RDD 的每次转换都会生成一个新的 RDD,所以 RDD 之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时,Spark 可以通过这个依赖关系重新计算丢失的分区数据, 而不是对 RDD 的所有分区进行重新计算。
宽依赖和窄依赖
Spark Core 的RDD
窄依赖(完全依赖):一个父分区唯一对应一个子分区,例:map操作
宽依赖(部分依赖):一个父分区对应多个子分区,如:reduce、group操作
区分宽依赖和窄依赖:当前这个算子的执行过程中是否有shuffle操作。

 4)Optionally a Partitioner for key-value RDDs(分区函数)

  当前 Spark 中实现了两种类型的分片函数,一个是基于哈希的 HashPartitioner,另外一个是基于范围的 RangePartitioner。只有对于 key-value 的 RDD,才会有 Partitioner,非 key-value的 RDD 的 Parititioner 的值是 None。Partitioner 函数不但决定了 RDD 本身的分片数量,也决 定了 parent RDD Shuffle 输出时的分片数量。

 5)Optionally a list of preferred locations to compute each split on

  一个列表,存储存取每个 Partition 的优先位置(preferred location)。按照”移动数据不如移动计算”的理念,Spark 在进行任务调度的时候,会尽可能地将计算任务分配到其所要处理数据块的存储位置。而这个列表中就存放着每个分区的优先位置。

(3)RDD的API(相关算子)

  RDD编程中有两种中形式:Transformation(转换)和Action(行动)。
 Transformation:表示把一个RDD ---->RDD。
 Action:表示把RDD----集合或者scala对象。

 1)RDD的创建:

object SparktTest {
    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf()
        conf.setAppName("SparktTest")
        conf.setMaster("local[2]")
        val sc: SparkContext = new SparkContext()
        //由一个已经存在的 Scala 数据集合创建
        val arr=Array(1,2,3,4)
        val arr1RDD: RDD[Int] = sc.parallelize(arr)
        val arr2RDD: RDD[Int] = sc.makeRDD(arr)

        //由外部存储系统的数据创建(HDFS、HBase...)
        val HDFSRDD: RDD[String] = sc.textFile("/data/input")
    }
}

 2)Transformation:

  官网:http://spark.apache.org/docs/latest/rdd-programming-guide.html#transformations
注意:RDD中所有的转换(Transformation)都是延迟加载,也就是说,他们并不是直接计算结果,相反的,他们只是记住这些应用到基础数据集,上的一个转换动作,只有当发生一个要求返回一个Driver动作的时候,这些转换才真正运行。

map()算子

        val HDFSRDD: RDD[String] = sc.textFile("/data/input")
        /**
          * map 算子,返回一个新的RDD,该RDD由每一个输入元素经过function函数转换后组成
          */
        val mapRDD: RDD[(String, Int)] = HDFSRDD.map(ele=>(ele,1))

flatMap()算子:

val arr=Array("hive hbase hadoop","spark hadoop","yarn hdfs")
        val lineRDD: RDD[String] = sc.parallelize(arr)
        /**
          * flagMap:类似于map,但是每一个元素输入的元素可以被
          * 映射成为0个或者多个输出的元素(返回的是一个序列,而不是单一的元素)
          */
        //返回一个集合hive hbase hadoop spark hadoop yarn hdfs
        val wordRDD: RDD[String] = lineRDD.flatMap(line=>line.split("\\s+"))

filter()算子:

        val arr=Array(1,2,3,4,5)
        val arrRDD: RDD[Int] = sc.parallelize(arr)
        /**
          * filter过滤:返回一个新的RDD,该RDD由经过func函数计算后返回
          * 值为true的输入元素组成
          */
        val filterRDD: RDD[Int] = arrRDD.filter(num=>num%2==0)

mapPartitions()算子:

        val hdfsRDD: RDD[String] = sc.textFile("/data/input")
        /**
          * mapPartitions与map的唯一区别就是,mapPartitions迭代的是一个分区,
          * 而map遍历的每一个元素,mapPartitions参数是一个迭代对象,返回的也是一个迭代对象
          */
        val partitionRDD: RDD[String] = hdfsRDD.mapPartitions((x: Iterator[String]) => {
            val temp = x.toList.map(line => line + "!")
            temp.toIterator
        })

mapPartitionsWithIndex()算子:

        val hdfsRDD: RDD[String] = sc.textFile("/data/input")
        /**
          * 第一个参数是分区编号:分区编号是从0开始的不间断的连续编号
          * 第二个参数和mapPartitions相同
          */
        val partitionRDD: RDD[String] = hdfsRDD.mapPartitionsWithIndex((parnum:Int,x: Iterator[String]) => {
            println(parnum) //分区编号
            val temp = x.toList.map(line => line + "!")
            temp.toIterator
        })

sample()算子:

        val list=1 to 5000
        /**
          * sample方法有三个参数:
          * withReplacement:代表是否有放回的抽取(false 不放回,true:放回)
          * fraction:抽取样本空间占总体的比例,(以分数的形式) 0<=fraction <=1
          * seed:随机数生成器,new Random().nextInt(10),不传表示使用系统的
          * 注意:我们使用的sample算子,不能保证提供集合大小就恰巧是rdd.size()*fraction,结果大小会上下浮动
          * sample在做抽样调查的时候,特别受用
          */
        val listRDD: RDD[Int] = sc.parallelize(list)
        val sampleRDD: RDD[Int] = listRDD.sample(false,0.2)
        println(sampleRDD.count())  //大概是5000*0.2 上下浮动

groupByKey()算子:

        val list=List(("math",18),("hbase",18),("hive",22),("hive",18))
        /**
          * groupByKey,分组
          * 建议groupByKey在实践中,能不用就不用,主要因为groupByKey的效率低,
          * 因为有大量的数据在网络中传输,而且还没有进行本地的预处理
          * 可以使用reduceByKey或者aggregateByKey或者combineByKey代替这个groupByKey
          */

        val stuRDD: RDD[(String, Int)] = sc.parallelize(list)
        //分组
        val groupRDD: RDD[(String, Iterable[Int])] = stuRDD.groupByKey()
        //求平均值
        val result: RDD[(String, Double)] = groupRDD.map { case (name, score) => {
            val avg = score.sum.toDouble / (score.size)
            (name, avg)
        }
        }

reduceByKey算子:

        val list=List(("math",18),("hbase",18),("hive",22),("hive",18))
        /**
          * reduceByKey:在一个(K,V)对的数据集上使用,返回一个(K,V)对的数据
          * 集,key 相同的值,都被使用指定的 reduce 函数聚合
          * 到一起。和 groupByKey 类似,任务的个数是可以通过
          * 第二个可选参数来配置的。
          */
        val stuRDD: RDD[(String, Int)] = sc.parallelize(list)
        //分组,求总分
        val sumRDD: RDD[(String, Int)] = stuRDD.reduceByKey((x, y)=>x+y)
        sumRDD.foreach(println) //打印:(hbase,36)(math,18)(hbase,18)

sortByKey()算子:

        val list=List(("math",18),("hbase",18),("hive",22),("hive",18))
        /**
          * sortByKey:在一个(K,V)的 RDD 上调用,K 必须实现 Ordered 接口,
          * 返回一个按照 key 进行排序的(K,V)的 RDD
          */

        //分组,求总分,排序
        val stuRDD: RDD[(String, Int)] = sc.parallelize(list)
        val sumRDD: RDD[(String, Int)] = stuRDD.reduceByKey((x, y)=>x+y)
        sumRDD.foreach(println) //打印:(hbase,36)(math,18)(hbase,18)
        val sortRDD: RDD[(String, Int)] = sumRDD.map(kv=>(kv._2,kv._1)).sortByKey().map(kv=>(kv._2,kv._1))
        sortRDD.foreach(println)

sortBy算子:

        val list=List(("math",18),("hbase",18),("hive",22),("hive",18))
        /**
          * sortBy(func,[ascending], [numTasks])
          * 与 sortByKey 类似,但是更灵活
          * 第一个参数是根据什么排序
          * 第二个是怎么排序,true 正序,false 倒序
          * 第三个排序后分区数,默认与原 RDD 一样
          */
        //分组,求总分,排序
        val stuRDD: RDD[(String, Int)] = sc.parallelize(list)
        val sumRDD: RDD[(String, Int)] = stuRDD.reduceByKey((x, y)=>x+y)
        sumRDD.foreach(println) //打印:(hbase,36)(math,18)(hbase,18)
        val sortRDD: RDD[(String, Int)] = sumRDD.sortBy(kv=>kv._2,false,2)

aggregateByKey()算子:

object SparktTest {
    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf()
        conf.setAppName("SparktTest")
        conf.setMaster("local[2]")
        val sc: SparkContext = new SparkContext()
        /**
          * aggregateByKey(zeroValue)(seqOp,combOp, [numTasks])
          * 先按分区聚合再总的聚合,每次要跟初始值交流
          * zeroValue:初始值
          * seqOp:迭代操作,拿RDD中的每一个元素跟初始值进行合并
          * combOp:分区结果的最终合并
          * numTasks:分区个数
          * aggregate+groupByKey=aggregateByKey
          * aggregate对单个值进行RDD,aggregateByKey对(K,V)值进行RDD
          */
        //aggregate
        val list = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
        val listRDD: RDD[Int] = sc.parallelize(list)
        //求平均值
        /**
          * seqOp: (U, T) => U
          * combOp: (U, U) => U
          * u:(Int,Int)   总和,总次数
          */
        val result: (Int, Int) = listRDD.aggregate(0, 0)((u: (Int, Int), x: Int) => {
            (u._1 + x, u._2 + 1)
        }
            , (u1: (Int, Int), u2: (Int, Int)) => {
                (u1._1 + u2._1, u1._2 + u2._2)
            })
        println(result._1 / result._2)

        //aggregateByKey已经根据(k,v)k 进行分组,以下的操作,是对v进行操作
        //以下操作时求平均值
        val list1 = List(("math", 18), ("hbase", 18), ("hive", 22), ("hive", 18))
        val stuRDD: RDD[(String, Int)] = sc.parallelize(list1)
        val reslutRDD2: RDD[(String, (Int, Int))] = stuRDD.aggregateByKey((0, 0))((x: (Int, Int), y: Int) => {
            (x._1 + y, x._2 + 1)
        }, (x: (Int, Int), y: (Int, Int)) => {
            (x._1 + y._1, x._2 + y._2)
        })
        reslutRDD2.foreach(kv=>{
            val name=kv._1
            val avg=kv._2._1.toDouble/kv._2._2
        })
    }
}

foldLeft()算子:(不是spark的算子,是scala的高级操作)

        /**
          *  foldLeft
          * (zeroValue: T)  初值值
          * (B, A) => B  B是一个元组,B._1 表示累加元素,B._2 表示个数, A 表示下一个元素
          */

        //aggregate
        val list = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
        val result: (Int, Int) = list.foldLeft((0,0))((x, y)=>{(x._1+y,x._2+1)})
        println(result._1.toDouble/result._2)

combineByKey()算子:

object SparktTest {
    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf()
        conf.setAppName("SparktTest")
        conf.setMaster("local[2]")
        val sc: SparkContext = new SparkContext(conf)
        /**
          * combineByKey:
          * 合并相同的 key 的值 rdd1.combineByKey(x => x, (a: Int,
          * b: Int) => a + b, (m: Int, n: Int) => m + n)
          */
        //求平均值
        val list1 = List(("math", 18), ("hbase", 18), ("hive", 22), ("hive", 18))
        val listRDD: RDD[(String, Int)] = sc.parallelize(list1)
        /**
          * createCombiner: V => C,
          * mergeValue: (C, V) => C,
          * mergeCombiners: (C, C) => C): RDD[(K, C)]
          */
        val resultRDD: RDD[(String, (Int, Int))] = listRDD.combineByKey(x => {
            (x, 1)
        },
            (x: (Int, Int), y: Int) => {
                (x._1 + y, x._2 + 1)
            },
            (x: (Int, Int), y: (Int, Int)) => {
                (x._1 + y._1, x._2 + y._2)
            })
        resultRDD.foreach{case (name,(sum,count))=>{
            val avg=sum.toDouble/count
            println(s"${name}:${avg}")
        }}
    }
}

连接操作

object SparktTest {
    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf()
        conf.setAppName("SparktTest")
        conf.setMaster("local[2]")
        val sc: SparkContext = new SparkContext(conf)
        val arr1 = Array(1, 2, 4, 5)
        val arr1RDD = sc.parallelize(arr1)
        val arr2 = Array(4, 5, 6, 7)
        val arr2RDD = sc.parallelize(arr2)
        //cartesian  笛卡尔积
        val cartesianRDD: RDD[(Int, Int)] = arr1RDD.cartesian(arr2RDD)
        //union : 连接
        val unionRDD: RDD[Int] = arr1RDD.union(arr2RDD)
        //subtract,求,差集
        val sbutractRDD: RDD[Int] = arr1RDD.subtract(arr2RDD)

        //join
        val list1 = List(("a", 1), ("b", 2), ("c", 3))
        val list1RDD = sc.parallelize(list1)
        val list2 = List(("a", "zs"), ("b", "sl"))
        val list2RDD = sc.parallelize(list2)
        /**
          * 根据元组中的key进行join 操作,相同的key向连接
          * 返回的是RDD[(String, (Int, String))] (key,连接结果)
          */
        val joinRDD: RDD[(String, (Int, String))] = list1RDD.join(list2RDD)

        //cogroup
        /**
          * (String key   ,
          * (Iterable[Int] arr1中的相应的key所有value的集合
          * , Iterable[String]))  arr2中的相应的key所有value的集合
          */
        val cogroupRDD: RDD[(String, (Iterable[Int], Iterable[String]))] = list1RDD.cogroup(list2RDD)
    }
}

分区操作

object SparktTest {
    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf()
        conf.setAppName("SparktTest")
        conf.setMaster("local[2]")
        val sc: SparkContext = new SparkContext(conf)
        val hdfsRDD: RDD[String] = sc.textFile("/data/word.txt")
        /**
          * 表示在执行了filter操作之后,由于大量的数据被过滤,导致之前设定的分区task个数,
          * 处理剩下的数据导致资源浪费,为了合理高效的利用资源,
          * 可以对task重新定义,在coalesce方法中的分区个数一定要小于之前设置的分区个数。
          */
        hdfsRDD.coalesce(2)
        //打乱数据,重新分区,分区规则为随机分区
        hdfsRDD.repartition(3)

        //自定义分区规则(注意,只在有key-value的RDD中可以使用)
        var arr1 = Array(("a", 1), ("a", 2), ("c", 1), ("b", 2), ("d", 2)
            ("b", 2), ("e", 2)
            , ("b", 2)
            , ("f", 2), ("g", 2), ("h", 2))
        val arrRDD: RDD[(String, Int)] = sc.parallelize(arr1,4)
        arrRDD.partitionBy(new MyPartitioner(3))

    }
}
class MyPartitioner(val numPTN:Int) extends Partitioner{
    //分区个数
    override def numPartitions: Int = numPTN
    //分区规则
    override def getPartition(key: Any): Int = {
        val num=key.hashCode()&Integer.MAX_VALUE%numPTN
        return num
    }
}

总结
   - Transformation返回的仍然是一个RDD
   - 它使用了链式调用的设计模式,对一个 RDD 进行计 算后,变换成另外一个 RDD,然后这个 RDD 又可以进行另外一次转换。这个过程是分布式的。

 3)Action:

常见操作

object SparktTest {
    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf()
        conf.setAppName("SparktTest")
        conf.setMaster("local[2]")
        val sc: SparkContext = new SparkContext(conf)
        val list = List(("math", 18), ("hbase", 18), ("hive", 22), ("hive", 18))
        val listRDD: RDD[(String, Int)] = sc.parallelize(list)
        //action  rdd ---map
        listRDD.reduceByKeyLocally((x,y)=>x+y)

        //调用collect的目的是:触发所有的计算,最终收集当前这个调用者RDD的所有数据,返回到客户端,如果数据量比较大,谨慎使用
        listRDD.collect()

        //统计RDD中有多少记录
        listRDD.count()
        //取出RDD中的第一条记录
        listRDD.first()
        //取出RDD前几条记录
        listRDD.take(5)
        //随机采样
        listRDD.takeSample(false,20)
        //按照某种格式,排序后的前几条
        listRDD.top(50)
        //按照升序或者降序,取相应的条数的记录(其中的元素必须继承Ordered)
        listRDD.takeOrdered(3)
        //统计每一个key中的value有多少个
        listRDD.countByKey()
        //统计有多少个元素
        listRDD.countByValue()
        //遍历RDD中每一个元素
        listRDD.foreach(kv=>{})
        //分区遍历RDD中的元素
        listRDD.foreachPartition(kv=>{})
        //将RDD的结果,保存到相应的文件系统中(注意这个目录一定是不存在的目录)
        listRDD.saveAsTextFile("/data/output")
    }
}

总结:Action返回值不是一个RDD。它要么是一个scala的集合,要么是一个值,要么是空。最终返回到Driver程序,或者把RDD写入到文件系统中。


分享题目:SparkCore的RDD
文章路径:http://scyanting.com/article/pcgcsc.html