创建RDD的常用方式【并行化集合创建、从外部存储系统创建、RDD衍生、小文件读取】，创建SparkContext对象的方式

创建RDD的常用方式【并行化集合创建、从外部存储系统创建、RDD衍生、小文件读取】，创建SparkContext对象的方式

文章目录

• 创建 SparkContext 对象
• RDD 创建方式
• • 1. 创建并行化集合RDD（parallelize、makeRDD）
  • • parallelize
    • makeRDD
  • 2. 从外部存储系统创建RDD（textfile）
  • 3. 从RDD衍生新的RDD（原地计算）
  • 4. 小文件读取（wholeTextFiles）
• RDD 分区数目（总结）
• • 获取 RDD 分区数目的俩种方式
  • RDD 分区的数据取决于那些因素？

---

创建 SparkContext 对象

SparkApplication应用，创建SparkContext对象，及应用结束关闭资源，代码如下：

import org.apache.spark.{SparkConf, SparkContext}

object MydemoTest {

  def main(args: Array[String]): Unit = {
    // 构建SparkContext对象
    val sc: SparkContext = {
      // a. 创建SparkConf对象
      val sparkConf = new SparkConf()
        .setAppName(this.getClass.getSimpleName.stripSuffix("$"))
        .setMaster("local[2]")
      // b. 传递SparkConf对象，创建实例
      val context = SparkContext.getOrCreate(sparkConf) //有就获取，没有创建
      // c. 返回实例对象
      context
    }


    // 应用结束，关闭资源
    sc.stop()
  }

}

或者写的简单点

val conf = new SparkConf().setAppName("word_count").setMaster("local[6]")
val sc = new SparkContext(conf)

---

RDD 创建方式 1. 创建并行化集合RDD（parallelize、makeRDD） parallelize

源码：

def parallelize[T: ClassTag](
  seq: Seq[T],
  numSlices: Int = defaultParallelism
): RDD[T]

需要传入的是一个Seq集合，和一个分区数Int

---

makeRDD

源码结构和parallelize一样，还是调用的parallelize。

两者区别：（源码的区别）

• parallelize可以不指定分区数，使用比较多。
• makeRDD不指定就会调用成别的同名方法，可以看下源码来验证，一般不用。

上述俩种方式的写法如下：

//从本地集合创建
@Test
def rddCreationLocal() = {
  val seq = Seq("Hello1", "hello2", "Hello3")
  val rdd1: RDD[String] = sc.parallelize(seq, 2)  //指定分区数目
  val rdd2: RDD[String] = sc.makeRDD(seq, 2) //底层依旧调用的parallelize
}

---

2. 从外部存储系统创建RDD（textfile）

由外部存储系统的数据集创建，包括本地的文件系统，还有所有 Hadoop支持的数据集，比如HDFS、Cassandra、Hbase 等。实际使用最多的方法：textFile，读取HDFS或LocalFS上文本文件，指定文件路径和RDD分区数目。

源码：

def textFile(
	path : scala.Predef.String, 
	minPartitions : scala.Int = {  }
) : org.apache.spark.rdd.RDD[scala.Predef.String] = {  }

注意：

• 传入的文件路径，可以是 hdfs://… ， 也可以是 file://… 。这种方式分为在集群中执行和在本地执行，在集群中是hdfs://，本地则是file:// 。
• 支持分区，如果传入的path是 hdfs://… ，分区则由hdfs文件的block决定。默认情况下，RDD分区数目等于HDFS上Block块数目。其中文件路径：最好是全路径，可以指定文件名称，可以指定文件目录，可以使用通配符指定。
• 支持外部数据源，比如阿里源等等。

---

3. 从RDD衍生新的RDD（原地计算）

通过在RDD上进行算子 *** 作，会生成新的RDD，那么新的RDD是原来的RDD吗？提一个概念，原地计算？类比一下字符串 *** 作 str.substr 返回的是新的 str。那么这个叫做非原地计算，那么原来的字符串变了吗？当然没变？同理RDD不可变！！！

---

4. 小文件读取（wholeTextFiles）

在实际项目中，有时往往处理的数据文件属于小文件（每个文件数据数据量很小，比如KB，几十MB等），文件数量又很大，如果一个个文件读取为RDD的一个个分区，计算数据时很耗时性能低下，使用SparkContext中提供：wholeTextFiles类，专门读取小文件数据。

def wholeTextFiles(
	path: String,	//文件存储目录
	minPartitions: Int = defaultMinPartitions	//RDD分区数目
): RDD[(String, String)]

范例演示：读取100个小文件数据，每个文件大小小于1MB，设置RDD分区数目为2。

实际项目中，可以先使用wholeTextFiles方法读取数据，设置适当RDD分区，再将数据保存到文件系统，以便后续应用读取处理，大大提升性能。

---

RDD 分区数目（总结）

在讲解 RDD 属性时，多次提到了分区（partition）的概念。分区是一个偏物理层的概念，也是 RDD 并行计算的核心。数据在 RDD 内部被切分为多个子集合，每个子集合可以被认为是一个分区，运算逻辑最小会被应用在每一个分区上，每个分区是由一个单独的任务（task）来运行的，所以分区数越多，整个应用的并行度也会越高。

获取 RDD 分区数目的俩种方式

• rdd.getNumPartitions
• rdd.partitions.length

RDD 分区的数据取决于那些因素？

• 第一点：RDD分区的原则是使得分区的个数尽量等于集群中的CPU核心(core)数目，这样可以充分利用CPU的计算资源；
• 第二点：在实际中为了更加充分的压榨CPU的计算资源，会把并行度设置为cpu核数的2~3倍；
• 第三点：RDD分区数和启动时指定的核数、调用方法时指定的分区数、如文件本身分区数有关系，具体如下说明：
  • 1.启动的时候指定的CPU核数确定了一个参数值：
    • spark.default.parallelism = 指定的CPU核数（集群模式最小为2）
  • 2.对于Scala集合调用parallellize（集合，分区数）方法：
    • 如果没有指定分区数目，就使用 spark.default.parallelism
    • 如果指定了就使用指定的分区数（不要指定大于spark.default.parallelism）
  • 3.对于textFile（文件，分区数）
    • defaultMinPartitons
      • 如果没有指定分区数目 sc.defaultMinPartitions=min(defaultParallelism, 2)
      • 如果指定了就使用指定的分区数 sc.defaultMinPartitions=指定的分区数rdd的分区数
    • rdd的分区数
      • 对于本地文件：
        • rdd的分区数 = max(本地file的分片数， sc.defaultMinPartitions)
      • 对于HDFS文件：
        • rdd的分区数 = max(hdfs文件的block数目，sc.defaultMinPartitions)
      • 所以如果分配的核数为多个，且从文件中读取数据创建RDD，即使hdfs文件只有1个切片，最后的Spark的RDD的partition数也有可能是2