为啥Spark 的Broadcast要用单例模式

可以看到代码里首先会执行job生成代码

graph.generateJobs(time) 
 
具体代码块儿 
 
def generateJobs(time: Time): Seq[Job] = { 
    logDebug("Generating jobs for time " + time) 
    val jobs = this.synchronized { 
      outputStreams.flatMap { outputStream => 
        val jobOption = outputStream.generateJob(time) 
        jobOption.foreach(_.setCallSite(outputStream.creationSite)) 
        jobOption 
      } 
    } 
    logDebug("Generated " + jobs.length + " jobs for time " + time) 
    jobs 
  } 

每个输出流都会生成一个job，输出流就类似于foreachrdd，print这些。其实内部都是ForEachDStream。所以生成的是一个job集合。

然后就会将job集合提交到线程池里去执行，这些都是在driver端完成的哦。

jobScheduler.submitJobSet(JobSet(time, jobs, streamIdToInputInfos)) 
 
具体h函数内容 
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) 
    } 
  } 

其实就是遍历生成的job集合，然后提交到线程池jobExecutor内部执行。这个也是在driver端的哦。

jobExecutor就是一个固定线程数的线程池，默认是1个线程。

private val numConcurrentJobs = ssc.conf.getInt("spark.streaming.concurrentJobs", 1) 
  private val jobExecutor = 
    ThreadUtils.newDaemonFixedThreadPool(numConcurrentJobs, "streaming-job-executor") 

需要的话可以配置spark.streaming.concurrentJobs来同时提交执行多个job。

那么这种情况下，job就可以并行执行了吗?

显然不是的!

还要修改一下调度模式为Fair，详细的配置可以参考：

http://spark.apache.org/docs/2.3.3/job-scheduling.html#scheduling-within-an-application

简单的均分的话只需要

conf.set("spark.scheduler.mode", "FAIR") 

然后，同时运行的job就会均分所有executor提供的资源。

这就是整个job生成的整个过程了哦。

因为Spark Streaming的任务存在Fair模式下并发的情况，所以需要在使用单例模式生成broadcast的时候要注意声明同步。

（编辑：鞍山站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!