Spark定制班第2课：通过案例对Spark Streaming透彻理解三板斧之二：解密Spark Streaming运行机制和架构

r囧r小猫 2022-08-22 00:08 26阅读 0赞

本期内容：

1 解密Spark Streaming运行机制

2 解密Spark Streaming架构

**1 解密Spark Streaming运行机制**

上节课我们谈到了技术界的寻龙点穴。这就像过去的风水一样，每个领域都有自己的龙脉，Spark就是龙脉之所在，它的龙穴或者关键点就是SparkStreaming。这是上一节课我们非常清晰知道的结论之一。而且上一节课，我们采用了降维的方式。所谓降维的方式，是指把时间放大，就是把时间变长的情况下，我们做SparkStreaming的案例演示的实战，实战的结果是，我们发现在特定的时间段里面，确实是具体的RDD在工作，那么这一节课有必要在上一节课的基础上去谈一下它的运行机制和具体架构。

SparkStreaming在运行的时候，与其说是基于Spark Core上的一个流式处理的框架，不如说是它更像是SparkCore上的一个应用程序。而且在我们上一节课演示的时候，Spark Streaming在运行的时候启动了很多的Job，这个Job包含了两个层面，第一个就是围绕你每个batch，或者每个window运行的具体的Job，也有是为了框架正常运行而启动的Job，例如我们Receiver启动的时候，它启动了Job，而这个启动的Job是为其他Job服务的，我们由此得出了一个结论：要做复杂的Spark应用程序，往往多个Job之间会进行配合。所以，我们看SparkStreaming运行的时候，我们更多的是感受它是一个复杂的应用程序。上一节课我们说过这样一句话，SparkStreaming就是最复杂的应用程序。如果你对Spark Streaming了如指掌，你对任意的Spark程序的编写也一定没有问题。

我们先回顾下Spark官网（[http://spark.apache.org][http_spark.apache.org]）下面这张图，ApacheSpark就是Spark Core，最开始Spark发布的时候没有Apache Spark上面的子框架，他们是逐步开发出来的。这句话像废话，其实是有意义的，因为我们可以通过上层框架的使用，来洞察出Spark内部的机制到底是怎么样的。我们上节课也讲了从Spark Streaming入手定制Spark源码的原因，也讲了为什么不从其他框架入手的原因，所以我们要通过Spark Streaming彻底地研究Spark。

![SouthEast][]

我们以前基于SparkCore的时候都是基于RDD编程，包括以前的DataFrame，以及Spark 1.6.x推出的DataSet（据说Spark 2.x会成为主流），其实都是基于RDD编程。整个Spark都是基于RDD构建起来的，SparkCore的处理每一步都基于RDD进行操作的，RDD之间有依赖关系，这已经是大家的常识了。

我们再来看下面这张图。右边的RDDGraph显示RDD A依赖RDD M，RDD M进一步依赖RDD J，RDDJ又依赖RDD B。这个A从操作的角度讲就是Sparkaction。这里我们看到有3个Spark Job。如果从DStream的角度讲，这3个Job并没有执行。但从RDD的角度讲，因为是action，所以会具体的执行。为什么这么说呢？首先我们看得是RDD的DAG，Spark Streaming在RDD DAG的基础上加上了时间维度。

![SouthEast 1][]

回顾昨天课上运行的一个SparkStreaming程序，我们看到Spark Streaming除了自己框架的scheduler.JobScheduler、dstream.SocketInputDStream等内容，还使用了SparkCore的各种类型的RDD，比如rdd.MapPartitionRDD、rdd.PartitionerAwareUnionRDD、rdd.ShuffleRDD和rdd.BlockRDD等。这个Spark Streaming程序跟我们以前写的程序没有太多的区别，只不过是我们看见它在不断的运行，它之所以在不断地运行，就是因为在不断地循环，循环的依据就是基于时间维度。也就是说，DStream是RDD基础之上加上了时间维度，而RDD DAG依赖又叫空间维度，所以说整个Spark Streaming就是时空维度的一个东西。

![SouthEast 2][]

我们再来看一个下面的图。数据通过inputdata stream不断流进来；流进来就表示是有时间的，Spark Streaming根据时间把流进来的数据划分成不同Job（batches of inputdata），每个Job都有对应的RDD依赖，而每个RDD依赖都有输入的数据，所以这里可以看作是由不同RDD依赖构成的batch，而batch就是Job；然后由Spark Engine得出一个又一个的结果。

![SouthEast 3][]

我们继续看上图的最下面部分，操作的时候基于RDD的空间维度（空间维度就是具体做什么事情）又加上了时间维度。随着时间的流逝，在time1、time2、time3、time4上分别产生了4个RDD。具体产生了什么样内容的RDD，完全以时间为依据的。这个是SparkStreaming非常厉害的一点，只以时间为依据，和其他的一切逻辑和架构解耦合。我们可以认为SparkStreaming这种以时间为依据划分Job的方式是软件系统中最大解耦合的一种实现。这个确实是非常精妙的。

**2 解密Spark Streaming架构**

对于时空维度，我们可以想象一下有一个坐标，有X轴和Y轴，Y轴就是对RDD的操作，也就是所谓的RDD的依赖关系构成整个Job的逻辑，X轴就是时间。随着时间的流逝，假设是1秒钟，它会生成一个Job，即Job的实例，由于它在时空中，而时空中万物是运动的，所以1秒钟产生的Job在集群中运行就是运动。我们甚至可以深究其中蕴含的哲学级别的东西，这时候你看它就有一种心旷神怡的感觉。

我们用一个图来更清晰的说明刚才提到的时空维度坐标。我们用4个RDD Graph（有时也叫RDD DAG）表示随着时间的流逝，基于DStream Graph不断地生成RDDGraph，也就是DAG的方式产生Job，并通过JobScheduler的线程池的方式提交给SparkCluster不断地执行。数据不断地流进来，Spark Streaming不断地产生Job，同时也在不断地积攒数据，积攒数据的方式就是通过Batch Interval，比如1秒钟。在这1秒钟，一般会有很多的event或者说是数据，如果我们基于Flume或者Kafka的方式，我们就会感受到很多event的存在。比如看Flume这种方式，如果100毫秒产生一个event，那么1秒钟就有10个event，这个event就是代表数据。这样就构成了一个数据的集合。而RDD处理就是基于固定不变的数据集合。随着时间的积累，会积累很多的event。1秒钟积累10个event，那么1秒钟的RDD就是这些数据集合生成的。Spark中的Job有些能做到50毫秒级别的运行，只是现在的Spark调度机制导致了在生产环境中，大多数Job大于500毫秒的样子。我们已经找到了控制Job在100毫秒以内的方式。由于时间固定，所以每个时间间隔产生的RDD是固定的。每个RDDDAG的依赖关系基于时间间隔内的一个batch的数据。虽然图中每个时间间隔有2个DStream（2个B），但是这个时间间隔内抓取的数据都属于这个batch的数据。这里图中的DAG依赖关系为什么可以产生3个Job？因为它是RDDDAG依赖，而RDD DAG依赖不会限定产生多少个Job。所以我们这个DAG中，它处理的就是我们这个时间间隔的这1秒钟的batch，图中有4个连续的1秒，所以有4个对应的batch。从图中看，空间维度是一样的，只不过是时间维度（特指时间点）的不同，导致了流入数据的不同，从而处理的数据量和数据内容的不同，所以导致了不同的RDDDAG依赖关系的实例。而由于RDD Graph脱胎于DStreamGraph，所以DStream Graph就是RDDGraph的模版。

注意：随着时间的流逝，SparkStreaming会继续处理，即使没有输入数据，只不过没有有效输出而已。

![SouthEast 4][]

我们现在看看从SparkStreaming角度出发，架构中应该有些什么要点：

> > **1) 需要RDD DAG的生成模版：DStreamGraph**

