SparkCore(一)

忘是亡心i 2022-02-03 05:25 326阅读 0赞

# **1.什么是RDD** #

RDD分布式数据集，是Spark中最基础的数据抽象，代码中谁一个抽象类，它代表一不可变、可分区、元素可并行计算的集合。

## 1.2RDD的特点 ##

分区、只读、依赖、缓存、CheckPoint

分区：逻辑上分区，每个分区的数据都是抽象的存在，计算的时候云通过一个compute函数得到每个分区的数据。如果RDD是通过已有的文件系统构建，则compute函数是读取指定文件系统中的数据，如果RDD是通过其他RDD转换而来，则compute函数是执行转换逻辑将其他RDD的数据进行转换。  
只读：RDD是只读的，要想改变RDD中的数据，只能在现有的RDD基础上创建新的RDD。  
依赖：RDD通过算子进行转换， 转换得到新的RDD包含了从其他RDD衍生所必需的信息，RDD之间维护着这种血缘关系，也称之为依赖。  
缓存：在应用程序中多次使用同一个RDD,但是只需要第一次使用的时候会根据血缘关系得到分区的数据。在其使用的时候会直接缓存，从缓存处读取而不是再根据血缘关系计算，这样就加速了后期的重用。  
CheackPoint：RDD的血缘关系天然地可以实现容错，当RDD的某个分区的数据失败或丢失，可以通过血缘关系重建。但是对于长时间的迭代来说，随着迭代的进行，RDD之间的血缘关系会越来越长，一旦在后续迭代过程重出错，则需要通过非常长的血缘关系去重建，势必会影响性能。 因此 RDD支持Checkpoint将数据保存持久化到存储中，这样就可以切断之前的血缘关系，可以直接从Checkpoint拿到数据。

# 2.RDD编程 #

在Spark中创建RDD有3种创建方式： 1.从集合中创建RDD。 2.从外部存储创建RDD。3从其他RDD创建。

## 2.1从集合中创建 ##

从集合中创建RDD,Spark主要提供了2中函数 ：parallelize 和 makeRDD(从底层源码可以看到makeRDD调用了 parallelize)

object Spar01_RDD {
      def main(args: Array[String]): Unit = {
        val wordCount: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SiyuanChen")
        val sc: SparkContext = new SparkContext(wordCount)
    
        /**
          * 创建RDD
          * 1从内存中创建 makeRDD 底层实现就是parallelize
          * 
          */
        val listRDD: RDD[Int] = sc.makeRDD(List(1,2,3,4))
        val listRDD1: RDD[Int] = sc.parallelize(List(1,2,3,4))
        
      }
    }

package com.chensy.spark
    
    import org.apache.spark.rdd.RDD
    import org.apache.spark.{SparkConf, SparkContext}
    
    object Spar01_RDD {
      def main(args: Array[String]): Unit = {
        val wordCount: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SiyuanChen")
        val sc: SparkContext = new SparkContext(wordCount)
    
        /**
          * 创建RDD
          * 1从内存中创建 makeRDD 底层实现就是parallelize
          *
          */
        //val listRDD: RDD[Int] = sc.makeRDD(List(1,2,3,4))
        //val listRDD1: RDD[Int] = sc.parallelize(List(1,2,3,4))
    
        //使用自定义分区
        //内存中创建
        //val listRDD: RDD[Int] = sc.makeRDD(List(1,2,3,4),2)
        //val value: RDD[String] = sc.makeRDD(Array("1","2","3"))
    
        /**
          * 从外部存储中创建
          * 默认情况下可以读取项目路径，也可以读取其他路径：HDFS
          * 默认从文件中读取的数据都是字符串类型
          * 读取文件时，传递分区参数为最小分区，但是不一定是这个分区数，取决Hadoop读取文件的规则
          *
          *
          */
        val fileRDD: RDD[String] = sc.textFile("in",2)
    
        fileRDD.saveAsObjectFile("output")
    
      }
    }

## 2.2 RDD的转换 ##

RDD整体上分为Value 类型和Key-Value类型  
**2.2.1 map(func)案例**  
1.作用：返回一个新的RDD,该RDD由每一个输入元素经过func函数转换过后组成  
2需求：创建一个1-10的数组的RDD,将所有的元素\*2形成新的RDD

