• Resilient Distributed DataSet：弹性分布式数据集Spark最基本的数据抽象可以理解为数据集合。
  • Dataset：收集数据，存储大量数据。
  • Distributed：RDD分布式存储内部元素，便于后期分布式计算。
  • Resilient：表示弹性，RDD数据可以保存在内存或磁盘中。
• 它是代码中的抽象类，代表弹性、不可变、可分区，内部元素可并行计算。

• ① A list of partitions(分区性)
  • RDD分区很多，每个分区都包括在内RDD的部分数据
  • 因为有多个分区，所以一个分区(Partition)列表可视为数据集的基本组成部分
  • spark中任务是以task对于线程运行，RDD每个分区都将由一个计算任务处理， 一个分区对应一个task因此，分区决定了并行计算的粒度。
  • 可创建用户RDD时，指定RDD如果没有指定分区数，则采用默认值(程序分配的CPU Core的数目)。
  • 每个分配的存储是由的BlockManager每个分区都会实现逻辑映射成BlockManager的一个Block，而这个Block会被一个Task负责计算。
• ② A function for computing each split(计算每个分区的函数)
  • Spark中RDD计算以分区为单位，每个RDD都会实现compute实现这一目标的函数
• ③ A list of dependencies on other RDDs(依赖:一个rdd它将依赖于其他多个rdd)
  • spark根据这一特点，任务的容错机制是（血统）而来。
  • RDD每一个转换都会产生一个新的RDD，所以RDD在某些区域数据丢失时，会形成类似流水线的前后依赖关系，Spark丢失的分区数据可以通过这种依赖关系重新计算，而不是对RDD重新计算所有分区。
• ④ Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)–(对存储键值对RDD，还有一个可选的分区器)
  • 只有对于key-value的RDD(RDD[(String, Int)])，并且产生shuffle，才会有Partitioner。
  • 非key-value的RDD(RDD[String])的Parititioner的值是None。
  • Partitioner不但决定了RDD也决定了分区的数量parent RDD Shuffle输出时的分区数量
  • 当前Spark实现了两种分区函数
    • 以哈希为基础HashPartitioner，(key.hashcode % 分区数 = 分区号)，是默认值。
    • 另一个是基于范围的RangePartitioner。
• ⑤ Optionally, a list of preferred locations to compute each split on (e.g. block locations for an HDFS file)–(存储各分区优先位置列表(本地计算)
  • 比如一个HDFS就文件而言，这个列表保存每个列表Partition所在文件块的位置。
  • 按照移动数据不如移动计算的理念，Spark在任务调度过程中，计算任务将尽可能分配到要处理的数据块的存储位置，以减少数据的网络传输，提高计算效率。

• 需求：HDFS上面有一个3000的大小M文件，通过spark实现文件单词统计，最终保存结果数据HDFS上。
• 代码实现：

sc.textFile("/words.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).saveAsTextFile("/out") 

• 流程分析：

• Transformation 算子，也称转换算子，在没有 Action 算子(也称行动算子)去触发的时候，是不会执行的，可以理解为懒算子，而Action算子可以理解为触发算子。
• 还有一种 Shuffle类 算子，就是上面说到的洗牌算子。

• 去重

def distinct()
def distinct(numPartitions: Int)

• 聚合

def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)]
def reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)]
def groupBy[K](f: T => K, p: Partitioner):RDD[(K, Iterable[V])]
def groupByKey(partitioner: Partitioner):RDD[(K, Iterable[V])]
def aggregateByKey[U: ClassTag](zeroValue: U, partitioner: Partitioner): RDD[(K, U)]
def aggregateByKey[U: ClassTag](zeroValue: U, numPartitions: Int): RDD[(K, U)]
def combineByKey[C](createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C): RDD[(K, C)]
def combineByKey[C](createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C, numPartitions: Int): RDD[(K, C)]
def combineByKey[C](createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C, partitioner: Partitioner, mapSideCombine: Boolean = true, serializer: Serializer = null): RDD[(K, C)]

• 排序

def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.length): RDD[(K, V)]
def sortBy[K](f: (T) => K, ascending: Boolean = true, numPartitions: Int = this.partitions.length)(implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]

• 重分区

def coalesce(numPartitions: Int, shuffle: Boolean = false, partitionCoalescer: Option[PartitionCoalescer] = Option.empty)
def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null)

• 集合或者表操作

def intersection(other: RDD[T]): RDD[T]
def intersection(other: RDD[T], partitioner: Partitioner)(implicit ord: Ordering[T] = null): RDD[T]
def intersection(other: RDD[T], numPartitions: Int): RDD[T]
def subtract(other: RDD[T], numPartitions: Int): RDD[T]
def subtract(other: RDD[T], p: Partitioner)(implicit ord: Ordering[T] = null): RDD[T]
def subtractByKey[W: ClassTag](other: RDD[(K, W)]): RDD[(K, V)]
def subtractByKey[W: ClassTag](other: RDD[(K, W)], numPartitions: Int): RDD[(K, V)]
def subtractByKey[W: ClassTag](other: RDD[(K, W)], p: Partitioner): RDD[(K, V)]
def join[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (V, W))]
def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]
def join[W](other: RDD[(K, W)], numPartitions: Int): RDD[(K, (V, W))]
def leftOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (V, Option[W]))]

