转载

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

关于Spark Streaming中的任务有如下几个概念：

Batch
Job
Stage
Task

其实Job,Stage,Task都是Spark Core里就有的概念，Batch则是Streaming特有的概念。同一Stage里的Task一般都是并行的。同一Job里的Stage可以并行，但是一般如果有依赖则是串行,可以参考我这篇文章 Spark 多个Stage执行是串行执行的么？。

Job的并行度复杂些，由两个配置决定：

spark.scheduler.mode(FIFO/FAIR)
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()

  }
}

上面的TestInputStream的签名如下：

class TestInputStream[T: ClassTag](_ssc: StreamingContext, input: Seq[Seq[T]], numPartitions: Int)
  extends InputDStream[T](_ssc) {

所以等同于一个数据源，其中最后䘝参数是分区数。这里，我们把concurrentJobs设置为2，意味着TaskScheduler接受到了两个Job,然后setMaster[local(2)]表示只可以并发执行两个Task。因为input1,input2每个batch至少都有3个元素，每个元素需要运行5秒，所以有一个task需要运行两个元素，那么第一次input2需要运行10秒。input2在运行五秒后，空出了一个线程，这个时候input1的job开始运行，到第十秒的时候，input2完成，input1开始运行也已经完成一个元素的计算，这个时候启动另外两个元素运行。所以input2花了10秒，input1花了15秒，但是因为input1被延时了五秒才得以运行，所以input2其实相当于花了20秒。这段话是解释下面那张图。

WX20170211-225643@2x.png

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

WX20170211-230145@2x.png

有点绕，如果大家迷惑，可以把代码贴在自己的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其实也可以并行运行的，这里需要有几个条件：