import org.apache.spark.rdd.RDD
    import org.apache.spark.{SparkConf, SparkContext}
    
    object Spark02_Oper1 {
      def main(args: Array[String]): Unit = {
        val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SiyuanChen")
    
        // 创建Spark上下文对象
        val sc = new SparkContext(config)
    
        /**
          * map 算子
          *
          */
        val listRDD: RDD[Int] = sc.makeRDD(1 to 10)
        val mapRDD: RDD[Int] = listRDD.map(_*2)
        println(mapRDD.collect().mkString(",")) //2,4,6,8,10,12,14,16,18,20
      }
    }

**2.2.2 mapPartitions(func)案例**  
作用：类似于map,但独立地在RDD的每个分片上运行，因此在类型为T的RDD上运行时，**func的函数类型必须是Iterator\[T\]=>Iterator\[U\]**。假设有N个元素，有M个分区，那么map的函数的将被调用N次，而mapPartitions被调用M次，一个函数一次处理的所有分区。  
需求：创建一个RDD，使每个元素\*2组成新的RDD

object Spark03_Oper2 {
      def main(args: Array[String]): Unit = {
        val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SiyuanChen")
        val sc: SparkContext = new SparkContext(conf)
        val listRDD: RDD[Int] = sc.makeRDD(1 to 10)
    
        /**
          *mapPartitions可以对一个RDD中的所有分区进行遍历
          * mapPartitons效率高于map算子，减少了发送到执行器执行交互的次数
          * mapPartitons可能会出现 内存溢出（oom）
          *
          */
        println(listRDD.mapPartitions(x => {
          x.map(_ * 2)
        }).collect().mkString(","))
      }
    }

**2.2.3mapPartitionsWithIndex(func)案例**  
作用：类似于mapPartitions，但func带有一个整数参数表示分片的索引值，因此在类型t的RDD上运行，func的函数类型必须是(Int,Interator\[T\]=>Interator\[U\]);  
需求：创建一个RDD,使每个元素跟所在分区形成一个元祖组成一个新的RDD

object Spark04_Oper3 {
      def main(args: Array[String]): Unit = {
    
        val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SiyuanChen")
    
        // 创建Spark上下文对象
        val sc = new SparkContext(config)
        //mapPartitionsWithIndex
        val listRDD: RDD[Int] = sc.makeRDD(1 to 10,2)
        println(listRDD.mapPartitionsWithIndex {
          case (index, datas) => {
            datas.map((_, "分区号" + index))
          }
        }.collect().mkString(","))//(1,分区号0),(2,分区号0),(3,分区号0),(4,分区号0),(5,分区号0),(6,分区号1),(7,分区号1),(8,分区号1),(9,分区号1),(10,分区号1)
    
      }
    }

**2.3.4flatMap(func)案例**  
作用：类似于map，但是每一个输入元素可以被映射为0或者多个输出元素(所以func应该返回)  
需求：创建一个元素为1-5的RDD，运行flatMap创建一个新的RDD，新的RDD为1,1,2,1,2,3,1,2,3,4,1,2,3,4,5

import org.apache.spark.rdd.RDD
    import org.apache.spark.{SparkConf, SparkContext}
    
    object Spark05_Oper4 {
      def main(args: Array[String]): Unit = {
    
        val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SiyuanChen")
    
        // 创建Spark上下文对象
        val sc = new SparkContext(config)
       //flatMap
        //1,2,3,4
        val listRDD = sc.makeRDD(1 to 5)
        println(listRDD.flatMap(1 to _).collect().mkString(","))
      }//1,1,2,1,2,3,1,2,3,4,1,2,3,4,5
    }

2.2.6 glom案例  
作用：将每一个分区形成一个数组，形容新的RDD类型时RDD\[Array\[T\]\]  
需求：创建一个4个分区的RDD, 丙将每个分区的数据放到一个数组

import org.apache.spark.rdd.RDD
    import org.apache.spark.{SparkConf, SparkContext}
    object Spark06_Oper5 {
      def main(args: Array[String]): Unit = {
        val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SiyuanChen")
    
        // 创建Spark上下文对象
        val sc = new SparkContext(config)
    
        val listRDD: RDD[Int] = sc.makeRDD(List(1,2,8,4,3,5,6,7),3)
        //将一个分区的数据放在一个数组中
        val glomRDD: RDD[Array[Int]] = listRDD.glom()
        //glomRDD.collect().foreach(array=>{println(array.mkString(","))})
        glomRDD.collect().foreach(array=>{
          println(array.mkString(","))})
    
      }
    }

