Spark核心编程 的 三大数据结构

2022-05-31 00:00:00 变量 节点 数据处理 计算 累加器

介绍

Spark计算框架为了能够进行高并发和高吞吐的数据处理，封装了三大数据结构，用于处理不同的应用场景。三大数据结构分别是：

  • RDD : 弹性分布式数据集
  • 累加器：分布式共享只写变量
  • 广播变量：分布式共享只读变量

一、RDD

1. 什么是RDD

RDD（Resilient Distributed Dataset）叫做弹性分布式数据集，是Spark中基本的数据处理模型。代码中是一个抽象类，它代表一个弹性的、不可变、可分区、里面的元素可并行计算的集合。

  • 弹性
    • 存储的弹性：内存与磁盘的自动切换；
    • 容错的弹性：数据丢失可以自动恢复；
    • 计算的弹性：计算出错重试机制；
    • 分片的弹性：可根据需要重新分片。
  • 分布式：数据存储在大数据集群不同节点上
  • 数据集：RDD封装了计算逻辑，并不保存数据
  • 数据抽象：RDD是一个抽象类，需要子类具体实现
  • 不可变：RDD封装了计算逻辑，是不可以改变的，想要改变，只能产生新的RDD，在新的RDD里面封装计算逻辑可分区、并行计算

2. 核心属性

 * Internally, each RDD is characterized by five main properties:
 *
 *  - A list of partitions
 *  - A function for computing each split
 *  - A list of dependencies on other RDDs
 *  - Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)
 *  - Optionally, a list of preferred locations to compute each split on (e.g. block locations for
 *    an HDFS file)
复制代码
  • 分区列表
    • RDD数据结构中存在分区列表，用于执行任务时并行计算，是实现分布式计算的重要属性。
  /**
   * Implemented by subclasses to return the set of partitions in this RDD. This method will only
   * be called once, so it is safe to implement a time-consuming computation in it.
   *
   * The partitions in this array must satisfy the following property:
   *   `rdd.partitions.zipWithIndex.forall { case (partition, index) => partition.index == index }`
   */
  protected def getPartitions: Array[Partition]
复制代码
  • 分区计算函数
    • Spark在计算时，是使用分区函数对每一个分区进行计算
  /**
   * Implemented by subclasses to return how this RDD depends on parent RDDs. This method will only
   * be called once, so it is safe to implement a time-consuming computation in it.
   */
  protected def getDependencies: Seq[Dependency[_]] = deps
复制代码

3.执行原理

从计算的角度来讲，数据处理过程中需要计算资源（内存 & CPU）和计算模型（逻辑）。执行时，需要将计算资源和计算模型进行协调和整合。 Spark框架在执行时，先申请资源，然后将应用程序的数据处理逻辑分解成一个一个的计算任务。然后将任务发到已经分配资源的计算节点上, 按照指定的计算模型进行数据计算。后得到计算结果。 RDD是Spark框架中用于数据处理的核心模型，接下来我们看看，在Yarn环境中，RDD的工作原理: 1)启动Yarn集群环境

2)Spark通过申请资源创建调度节点和计算节点

3)Spark框架根据需求将计算逻辑根据分区划分成不同的任务

4)调度节点将任务根据计算节点状态发送到对应的计算节点进行计算

从以上流程可以看出RDD在整个流程中主要用于将逻辑进行封装，并生成Task发送给Executor节点执行计算，接下来我们就一起看看Spark框架中RDD是具体是如何进行数据处理的。

4.基础编程

Spark核心编程的三大数据结构 之 RDD基础编程 (一)

Spark核心编程的三大数据结构 之 RDD基础编程 (二)

二、累加器

1.实现原理

累加器用来把Executor端变量信息聚合到Driver端。在Driver程序中定义的变量，在Executor端的每个Task都会得到这个变量的一份新的副本，每个task更新这些副本的值后，传回Driver端进行merge。

2.系统累加器 (3种)

    def main(args: Array[String]): Unit = {

        val sc = new SparkContext(
            new SparkConf().setMaster("local[*]").setAppName("MapPartitions")
        )

        val rdd = sc.makeRDD(List(
            1, 2, 5, 9
        ), 2)

        //数字累加
        val sum = sc.longAccumulator("sum")
        //集合累加
        //sc.collectionAccumulator("xx")
        //浮点累加
        //sc.doubleAccumulator("xxx")

        rdd.foreach(sum.add(_))

        println(s"sum = ${sum.value}")

        sc.stop()
    }
复制代码

3.自定义累加器

    def main(args: Array[String]): Unit = {

        val sc = new SparkContext(
            new SparkConf().setMaster("local[*]").setAppName("MapPartitions")
        )

        val rdd = sc.makeRDD(List(
            "java scala js golang", "golang python java", "spark hadoop"
        ), 2)

        val myAccumulator = new MyAccumulator
        //注册累加器
        sc.register(myAccumulator)

        rdd.flatMap(_.split(" ")).foreach(myAccumulator.add)

        println(s"sum = ${myAccumulator.value}")

        sc.stop()
    }

    class MyAccumulator extends AccumulatorV2[String, mutable.Map[String, Int]] {

        private var map: mutable.Map[String, Int] = mutable.Map[String, Int]()

        /**
         * 执行顺序
         * 1。copy       先复制
         * 2。reset      然后重置
         * 3。isZero     判断是否初始化状态
         */

        /**
         * 判断为初始化状态
         *
         * @return 返回 false 会抛出异常
         */
        override def isZero: Boolean = map.isEmpty

        /**
         * 复制累加器对象
         *
         * @return
         */
        override def copy(): AccumulatorV2[String, mutable.Map[String, Int]] = new MyAccumulator

        /**
         * 重制
         */
        override def reset(): Unit = map.clear()

        /**
         * 分区内累加
         *
         * @param v
         */
        override def add(v: String): Unit = map.update(v, map.getOrElse(v, 0) + 1)

        /**
         * 分区间累加
         *
         * @param other
         */
        override def merge(other: AccumulatorV2[String, mutable.Map[String, Int]]): Unit = {
            map = map.foldLeft(other.value)((map, kv) => {
                map.update(kv._1, map.getOrElse(kv._1, 0) + kv._2)
                map
            })
        }

        /**
         * 取结果
         *
         * @return
         */
        override def value: mutable.Map[String, Int] = map
    }
复制代码

三、广播变量

1.实现原理

广播变量用来高效分发较大的对象。向所有工作节点发送一个较大的只读值，以供一个或多个Spark操作使用。比如，如果你的应用需要向所有节点发送一个较大的只读查询表，广播变量用起来都很顺手。在多个并行操作中使用同一个变量，但是 Spark会为每个任务分别发送。

2.代码

    def main(args: Array[String]): Unit = {

        val sc = new SparkContext(
            new SparkConf().setMaster("local[*]").setAppName("MapPartitions")
        )

        val rdd = sc.makeRDD(List(
            1, 2, 5, 9
        ), 2)

        val dict = Map((1, "前端"), (2, "后端"), (3, "大数据"), (5, "UI"), (6, "客户端"), (9, "产品"))
        //广播
        sc.broadcast(dict)

        rdd.map(id => (id, dict.getOrElse(id, "未知")))
            .collect()
            .foreach(println)

        sc.stop()
    }

相关文章