Spark04-RDD入门

RDD概念

Spark 的核心是建立在统一的抽象弹性分布式数据集（Resiliennt Distributed Datasets，RDD）之上的，这使得 Spark 的各个组件可以无缝地进行集成，能够在同一个应用程序中完成大数据的处理。

RDD是什么

RDD，全称弹性分布式数据集 (Resilient Distributed DataSet)。它是一种有容错机制的特殊数据集合，可以分布在集群的结点上，以函数式操作集合的方式进行各种并行操作。

通俗点来讲，可以将 RDD 理解为一个分布式对象集合，本质上是一个只读的分区记录集合。每个 RDD 可以分成多个分区，每个分区就是一个数据集片段。一个 RDD 的不同分区可以保存到集群中的不同结点上，从而可以在集群中的不同结点上进行并行计算。

RDD特点

RDD 具有容错机制，并且只读不能修改，可以执行确定的转换操作创建新的 RDD。具体来讲，RDD 具有以下几个属性。

只读：不能修改，只能通过转换操作生成新的 RDD。
分布式：可以分布在多台机器上进行并行处理。
弹性：计算过程中内存不够时它会和磁盘进行数据交换。
基于内存：可以全部或部分缓存在内存中，在多次计算间重用。
容错的：

RDD 实质上是一种更为通用的迭代并行计算框架，用户可以显示控制计算的中间结果，然后将其自由运用于之后的计算。

RDD 五大属性

Partition List：分片列表,记录RDD的分片,可以在创建 RDD的时候指定分区数目,也可以通过算子来生成新的RDD从而改变分区数目

Compute Function：为了实现容错,需要记录 RDD之间转换所执行的计算函数

RDD Dependencies ：RDD之间的依赖关系,要在RDD中记录其上级RDD是谁,从而实现容错和计算

Partitioner：为了执行 Shuffled操作,必须要有一个函数用来计算数据应该发往哪个分区

Preferred Location：优先位置,为了实现数据本地性操作,从而移动计算而不是移动存储,需要记录每个 RDD分区最好应该放在什么位置

RDD代码入门

主要是一个RDD大入口，sparkContext，和三种创建RDD的方法以及三个RDD算子初步使用，map , flatMap , reduceByKey。

package rdd

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

class Rdd {
  // SparkContext 创建 和 RDD 创建
  @Test
  def sparkContext(): Unit ={
    /*
    * 步骤
    * 1、创建 SparkConf，配置基本信息
    * 2、创建 SparkContext，利用上面的conf配置作为参数创建。
    * */
    val conf = new SparkConf().setMaster("local[6]").setAppName("spark_context")
    val sc = new SparkContext(conf)

    // sparkContext 作为 RDD 的大入口，具备许多功能，创建RDD，配置参数等。


    // 使用完后，需要对SparkContext进行关闭
  }
  // 可以将通用部分拿到方法外，类一加载就会执行，Scala的特点。
  val conf = new SparkConf().setMaster("local[6]").setAppName("spark_context")
  val sc = new SparkContext(conf)

  /*
  * 三种创建RDD的方式
  *   - 从本地集合创建
  *   - 从文件创建
  *   - 从 一个 RDD 衍生一个新的 RDD
  * */
  def rddCreate(): Unit ={
    /*
     * 从本地集合创建
     */
    val seq = Seq(1, 2, 3)
    sc.parallelize(seq,2) // 第二个参数可选，分区数
    sc.makeRDD(seq, 2)

    /*
     * 从文件创建
     */
    sc.textFile("")  // 路径传入由两种方式 hdfs://bigdata1//...   或   file://...
    // 支持aws(亚马逊) 和 阿里云 ：需要指定的 API
    }

  /*
   * 从 一个 RDD 衍生一个新的 RDD
   * 通过rdd的三个算子：map,flatMap,reduceByKey
   */
  @Test
  def rddSuanzi(): Unit ={
    /*
    * map:将集合里的每个元素进行操作，得到一个新集合
    * */
    // 1、创建RDD
    val map_rdd1 = sc.parallelize(Seq(1, 2, 3))
    // 2、执行 map 操作
    val map_rdd2 = map_rdd1.map(item => item * 10) // 接收到的参数，方法的返回值
    // 3、获取结果 collect
    val result_map = map_rdd2.collect()
    // 打印得到结果 10 20 30
    result_map.foreach(item => print(item + "	"))
    println()

    /*
    * flatMap:将集合每个数据进行操作，返回的是一个个数组，falt将其展开，得到一个新的大集合
    * */
    val flat_rdd1 = sc.parallelize(Seq("hello dongao", "hello zhangsan", "hello wangwu"))
    // split() 是得到一个数组，但flat自动将其展开
    val flat_rdd2 = flat_rdd1.flatMap(item => item.split(" "))
    val result_flat = flat_rdd2.collect()
    result_flat.foreach(item => print(item + "	"))
    println()

    /*
    * reduceByKey:先按照key分组，然后将每个组的value聚合起来
    * */
    val reduce_rdd1 = sc.parallelize(Seq("hello dongao", "hello zhangsan", "hello wangwu"))
    val reduce_rdd2 = reduce_rdd1.flatMap(item => item.split(" ")) // 将每个单词展开
      .map(item => (item, 1)) // 将展开的每个单词，作为一个词频为 1的 元组。
      .reduceByKey( (curr, agg) => curr + agg )
    val result_reduce = reduce_rdd2.collect()
    result_reduce.foreach(item => print(item + "	"))

    // 停止sparkContext
    sc.stop()
  }
}

其中，reduceByKey 的示例如下图：

reduceByKey 算子作用：

首先按照 Key 分组, 接下来把整组的 Value 计算出一个聚合值, 这个操作非常类似于 MapReduce 中的 Reduce

参数

func → 执行数据处理的函数, 传入两个参数, 一个是当前值, 一个是局部汇总, 这个函数需要有一个输出, 输出就是这个 Key 的汇总结果，并进行更新

注意点

ReduceByKey 只能作用于 Key-Value 型数据, Key-Value 型数据在当前语境中特指 Tuple2
ReduceByKey 是一个需要 Shuffled 的操作
和其它的 Shuffled 相比, ReduceByKey是高效的, 因为类似 MapReduce 的, 在 Map 端有一个 Cominer, 这样 I/O 的数据便会减少

map, flatMap 和 reduceByKey 算子总结

map 和 flatMap 算子都是转换, 只是 flatMap 在转换过后会再执行展开, 所以 map 是一对一, flatMap 是一对多
reduceByKey 类似 MapReduce 中的 Reduce