**2.2.7groupBy(func)案例**  
作用：分组，按照传入函数的返回值进行分组。将相同的key对应的值放入一个迭代器。  
需求：创建一个RDD，按照元素%2的值进行分组

import org.apache.spark.rdd.RDD
    import org.apache.spark.{SparkConf, SparkContext}
    object Spark07_Oper6 {
      def main(args: Array[String]): Unit = {
        val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SiyuanChen")
    
        // 创建Spark上下文对象
        val sc = new SparkContext(config)
        //生成数据，按照指定的规则进行分组
        //分组后的数据形成了对偶元组（k-v），k表示分组的key v表示分组的数据集
        val listRDD: RDD[Int] = sc.makeRDD(List(1,2,3,4))
        println(listRDD.groupBy(i => i % 2).collect().mkString(","))
      }
    }
    //(0,CompactBuffer(2, 4)),(1,CompactBuffer(1, 3))

**2.2.8filter(func)案例**  
作用：过滤。返回一个新的RDD。该RDD由经过func函数计算后返回true的输入元素组成  
需求：创建一个RDD(12345)，过滤出一个新的RDD(i%2==0)

import org.apache.spark.rdd.RDD
    import org.apache.spark.{SparkConf, SparkContext}
    
    object Spark08_Oper7 {
    
      def main(args: Array[String]): Unit = {
    
        val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SiyuanChen")
    
        // 创建Spark上下文对象
        val sc = new SparkContext(config)
        val listRDD: RDD[Int] = sc.makeRDD(List(1,2,3,4))
        println(listRDD.filter(x => x % 2 == 0).collect().mkString(","))
    //2,4
      }
    }

**2.2.9sample(withReplacement, fraction, seed) 案例**  
作用：以指定的随机种子随机抽象出书记fraction的数据，withReplacement 表示是抽搐的数据是否放回，true为有放回的抽象，false为无放回的抽样，seed用于指定随机数生成种子  
需求：创建 一个1-10 RDD 从中选择放回和不放回抽样

import org.apache.spark.rdd.RDD
    import org.apache.spark.{SparkConf, SparkContext}
    object Spark09_Oper8 {
      def main(args: Array[String]): Unit = {
        val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    
        // 创建Spark上下文对象
        val sc = new SparkContext(config)
        val listRDD: RDD[Int] = sc.makeRDD(1 to 10)
        val sampleRDD:  RDD[Int] = listRDD.sample(false,0.4,2)
        println(sampleRDD.collect().mkString(","))
    
    
      }
    
    }

**2.2.10coalesce(numPartitions) 案例**  
作用：缩减分区数，用于大数据集过滤后，提高小数据集的执行效率  
需求：创建一个4个分区的RDD,对其缩减分区

import org.apache.spark.rdd.RDD
    import org.apache.spark.{SparkConf, SparkContext}
    object Spark10_Oper9 {
      def main(args: Array[String]): Unit = {
        val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SiyuanChen")
    
        // 创建Spark上下文对象
        val sc = new SparkContext(config)
        val listRDD: RDD[Int] = sc.makeRDD(1 to 16,4)
        println("缩减分区前 = " + listRDD.partitions.size)
        val coalesceRDD: RDD[Int] = listRDD.coalesce(3)
        println("缩减分区后 = " + coalesceRDD.partitions.size)
        coalesceRDD.saveAsTextFile("outpou")
      }
    
    }

**2.2.11distinct(\[numTasks\])) 案例**  
作用：对源RDD进行去重后返回一个新的RDD。默认情况下，只有8个并行任务来操作，但是可以传入一个可选的numtask参数去改变它  
需求：创建一个RDD 使用distinct对其去重

import org.apache.spark.rdd.RDD
    import org.apache.spark.{SparkConf, SparkContext}
    object Spark11_Oper10 {
      def main(args: Array[String]): Unit = {
        val config: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SiyuanChen")
    
        // 创建Spark上下文对象
        val sc = new SparkContext(config)
        val listRDD: RDD[Int] = sc.makeRDD(List(1,2,1,5,2,9,6,1))
        val distinctRDD: RDD[Int] = listRDD.distinct()
        distinctRDD.saveAsTextFile("output")
      }
    }