spark streaming源码解读之job动态生成和深度思考

本节主要内如如下：

一、spark streaming job生成深度思考

二、spark streaming job生成源码解析

 输入的ds有很多来源Kafka、Socket、Flume，输出的DStream其实是逻辑级别的Action，是Spark Streaming框架提出的，其底层翻译成为物理级别的额Action，是RDD的Action，中间是处理过程是transformations，状态转换也就是业务处理逻辑的过程。

Spark Streaming二种数据来源：

1、基于DStream数据源。

2、基于其他DStream产生的数据源。

关键性的观点:做大数据的时候不是流失处理，一般会有定时任务，定时任务一般十分钟触发一次、一天触发一次，做大数据的定时任务就是流失处理的感觉，虽然不规范，一切和流处理没有关系的数据都是没有价值的。即使做批处理或数据挖掘其实也是在做数据流处理，只不过是隐形的流处理，所有的数据都会变成流处理。

所以就有统一的抽象，所有处理都是流处理的方式，所有的处理都将会被纳入流处理。企业大型开发项目都有j2ee后台支撑来提供各种人操作大数据中心。

Spark streaming程序入口就有batchDuration时间窗口，每隔五秒钟JobGeneration都会产生一个job，这个job是逻辑级别的，所以逻辑级别要有这个job，并且这个job该琢磨做，但环没有做，由底层物理级别的action去做，底层物理级别是基于rdd的依赖关系。Ds的action操作也是逻辑级别的。Ss根据axtion操作产生逻辑级别的job，但是不会运行，就相当线程runnable接口。逻辑级别的暂时没有身材物理级别的，所以可以去调度和优化，假设讲ds的操作翻译成rdd的action，最后一个操作是rdd的action操作，是不是已翻译就立即触发job，纪要完成翻译又不要生成job的话需要

JavaStreamingContext jsc = new JavaStreamingContext(conf, Durations.seconds(5));

下面主要从三个类进行解析：

1、JobGenerator类：根据batchDuration及内部默认的时间间隔生成Jobs；

2、JobScheduler：根据batchDuration负责Spark Streaming Job的调度；

3、ReceiverTracker：负责Driver端元数据的接收和启动executor中的接收数据线程；

1、JobGenerator类：

**
 * This class generates jobs from DStreams as well as drives checkpointing and cleaning
 * up DStream metadata.
 */
