Spark Streaming 不同Batch任务可以并行计算么?-阿里云开发者社区

开发者社区> 大数据> 正文
登录阅读全文

Spark Streaming 不同Batch任务可以并行计算么?

简介: 其实Job,Stage,Task都是Spark Core里就有的概念,Batch则是Streaming特有的概念。同一Stage里的Task一般都是并行的。同一Job里的Stage可以并行,但是一般如果有依赖则是串行,可以参考我这篇文章Spark 多个Stage执行是串行执行的么?。
关于Spark Streaming中的任务有如下几个概念:
  • Batch
  • Job
  • Stage
  • Task
其实Job,Stage,Task都是Spark Core里就有的概念,Batch则是Streaming特有的概念。同一Stage里的Task一般都是并行的。同一Job里的Stage可以并行,但是一般如果有依赖则是串行,可以参考我这篇文章Spark 多个Stage执行是串行执行的么?

Job的并行度复杂些,由两个配置决定:
  1. spark.scheduler.mode(FIFO/FAIR)
  2. spark.streaming.concurrentJobs
我们知道一个Batch可能会有多个Action执行,比如你注册了多个Kafka数据流,每个Action都会产生一个Job,所以一个Batch有可能是一批Job,也就是JobSet的概念,这些Job由jobExecutor依次提交执行,而JobExecutor是一个默认池子大小为1的线程池,所以只能执行完一个Job再执行另外一个Job。这里说的池子,他的大小就是由spark.streaming.concurrentJobs 控制的。

concurrentJobs 其实决定了向Spark Core提交Job的并行度。提交一个Job,必须等这个执行完了,才会提交第二个。假设我们把它设置为2,则会并发的把Job提交给Spark Core,Spark 有自己的机制决定如何运行这两个Job,这个机制其实就是FIFO或者FAIR(决定了资源的分配规则)。默认是FIFO,也就是先进先出,你把concurrentJobs设置为2,但是如果底层是FIFO,那么会优先执行先提交的Job,虽然如此,如果资源够两个job运行,还是会并行运行两个Job。

我们搞个例子来论证下上面的结论:

object JobTest {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName("test")
    conf.setMaster("local[2]")
    conf.set("spark.streaming.concurrentJobs", "2")
    val sc = new StreamingContext(conf, Seconds(10))

    val input = new TestInputStream[String](sc, Seq(Seq("1", "2", "3"), Seq("1", "2", "3"), Seq("1", "2", "3")), 2)
    val input2 = new TestInputStream[String](sc, Seq(Seq("1", "2", "3"), Seq("1", "2", "3"), Seq("1", "2", "3")), 2)

    input.map{f=>
      Thread.sleep(5000)
      f
    }.foreachRDD(f=>f.count())

    input2.map{f=>
      Thread.sleep(5000)
      f
    }.foreachRDD(f=>f.count())

    sc.start()
    sc.awaitTermination()

  }
}
源码github地址

上面的TestInputStream的签名如下:
class TestInputStream[T: ClassTag](_ssc: StreamingContext, input: Seq[Seq[T]], numPartitions: Int)
  extends InputDStream[T](_ssc) {
所以TestInputStream其实就是我Mock的一个数据源,最后numPartitions表示的是分区数。这里,我们把concurrentJobs设置为2,意味着TaskScheduler接受到了两个Job,然后setMaster[local(2)]表示只可以并发执行两个Task。

因为input,input1每个batch至少都有3个元素,每个元素需要运行5秒,所以有一个task需要运行两个元素,那么第一次input1需要运行10秒。input1在运行五秒后,空出了一个线程,这个时候input的job开始运行,到第十秒的时候,input1完成,input开始运行也已经完成一个元素的计算,这个时候启动另外两个元素运行。所以input1花了10秒,input花了15秒,但是因为input被延时了五秒才得以运行,所以input1其实相当于花了20秒。

这里你会好奇,为啥我先声明的input,接着再申明的input1,但是input1却先运行呢?因为这两个数据源对应的job是被并发提交的,有一定的随机性。如果你多启动几次,你会发现input对应job id有可能是0,也有可能是1。

还有两点值的注意的是:
  1. job id的产生是在job提交的时候才产生,而不是job在产生的时候生成的。
  2. job被提交后会直接进入Scheduler的pool,在scheduler给你分配资源的时候,虽然说FIFO是先按job id 小的优先处理,但是job id大的先进来,在分配资源的时候,小的还没进来呢,所以job id 大的可能被优先执行了。

上面的流程解说解释的是下面这张图:
e4e4e81e2c7639286fecb4bf99d4aa137a6f7912

接着呢,input2在剩下两条记录处理的10秒过程中,其实第二个周期已经开始了,input的任务又得以开始运行,这个时候因为只有一个线程可以用,所以运行了两个元素,input1处理完成,空出线程,第二个周期的input1继续调度,input的剩下的一个元素也继续运行,最后input,input1都花了15秒。

f6b5d18db968677665a4e6de429a184c157936a9

有点绕,如果大家迷惑,可以把代码贴在自己的IDE上运行一下,然后观察他们的交错时间。
如果我们再做个调整:

conf.setMaster("local[4]")
    conf.set("spark.streaming.concurrentJobs", "3")
    conf.set("spark.scheduler.mode", "FIFO")
    val sc = new StreamingContext(conf, Seconds(5))
你会发现,不同batch的job其实也可以并行运行的,这里需要有几个条件:
  1. 有延时发生了,batch无法在本batch完成
  2. concurrentJobs > 1
  3. 如果scheduler mode 是FIFO则需要某个Job无法一直消耗掉所有资源

Mode是FAIR则尽力保证你的Job是并行运行的,毫无疑问是可以并行的。

回到我们的标题,不同Batch的job有可能会同时在运行么,只要满足我前面提到的三个条件,就有可能。

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:
大数据
使用钉钉扫一扫加入圈子
+ 订阅

大数据计算实践乐园,近距离学习前沿技术

其他文章
最新文章
相关文章