Spark 获取RDD案例详解

1）获取内存中的RDD

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
//从内存中获取RDD
object _01FromMemory {
    def main(args: Array[String]): Unit = {
        //配置文件
        val conf: SparkConf = new SparkConf().setMaster("local").setAppName("createRDD")
        //获取上下文对象
        val sc = new SparkContext(conf)
​
        //从内存中创建RDD
        val list = List(3, 6, 78, 9)
​
        
        val rdd: RDD[Int] = sc.parallelize(list)
        rdd.foreach(println)
​
        println("--------------------------------")
        
        val rdd2 = sc.makeRDD(list)
        rdd2.foreach(println)
​
        //3
        //6
        //78
        //9
        //--------------------------------
        //3
        //6
        //78
        //9
    }
}
​

2）获取外部数据的RDD

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
​
//从外部数据获取RDD
object _02FromOthers {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local").setAppName("getRDD")
        val sc = new SparkContext(conf)
​
        
        val rdd = sc.textFile("D://input//words.txt")
​
        
        val rdd2: RDD[String] = sc.textFile("data/1.txt")
        //打印结果
        rdd.collect().foreach(println)
        println("-------------------------")
        rdd2.collect().foreach(println)
        
        //hello world
        //hello spark
        //hello scala
        //hello java
        //hello spark
        //-------------------------
        //12
        //12345
        //3
    }
}

3）读取内存数据的并行度和分区

import org.apache.spark.{SparkConf, SparkContext}
​
//读取内存中的分区和并行度
//并行度指的就是分区的数量，也同时指的就是Task的数量
object _03PartitionFromMemory {
    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("readRDD")
        
        conf.set("spark.default.parallelism","4")  //如果取消掉就是走local[*]
        val sc = new SparkContext(conf)
​
        val rdd = sc.makeRDD(List(3, 5, 6, 6, 7))
​
        
        rdd.saveAsTextFile("out")
        sc.stop()
        //我使用的是4core 所以会有四个分区 每个分区平均分配到一个数据
    }
}
​

4）读取外部数据的并行度和分区

package com.qf.spark.wc.day03.getrdd
​
import org.apache.spark.{SparkConf, SparkContext}
​
object _04PartitionFromOuterFile {
    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("readRDD")
        val sc = new SparkContext(conf)
​
        //textFile的第二个参数解析：表示设置最小分区数，因此实际分区数可能大于该值
        val rdd = sc.textFile("data/1.txt", 2)
        rdd.saveAsTextFile("out")
        sc.stop()
    }
    //这里会生成了两个分区，因为第二个参数进行了设置
}
//生成分区1的数据
//12
//12345
​
//分区2
//3

5）内存分区的数据

import org.apache.spark.{SparkConf, SparkContext}
​
object _05PartitionDataFromMemory {
    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("MemoryData")
        val sc = new SparkContext(conf)
​
        val rdd = sc.makeRDD(List(1, 3, 8, 9),2)
        //获取分区数的方法
        rdd.saveAsTextFile("out")//生成分区
        sc.stop()
​
        //result
        //1 3 一个分分区
        //8 9 一个分区
        //总结：均分，如不能均分就从后面的分区开始加入数据
    }
}
​

6）外部文件的分区的数据

import org.apache.spark.{SparkConf, SparkContext}
//获取外部文件分区的数据
//spark的分区数量就是和MR的分片数量有关，而MR的分片计算有一个阈值1.1
object _06PartitionDataFromFile {
    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("MemoryData")
        val sc = new SparkContext(conf)
​
        val rdd = sc.textFile("data/1.txt", 2)
        rdd.saveAsTextFile("out")
        sc.stop()
​
        //解释
        //       比如数据如下：
        //              12@@
        //              12345@@
        //              1
        //
        //              总共12个字节，期望每个分区6个字节，但是通常来说，上一个分区和下一个分区的数据的临界行，会跨这两个分区
        //              也就是上一个分区的最后一行的数据是溢出的，因此要读取到换行符号才行
        //              第一个分区的数据： 12@@
        //                                12345@@  在同一个分区
        //              第二个分区的数据，不会重读读过的数据，因此只有1
        
        //result 生成如下两个分区
        //12
        //12345
        //-----------------
        //1
    }
}
​

6）RDD的读取和保存

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
​
object _07RDDIO {
    def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("MemoryData")
        val sc = new SparkContext(conf)
​
        
        //        val value = sc.textFile("data/1.txt")
        //        val value: RDD[(String, String)] = sc.wholeTextFiles("data")
        //        val value: RDD[Nothing] = sc.objectFile("data/part-00001")
        val value: RDD[(String, Int)] = sc.sequenceFile("data/part-00001")
​
        
        value.saveAsTextFile("out")
        //        value.saveAsObjectFile("outObject")
        //        value.map((_,1)).saveAsSequenceFile("outSequence")
        value.foreach(println)
    }
}
​