object RddDemo { 
        
  def main(args: Array[String]): Unit = { 
        
    val conf = new SparkConf().setAppName(RddDemo.getClass.getSimpleName)
      .setMaster("local[*]")
    val sc = new SparkContext(conf)
    /* 传入分区数为1，结果顺序打印；传入分区数大于1，结果顺序不定，因为数据被打散在N个分区里 */
    val rdd1: RDD[Int] = sc.parallelize(1.to(10), 1)
    print("rdd1->:")
    rdd1.foreach(x => print(x + "\t"))
    println("")

    val rdd2 = sc.parallelize(List(1, 2, 3, 4, 5), 2)
    print("rdd2->:")
    rdd2.foreach(x => print(x + "\t"))
    println("")

    val rdd3 = sc.parallelize(Array("hadoop", "hive", "spark"), 1)
    print("rdd3->:")
    rdd3.foreach(x => print(x + "\t"))
    println("")

    val rdd4 = sc.makeRDD(List(1, 2, 3, 4), 2)
    print("rdd4->:")
    rdd4.foreach(x => print(x + "\t"))
    println("")

    sc.stop()
  }
}

• 比如：WordCount 案例，读取 textFile 就是这种用法。
• 读取 json 文件

object RddDemo { 
        
  def main(args: Array[String]): Unit = { 
        
    val conf = new SparkConf().setAppName(RddDemo.getClass.getSimpleName)
      .setMaster("local[*]")
    val sc = new SparkContext(conf)

    val rdd1: RDD[String] = sc.textFile(this.getClass.getClassLoader.getResource("word.json").getPath)
    val rdd2: RDD[Option[Any]] = rdd1.map(JSON.parseFull(_))
    rdd2.foreach(println)
  }
}

• 读取 Object 对象文件

object RddDemo { 
        
    def main(args: Array[String]): Unit = { 
        
    val conf = new SparkConf().setAppName(RddDemo.getClass.getSimpleName)
      .setMaster("local[*]")
    val sc = new SparkContext(conf)

    val rdd1: RDD[Int] = sc.parallelize(Array(1,2,3,4,5))
    rdd1.saveAsObjectFile("hdfs://node01:8020/test")

    val rdd2: RDD[Nothing] = sc.objectFile("hdfs://node01:8020/test")
    rdd2.foreach(println)
  }
}

• 案例一：

object RddDemo { 
        
  def main(args: Array[String]): Unit = { 
        
    val conf = new SparkConf().setAppName(RddDemo.getClass.getSimpleName)
      .setMaster("local[*]")
    val sc = new SparkContext(conf)

    /* map算子，一共有多少元数会执行多少次，和分区数无关，可以修改分区数进行测试 */
    val rdd: RDD[Int] = sc.parallelize(1.to(5), 1)
    val mapRdd: RDD[Int] = rdd.map(x => { 
        
      println("执行") // 一共被执行5次
      x * 2
    })
    val result: Array[Int] = mapRdd.collect()
    result.foreach(x => print(x + "\t"))
  }
}

// 结果输出
执行
执行
执行
执行
执行
2	4	6	8	10

• 案例二：

object RddDemo { 
        
  def main(args: Array[String]): Unit = { 
        
    val conf = new SparkConf().setAppName(RddDemo.getClass.getSimpleName)
      .setMaster("local[*]")
    val sc = new SparkContext(conf)

    /* mapPartitions算子，一个分区内处理，有几个分区就执行几次，优于map函数 常用于时间转换，数据库连接 */
    val rdd: RDD[Int] = sc.parallelize(1.to(10), 2)
    val mapRdd: RDD[Int] = rdd.mapPartitions(it => { 
        
      println("执行") // 分区2次，共打印2次
      it.map(x => x * 2)
    })
    val result: Array[Int] = mapRdd.collect()
    result.foreach(x => print(x + "\t"))
  }
}

// 结果输出
执行
执行
2	4	6	8	10	12	14	16	18	20

• 案例三：

object RddDemo { 
        
  def main(args: Array[String]): Unit = { 
        
    val conf = new SparkConf().setAppName(RddDemo.getClass.getSimpleName)
      .setMaster("local[*]")
    val sc = new SparkContext(conf)

    /* mapPartitionsWithIndex算子，一个分区内处理，有几个分区就执行几次 返回带有分区号的结果集 */
    val rdd: RDD[Int] = sc.parallelize(1.to(10), 2)
    val mapRdd: RDD[(Int, Int)] = rdd.mapPartitionsWithIndex((idx, it) => { 
        
      println("执行") // 执行两次
      it.map((idx, _))
    })

    val result: Array[(Int, Int)] = mapRdd.collect()
    result.foreach(x => print(x + "\t"))
  }
}

// 结果输出
执行
执行
(0,1)	(0,2)	(0,3)	(0,4)	(0,5)	(1,6)	(1,7)	(1,8)	(1,9)	(1,10)