Spark系列--SparkCore(二)RDD创建方式

桃扇骨 2022-05-20 05:04 202阅读 0赞

## 一、RDD的创建方式 ##

**创建RDD的创建方式大概可以分为三种：**

*  从集合中创建RDD；
 *  从外部存储创建RDD；
 *  从其他RDD转换。

### 1.1 从集合中创建RDD\*\* ###

从集合中创建RDD主要有下面两个方法：**makeRDD**和**parallelize**

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

默认的并行度defaultParallelism是通过以下代码来确定：

conf.getInt("spark.default.parallelism", math.max(totalCoreCount.get(), 2))

**makeRDD**本质上其实就是调用了**parallelize**

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

**Spark-Shell实例：**

scala> sc.makeRDD(1 to 100)
    res0: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at makeRDD at <console>:25
    
    scala> sc.makeRDD(Array("1","2","3"))
    res1: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[1] at makeRDD at <console>:25
    
    scala> res1.partitions.size
    res3: Int = 2

scala> sc.makeRDD(List(1,2,3),10)
    res0: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at makeRDD at <console>:25
    
    scala> res0.partitions.size
    res1: Int = 10

scala> sc.parallelize(List(1,3,5))
    res2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[1] at parallelize at <console>:25
    
    scala> res2.partitions.size
    res3: Int = 2

### 1.2 从外部存储创建RDD ###

**textFile(path) 从HDFS兼容的文件系统加载数据成为一个RDD**

Spark-Shell示例如下：

sc.textFile("hdfs://hadoop0:9000/RELEASE").flatMap(_.split(" ")) .map((_,1)).reduceByKey(_+_).saveAsTextFile("hdfs://hadoop0:9000/out")

正确的来讲，如下：

val rdd=sc.textFile("hdfs://hadoop0:9000/RELEASE")

**后面的flaMap、map、reduceByKey都是一系列转换操作，形成新的RDD，也就是第三种方式，saveAsTextFile是行动操作，这时候RDD才会真正执行。**