重要概念

  • 架构

    • Standalone模式下才有Master和Worker的叫法。
    • Master：类似YARN中的ResourceManager。作用：①监控Worker，接收对Worker的注册；②接收Client提交的Application，调度等待的Application并向Worker提交。
    • Worker：类似YARN中的NodeManager，掌管着所在节点的资源信息。作用：①通过RegisterWorker注册到Master;②发送心跳信息；③根据Master发送的Application配置进程环境，启动ExecutorBackend（执行Task所需的临时进程）。
    • Executor：SparkContext对象一旦连接到集群管理器，就可以获取到集群中每个节点上的Executor。Executor是一个进程（进程名：ExecutorBackend，运行在Work节点上），用来执行计算和为应用程序存储数据，然后Spark会发送应用程序代码（比如Jar包）到每个Ececutor，最后SparkContext对象发送任务到执行器开始执行程序。
    • Driver Program：运行应用程序的main()函数并创建SparkContext的线程。每个Spark应用程序都包括一个驱动程序，驱动程序负责把并行操作发布到集群中。Driver会将用户程序划分为不同的执行阶段，每个执行阶段由一组完全相同的Task组成，这些Task分别作用于待处理数据的不同分区。在阶段划分完成和Task创建后，Driver会向Executor发送Task。

  • 重要类：

    • SparkContext：负责连接到集群管理器。

  • RDD（弹性分布式数据集）

    • RDD类中的特点
      • 分区列表，即 一个RDD中有多个分区，每个分区中有多条数据，内部有方法getPartitions： Array[Partition]，默认等于核心数
      • 切片
      • 依赖列表，每个RDD都会依赖于其他RDD，内部有方法getDependencies: Seq[Dependency[_]]，窄依赖(NarrowDependency)分为OneToOneDependency和RangeDependency，宽依赖，没有对应的类，就只有ShuffleDependency。并且宽依赖一定会有shuffle阶段。
      • RDD是一个泛型类

  • 弹性：

    • 存储的弹性：内存和磁盘的自动切换
    • 容错的弹性：数据丢失可以自动恢复
    • 计算的弹性：计算出错重试机制，即出错时可能会切换Executor
    • 分片的弹性：可根据需要重新分片
      • 只读
    • RDD是只读的
    • RDD包括两类算子
      • transformation：将RDD进行转化，构建RDD的血缘关系
      • action：触发RDD的计算，得到结果或者保存文件系统，比如count、collect、savaAsTextFile
      • 依赖：RDDs通过操作算子开始进行转换，转换之后得到新的RDD包含从其他RDDs衍生所必须的信息，RDDs之间维护者这种血缘关系
      • 隐式转换：RDD的伴生类中有一些隐式转换，

  • 转换算子

    • 序列

      • 单值

        • map和mapPartitions的区别：map是对每个分区内的每个元素都执行，mapPartitions是对每个分区都执行一次
        • flatMap：一进多出，即flatMap中的方法返回的应该是一个可遍历的集合
        • glom：将分区中的所有元素合并成一个Array

      • 多值

        • zip：将两个RDD中索引相同的值合并为一个二元组

    • 键值对

      • 单值

        • partitionBy：重分区
        • reduceByKey：聚合算子，在shuffle之前聚合（即预聚合），并且分区内和分区间聚合逻辑相同
        • groupByKey：分组算子，
        • foldByKey：聚合算子，可以有0值，在分区内会先进行聚合，并且分区内和分区间聚合逻辑相同
        • aggregateByKey：聚合算子，三个参数，可以有0值，分区内调用func1，分区间调用func2
        • combineByKey：聚合算子，比aggregateByKey更加灵活
        • sortByKey：对key进行排序
        • mapValues：对value进行map操作

      • 多值

        • join和leftOuterJoin和rightOuterJoin：

  • 行动算子

    • collect
    • count
    • take
    • takeOrdered
    • countByKey
    • reduce：聚合，无0值，分区内和分区间相同
    • fold：聚合，有0值，分区内和分区间相同
    • aggregate：聚合，有0值，分区内和分区间不同
    • foreach：这个遍历是在executor上完成的，

  • 概念：

    • Application：一个Driver Program一个Application。基于Spark 构建的用户程序。 由集群上的驱动程序和执行程序组成。
    • Job：一个Application中，每个Job包含多个Stage，每个action会产生一个Job。由多个Task组成的并行计算，这些任务响应 Spark action而产生；你会在驱动程序的日志中看到这个术语。
    • Stage：每个宽依赖会产生一个新stage，也和分区器有关系（如果分区器相同，不产生新的Stage）。每个Job被划分为更小的task集，称为stage，彼此依赖(类似于MapReduce中的map和reduce阶段);您将在驱动程序日志中看到这个术语。
    • Task：一个Stage包含多个Task，Task是一个线程，是执行代码的最小单位，Task和分区数相等。将发送给一个Executor的工作单元去执行。

  • RDD缓存级别，聚合算子默认会进行缓存

  • 划分DAG：

    • 首先，根据依赖关系的不同将DAG划分为不同的阶段（Stage）。
      • 对于窄依赖，由于Partition依赖关系的确定性，Partition的转换处理就可以在同一个线程里完成，窄依赖被Spark划分到同一个执行阶段；
      • 对于宽依赖，由于Shuffle的存在，只能在parent RDD(s)Shuffle处理完成后，才能开始接下来的计算，因此宽依赖就是Spark划分Stage的依据，即Spark根据宽依赖将DAG划分为不同的Stage。
    • 在一个Stage内部，每个Partition都会被分配一个计算任务（Task），这些Task是可以并行执行的。
    • Stage之间根据依赖关系变成了一个大粒度的DAG，这个DAG的执行顺序也是从前向后的。也就是说，Stage只有在它没有parent Stage或者parent Stage都已经执行完成后，才可以执行。
    • 举例
      • 图片说明
      • 图片说明

  • 分区算法：

    • 代码

      private object ParallelCollectionRDD {
  /**
   * Slice a collection into numSlices sub-collections. One extra thing we do here is to treat Range
   * collections specially, encoding the slices as other Ranges to minimize memory cost. This makes
   * it efficient to run Spark over RDDs representing large sets of numbers. And if the collection
   * is an inclusive Range, we use inclusive range for the last slice.
   */
  def slice[T: ClassTag](seq: Seq[T], numSlices: Int): Seq[Seq[T]] = {
    if (numSlices < 1) {
      throw new IllegalArgumentException("Positive number of slices required")
    }
    def positions(length: Long, numSlices: Int): Iterator[(Int, Int)] = {
      (0 until numSlices).iterator.map { i =>
        val start = ((i * length) / numSlices).toInt
        val end = (((i + 1) * length) / numSlices).toInt
        (start, end)
      }
    }
    seq match {
      case r: Range =>
        positions(r.length, numSlices).zipWithIndex.map { case ((start, end), index) =>
          // If the range is inclusive, use inclusive range for the last slice
          if (r.isInclusive && index == numSlices - 1) {
            new Range.Inclusive(r.start + start * r.step, r.end, r.step)
          }
          else {
            new Range(r.start + start * r.step, r.start + end * r.step, r.step)
          }
        }.toSeq.asInstanceOf[Seq[Seq[T]]]
      case nr: NumericRange[_] =>
        // For ranges of Long, Double, BigInteger, etc
        val slices = new ArrayBuffer[Seq[T]](numSlices)
        var r = nr
        for ((start, end) <- positions(nr.length, numSlices)) {
          val sliceSize = end - start
          slices += r.take(sliceSize).asInstanceOf[Seq[T]]
          r = r.drop(sliceSize)
        }
        slices
        // 如果是一个Array，会将Array分割为多个Seq  
      case _ =>
        val array = seq.toArray 
        positions(array.length, numSlices).map { case (start, end) =>
            array.slice(start, end).toSeq
        }.toSeq
    }
  }
}