private[streaming]
class JobGenerator(jobScheduler: JobScheduler) extends Logging {

注释说基于dsg产生数据源，JobGenerator随着时间推移产生很多jobs，ss中除了定时身材的job，患有其他方式身材的job，例如基于各种聚合和状态的操作，状态操作不是基于batchd，他会对很多btchd处理。为了窗口之类的操作会触发JobGenerator，元素局的清理，作业生成的类。

// eventLoop is created when generator starts.
// This not being null means the scheduler has been started and not stopped
private var eventLoop: EventLoop[JobGeneratorEvent] = null //消息循环体定义
// last batch whose completion,checkpointing and metadata cleanup has been completed
private var lastProcessedBatch: Time = null
/** Start generation of jobs */
def start(): Unit = synchronized {
  if (eventLoop != null) return // generator has already been started
  // Call checkpointWriter here to initialize it before eventLoop uses it to avoid a deadlock.
  // See SPARK-10125
  checkpointWriter //执行checkpoint检查点
  eventLoop = new EventLoop[JobGeneratorEvent]("JobGenerator") {//内部匿名类创建
    override protected def onReceive(event: JobGeneratorEvent): Unit = processEvent(event) //事件处理逻辑
    override protected def onError(e: Throwable): Unit = {
      jobScheduler.reportError("Error in job generator", e)
    }
  }
  eventLoop.start()//启动事件处理线程对队列事件进行处理
  if (ssc.isCheckpointPresent) {
    restart()
  } else {
    startFirstTime()
  }
}

/**

 * An event loop to receive events from the caller and process all events in the event thread. It
 * will start an exclusive event thread to process all events.
 *
 * Note: The event queue will grow indefinitely. So subclasses should make sure `onReceive` can
 * handle events in time to avoid the potential OOM.
 */
private[spark] abstract class EventLoop[E](name: String) extends Logging {

  private val eventQueue: BlockingQueue[E] = new LinkedBlockingDeque[E]()//消息队列数据结构

  private val stopped = new AtomicBoolean(false)//原子变量

  private val eventThread = new Thread(name) {//封装线程对象
    setDaemon(true) //后台为线程

    override def run(): Unit = { //线程执行逻辑
      try {
        while (!stopped.get) {
          val event = eventQueue.take() //从消息队列中逐一获取消息对象
          try {
            onReceive(event) //对获取的消息对象进行业务处理
          } catch {
            case NonFatal(e) => { //处理失败后回调错误逻辑执代码
              try {
                onError(e)
              } catch {
                case NonFatal(e) => logError("Unexpected error in " + name, e)
              }
            }
          }
        }
      } catch {
        case ie: InterruptedException => // exit even if eventQueue is not empty
        case NonFatal(e) => logError("Unexpected error in " + name, e)
      }
    }
  }

  def start(): Unit = { //启动当前线程类
    if (stopped.get) {
      throw new IllegalStateException(name + " has already been stopped")
    }
    // Call onStart before starting the event thread to make sure it happens before onReceive
    onStart()
    eventThread.start() //启动当前线程类业务run方法的执行
  }

/** Processes all events */
private def processEvent(event: JobGeneratorEvent) {//根据消息对象执行相应的处理业务代码
  logDebug("Got event " + event)
  event match {
    case GenerateJobs(time) => generateJobs(time) //根据时间片生成Jobs
    case ClearMetadata(time) => clearMetadata(time) //时间片内的Jobs执行完毕，清理Driver上的元数据
    case DoCheckpoint(time, clearCheckpointDataLater) =>//时间片内的Jobs执行完毕，清理checkpint数据
　　      doCheckpoint(time, clearCheckpointDataLater)
　　case ClearCheckpointData(time) => clearCheckpointData(time)
  }
}

2、JobSchedule类：
/**

  * This class schedules jobs to be run on Spark. It uses the JobGenerator to generate

  * the jobs and runs them using a thread pool.
  */

private[streaming]

class JobScheduler(val ssc: StreamingContext) extends Logging {

  // Use of ConcurrentHashMap.keySet later causes an odd runtime problem due to Java 7/8 diff
  // https://gist.github.com/AlainODea/1375759b8720a3f9f094
  private val jobSets: java.util.Map[Time, JobSet] = new ConcurrentHashMap[Time, JobSet]//在指定的时间片内生成Jobs集合数据结构
  private val numConcurrentJobs = ssc.conf.getInt("spark.streaming.concurrentJobs", 1)
  private val jobExecutor =
    ThreadUtils.newDaemonFixedThreadPool(numConcurrentJobs, "streaming-job-executor")//启动指定大小的线程池
  private val jobGenerator = new JobGenerator(this)//启动JobGenerator对象
  val clock = jobGenerator.clock //jobGenerator时钟
  val listenerBus = new StreamingListenerBus() //linstenerBus消息总线

  // These two are created only when scheduler starts.
  // eventLoop not being null means the scheduler has been started and not stopped
  var receiverTracker: ReceiverTracker = null //driver端的元数据接收跟踪器
  // A tracker to track all the input stream information as well as processed record number
  var inputInfoTracker: InputInfoTracker = null //输入流信息跟踪器

  private var eventLoop: EventLoop[JobSchedulerEvent] = null //消息循环体对象

  def start(): Unit = synchronized { JobScheudler类启动主方法
    if (eventLoop != null) return // scheduler has already been started

    logDebug("Starting JobScheduler")
    eventLoop = new EventLoop[JobSchedulerEvent]("JobScheduler") {
      override protected def onReceive(event: JobSchedulerEvent): Unit = processEvent(event)

      override protected def onError(e: Throwable): Unit = reportError("Error in job scheduler", e)
    }
    eventLoop.start()
    // attach rate controllers of input streams to receive batch completion updates
    for {
      inputDStream <- ssc.graph.getInputStreams  //数据流
      rateController <- inputDStream.rateController //数据接收平率控制
    } ssc.addStreamingListener(rateController)

    listenerBus.start(ssc.sparkContext) //启动消息总线
    receiverTracker = new ReceiverTracker(ssc) //创建接收器对象
    inputInfoTracker = new InputInfoTracker(ssc) //创建数据输入对象
    receiverTracker.start() //启动数据接收器线程
    jobGenerator.start() //启动jobs产生器线程
    logInfo("Started JobScheduler")
  }

def submitJobSet(jobSet: JobSet) {

  if (jobSet.jobs.isEmpty) {
    logInfo("No jobs added for time " + jobSet.time)
  } else {
    listenerBus.post(StreamingListenerBatchSubmitted(jobSet.toBatchInfo))
    jobSets.put(jobSet.time, jobSet)
    jobSet.jobs.foreach(job => jobExecutor.execute(new JobHandler(job)))
    logInfo("Added jobs for time " + jobSet.time)
  }
}

private def handleJobStart(job: Job, startTime: Long) {
  val jobSet = jobSets.get(job.time)
  val isFirstJobOfJobSet = !jobSet.hasStarted
  jobSet.handleJobStart(job)
  if (isFirstJobOfJobSet) {
    // "StreamingListenerBatchStarted" should be posted after calling "handleJobStart" to get the
    // correct "jobSet.processingStartTime".
    listenerBus.post(StreamingListenerBatchStarted(jobSet.toBatchInfo))
  }
  job.setStartTime(startTime)
  listenerBus.post(StreamingListenerOutputOperationStarted(job.toOutputOperationInfo))
  logInfo("Starting job " + job.id + " from job set of time " + jobSet.time)
}

private class JobHandler(job: Job) extends Runnable with Logging {

    import JobScheduler._

    def run() {
      try {
        val formattedTime = UIUtils.formatBatchTime(
          job.time.milliseconds, ssc.graph.batchDuration.milliseconds, showYYYYMMSS = false)
        val batchUrl = s"/streaming/batch/?id=${job.time.milliseconds}"
        val batchLinkText = s"[output operation ${job.outputOpId}, batch time ${formattedTime}]"

        ssc.sc.setJobDescription(
          s"""Streaming job from <a href="$batchUrl">$batchLinkText</a>""")
        ssc.sc.setLocalProperty(BATCH_TIME_PROPERTY_KEY, job.time.milliseconds.toString)
        ssc.sc.setLocalProperty(OUTPUT_OP_ID_PROPERTY_KEY, job.outputOpId.toString)

        // We need to assign `eventLoop` to a temp variable. Otherwise, because
        // `JobScheduler.stop(false)` may set `eventLoop` to null when this method is running, then
        // it's possible that when `post` is called, `eventLoop` happens to null.
        var _eventLoop = eventLoop
        if (_eventLoop != null) {
          _eventLoop.post(JobStarted(job, clock.getTimeMillis()))
          // Disable checks for existing output directories in jobs launched by the streaming
          // scheduler, since we may need to write output to an existing directory during checkpoint
          // recovery; see SPARK-4835 for more details.
          PairRDDFunctions.disableOutputSpecValidation.withValue(true) {
            job.run()
          }
          _eventLoop = eventLoop
          if (_eventLoop != null) {
            _eventLoop.post(JobCompleted(job, clock.getTimeMillis()))
          }
        } else {
          // JobScheduler has been stopped.
        }
      } finally {
        ssc.sc.setLocalProperty(JobScheduler.BATCH_TIME_PROPERTY_KEY, null)
        ssc.sc.setLocalProperty(JobScheduler.OUTPUT_OP_ID_PROPERTY_KEY, null)
      }
    }
  }
}

新浪微博：http://weibo.com/ilovepains

微信公众号：DT_Spark

博客：http://blog.sina.com.cn/ilovepains

手机：18610086859

QQ：1740415547

邮箱：18610086859@vip.126.com

Spark发行版笔记6