Spark——RDD创建

1 什么是RDD

RDD（Resilient Distributed Dataset）叫做弹性分布式数据集，是Spark中最基本的数据抽象。代码中是一个抽象类，它代表一个不可变、可分区、里面的元素可并行计算的集合。

分布式数据集

RDD是只读的、分区记录的集合，每个分区分布在集群的不同节点上

RDD并不存储真正的数据，只是对数据和操作的描述

弹性

RDD默认存放在内存中，当内存不足，Spark自动将RDD写入磁盘

容错性

根据数据血统，可以自动从节点失败中恢复分区

简单的解释
RDD是将数据项拆分为多个分区的集合，存储在集群的工作节点上的内存和磁盘中，并执行正确的操作
复杂的解释
RDD是用于数据转换的接口
RDD指向了存储在HDFS、Cassandra、HBase等、或缓存（内存、内存+磁盘、仅磁盘等），或在故障或缓存收回时重新计算其他RDD分区中的数据
           

2. RDD的创建

在Spark中创建RDD的创建方式可以分为三种，内存集合、外部、其他RDD算子操作后的返回值。

使用parallelize()从集合创建

默认按照setMaster中设置的核数进行分区，但是也可以在parallelize(…，x)中x指定分区数

object Spark02 {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称，使用本地模式开启两核
    val conf = new SparkConf().setMaster("local[2]").setAppName("myjob")
    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc = new SparkContext(conf)

    //虽然开了多核，但是只有一个分区，所以多核竞争一个分区，并不是并行运算
    val rdd1 = sc.parallelize(List(1, 2, 3, 4, 5, 2, 3), 1)
    rdd1.foreach(print)//1234523
    println()
    //本地开了2核，同时分了两个区，所以就会并行运算，打印的结果就是无序的，理论上这是上面运行速度的两倍
    val rdd2 = sc.parallelize(List(1, 2, 3, 4, 5, 2, 3), 2)
    rdd2.foreach(print)//1425323
    println()
    
    //默认用系统给的分区数，这里是两个
    val rdd3 = sc.parallelize(List(1, 2, 3, 4, 5, 2, 3))
    rdd3.foreach(print)//4152233
    println()
    print(rdd3.partitions.size)//2
  }
}
           

使用makeRDD()从集合创建

默认按照setMaster中设置的核数进行分区，但是也可以在makeRDD(…，x)中x指定分区数

object Spark03 {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf并设置App名称，使用本地模式开启两核
    val conf = new SparkConf().setMaster("local[2]").setAppName("myjob")
    //2.创建SparkContext，该对象是提交Spark App的入口
    val sc = new SparkContext(conf)

    //虽然开了多核，但是只有一个分区，所以多核竞争一个分区，并不是并行运算
    val rdd1 = sc.makeRDD(List(1, 2, 3, 4, 5, 2, 3), 1)
    rdd1.foreach(print)//1234523
    println()
    //本地开了2核，同时分了两个区，所以就会并行运算，打印的结果就是无序的，理论上这是上面运行速度的两倍
    val rdd2 = sc.makeRDD(List(1, 2, 3, 4, 5, 2, 3), 2)
    rdd2.foreach(print)//1425323

    println()
    //为每个集合对象创建一个最佳分区，这里是两个
    val rdd3 = sc.makeRDD(List(1, 2, 3, 4, 5, 2, 3))
    rdd3.foreach(print)//4152233
    println()
    print(rdd3.partitions.size)//2
  }
}
           

parallelize和makeRDD的区别

1. makeRDD的实现是依赖于parallelize的实现，两者一致；
2. makeRDD的底层就是调用的parallelize方法

由外部存储系统的数据集创建

包括本地的文件系统，还有所有Hadoop支持的数据集，比如HDFS、Cassandra、HBase等

默认只创建两个分区，也可以自己指定sc.textFile("",x)中x的数量

object WordCount {
  //WordCount
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("myjob")
    val sc = new SparkContext(conf)
    //sc读取本地文件也是创建RDD
    sc.textFile("file:///C:\\Users\\wuyanxiang\\study\\hadoop\\spark01\\src\\main\\scala\\njbdqn\\words")
    sc.stop()
  }
}