IO装饰者设计模式

BufferedInputStream in = new BufferedInputStream(new FileInputStream());
in.close();
其实是将FileInputStream()关了,而FileInputStream是BufferedInputStream的内部属性,内部形成了互相引用的关系.

new RDD ==> new RDD (移动数据不如移动计算)
转换算子其实就是包装.
Spark RDD通过其Transactions操作,形成了RDD血缘关系图,即DAG,最后通过Action的调用,触发Job并调度执行.DAGScheduler负责Stage级的调度,主要是将job切分成若干Stages,并将每个Stage打包成TaskSet交给TaskScheduler调度.TaskScheduler负责Task级的调度,将DAGScheduler给过来的TaskSet按照指定的调度策略分发到Executer上执行,调度过程中SchedulerBackend负责提供可用资源,其中SchedulerBackend有多种实现,分别对接不同的资源管理系统.
eg:

  /**
   *  Return a new RDD by first applying a function to all elements of this
   *  RDD, and then flattening the results.
   */
  def flatMap[U: ClassTag](f: T => TraversableOnce[U]): RDD[U] = withScope {
    val cleanF = sc.clean(f)
    new MapPartitionsRDD[U, T](this, (context, pid, iter) => iter.flatMap(cleanF))
  }

Action算子

  /**
   * Return an array that contains all of the elements in this RDD.
   *
   * @note This method should only be used if the resulting array is expected to be small, as
   * all the data is loaded into the driver's memory.
   */
  def collect(): Array[T] = withScope {
    val results = sc.runJob(this, (iter: Iterator[T]) => iter.toArray)
    Array.concat(results: _*)
  }

Spark快,不断的包装RDD,让其功能越来越强大.Spark早期版本不落盘,是存放在内存中的,但是大数据的出现,数据太大,经常会出现GC,OOM,为了解决这个问题,Spark每一次落盘都要Shuffle,数据量小落盘也不影响.

MapReduce面向的是一次性操作,Map完肯定是Reduce,中间要落盘,效率很低.1.x存在很严重的问题,不适合迭代计算,图形计算也不行.

SparkJob执行


Driver初始化SparkContext过程中,会分别初始化DAGScheduler/TaskSCheduler/SchedulerBackend以及HeartbeatReceiver,并启动SchedulerBackend以及HeartbeatReceiver.SchedulerBackend通过ApplicationMaster申请资源,并不断从TaskScheduler中拿到合适的Task分发到Executor执行.HeartbeatReceiver负责接收Executor的心跳信息,监控Executor的存活状况,并通知到TaskScheduler.

Spark执行流程