由于Spark Streaming基于SparkRDD，那么必须有一种东西表示RDD DAG这种处理逻辑，就是空间维度。

![SouthEast 5][]

> > **2) 需要基于Timeline的Job控制**
> > 
> > **3) 需要InputStreamings和OutputStreaming，代表数据的输入和输出**

DStreamGraph.scala:

privateval inputStreams = new ArrayBuffer\[InputDStream\[\_\]\]()  
private val outputStreams = new ArrayBuffer\[DStream\[\_\]\]()

InputDStream.scala:

abstractclass InputDStream\[T: ClassTag\] (ssc\_ : StreamingContext)  
extends DStream\[T\](ssc\_) \{

> > **4) 具体的Job运行在Spark Cluster之上，此时系统容错就至关重要**

\- SparkStreaming在流量过大时能限流，能动态的调整CPU、内存等资源。

> > **5) 需要事务处理**

\- 我们希望流进来的数据一定会被处理，而且只处理一次。在处理出现崩溃的情况下如何保证Exactly once的事务语义。

在结束这节课之前，我们初步看看DStream。

![SouthEast 6][]

上图中，从前面2个红色框中的代码可以看出，DStream就是Spark Streaming的核心，就像RDD是Spark Core的核心，它也有dependencies和compute。

而第3个红色框中的代码表明generatedRDDs是一个HashMap，以时间为key，RDD数组为value。所以说DStream就是RDD的模版。DStream可以说是逻辑级别的，RDD就是物理级别的，DStream所表达的最终都是转换成RDD去实现。前者是更高级别的抽象，后者是底层的实现。DStream实际上就是在时间维度上对RDD集合的封装，DStream与RDD的关系就是随着时间流逝不断的产生RDD，对DStream的操作就是在固定时间上操作RDD。

DStream有很多子类。

![SouthEast 7][]

DStream间的转换操作就是子类的转换。也实际上是RDD的转换，然后产生依赖关系的Job，并通过JobScheduler在集群上运行。

**总结：**

在空间维度上的业务逻辑作用于DStream，随着时间的流逝，每个BatchInterval形成了具体的数据集，产生了RDD，对RDD进行transform操作，进而形成了RDD的依赖关系RDD DAG，形成Job。然后JobScheduler根据时间调度，基于RDD的依赖关系，把作业发布到Spark集群上去运行，不断的产生Spark作业。

[http_spark.apache.org]: http://spark.apache.org
[SouthEast]: /images/20220724/67be5087481647f8be80bed2b4d18d7f.png
[SouthEast 1]: /images/20220724/c1263ff2ee3e4d66b2a03fc3e2331c53.png
[SouthEast 2]: /images/20220724/22357d48edbd4493bfe387c24b73bd37.png
[SouthEast 3]: /images/20220724/dde10e19ea0442bbaf6b91b7b85a0fa2.png
[SouthEast 4]: /images/20220724/8af1bcfa6a324429bb0e00b4d48bec99.png
[SouthEast 5]: /images/20220724/a9f150217654489f8a1c8f7f1f967a86.png
[SouthEast 6]: /images/20220724/c6b82fc496564de7b68d2e268658c411.png
[SouthEast 7]: /images/20220724/9b41401ac4524c52a40edc37e18c9996.png