Spark Core 的RDD

(1)RDD的介绍

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

(2)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 的所有分区进行重新计算。
宽依赖和窄依赖：

窄依赖（完全依赖）：一个父分区唯一对应一个子分区，例：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写入到文件系统中。