Spark Streaming(1) - 基本原理

2019年6月8日 242次阅读 来源: aaron1993

本文基于spark 2.11

1. 前言

spark使用RDD来抽象的表示数据，用户使用RDD提供的一些算子编写自己的spark application，使用RDD抽象表示数据要求对于输入数据是静态的，但是在流式数据处理中数据如同流水一样不停的在管道中产生，这不符合RDD的要求。Spark Streaming的处理方式是，从输入流中读区数据，将数据作为一个个batch保存起来，这样就有了静态的数据，就可以用RDD来表示这些数据，然后就可以基于RDD 创建任务了。

2. 基本原理

下面是一个从kafka读取数据处理的代码：

import org.apache.spark._
import org.apache.spark.streaming._

val conf = new SparkConf().setMaster("local[2]").setAppName("KafkaWordCount")
// batchDuration 设置为 1 秒，然后创建一个 streaming 入口
// 每1秒依据RDD中创建一次job，输入RDD就从已经已经收集的batch中取。
val ssc = new StreamingContext(conf, Seconds(1))

val kafkaParams: Map[String, String] = Map("group.id" -> "test",...)
val topics = Map("test" -> 1)
val lines = KafkaUtils.createStream(
      ssc,
      kafkaParams,
      topics,  StorageLevel.MEMORY_AND_DISK)
val words = lines.flatMap(_.split(" "))     
val pairs = words.map(word => (word, 1))    
val wordCounts = pairs.reduceByKey(_ + _)   
wordCounts.print()                      
wordCounts.foreachRDD(...)     
ssc.start()
ssc.awaitTermination()

和之前基于RDD的的wordcount程序不同：

  1. KafkaUtils.createStream(…)创建出来的不是RDD，和是一个DStream的类，
  2. DStream同样存在map、flatMap、reduceByKey这样的转换操作，但是它是从DStream到DStream的转换。
  3. print在RDD里表示一种action，会触发job的创建和提交，但是DStream的action操作不会，它的处理方式不同，后续会介绍。
  4. ssc.start会启动一下组建：
    • JobScheduler, 调度和追踪job
    • JobGenerator，由JobScheduler启动，定时（初始化StreamingContext指定的时间）从DStream创建出job
    • ReceiverTracker， 运行在Driver上，收集 从各个receiver上报的流数据batch信息
    • ReceiverSupervisor，由ReceiverTracker运行发送消息使其在executor上运行，接收Receiver汇报的batch数据，然后将数据信息汇报给ReceiverTracker。
    • Receiver， 运行在executor上，由ReceiverSupervisor启动，负责着流中读区数据，分成batch，汇报给ReceiverSupervisor。

从上介绍可以看出，job是在ssc.start过程中创建的，而且在运行期间会根据用户设置的duration不断的创建。

下图表示了使用kafka作为输入源时的streaming工作期间的流程：

《Spark Streaming(1) - 基本原理》 图1 streaming工作流程

  1. Receiver1从kafka中读入数据，将数据转发给ReceiverSupervisor
  2. ReceiverSupervisor，使用BlockManager存储并管理数据信息。
  3. ReceiverSupervisor，将数据信息发送给运行在Driver上的ReceiverTracker。
  4. JobGenerator，假设上面wordCount代码中DStream之间的转换看作一张DAG，DStreamGraph保存了所有的DAG。JobGenerate每隔一段时间从DStreamGraph中的DStream DAG生成RDD的DAG，然后提交RDD的job。RDD的数据则来自Generator根据ReceiverTracker中收集的batch数据信息。

2.1 DStream 到RDD的转换

由于基于RDD计算的基于静态的数据，而数据是不断产生的，spark streaming将输入数据切成一个个batch，因此需要不断的产生job去计算batch中的数据。

上面wordCount程序，描述了DStream之间的转换，看起来几乎和RDD之间的转换是一样的，JobGenerator运行期间根据DStream不停创建RDD，再由RDD生成job 经SparkContext提交运行。DStream相当于模板，RDD相当于使用模板创造出的零件，而JobGenerator则相当于操作模板的工人了。

下图描述了DStream和RDD在运行期间的关系：

《Spark Streaming(1) - 基本原理》 DStream和RDD

可以看到DStream的子类都有一个RDD的对应类，一句DStream生成的RDD DAG和DStream拥有一样的转换和依赖。采集输入流中的一段数据作为RDD的源数据。

RDD#compute方法中完成输入数据的计算，DStream也存在compute方法，但是其compute方法这是完成DStream到RDD的转换。

2.2 ReceiverInputDStream

所有继承DStream的类中，ReceiverInputDStream除了像其他DStream一样创建出RDD以外，还需要返回一个Receiver负责接收收据，例如ReceiverInputDStream的子类SocketInputDStream就能返回一个SocketReceiver的Receiver的实现类。

ReceiverInputDStream一般都是一个DStream DAG的源头。

当ReceiverTracker调用start启动时，它会从DStreamGraph持有的DStream DAG中获得所有的ReceiverInputDStream，然后取得Receiver，通过巧妙的方式将Reciver包装成Task，然后发送到executor上执行，然后在receiver端，Receiver和ReceiverSupervisor启动接收数据。

在SparkStreaming(3) ReceiverTracker和Receiver中，启动receiver时，receiver就是按上面方式获得的。

2.3 output 操作

DStream和RDD有着类似的操作，map这种使得RDD转换成新的RDD的操作称为Transformation，foreach这种触发job的创建和提交的操作称为Action， DStream类似，Dstream到DStream的称为Transformation, DStream的output操作有点类似rdd中的action操作，一个action意味着一个新的job被创建提交。DStream的output操作意味着一个DStream DAG模板的创建，也意味着到此处DStream转换成RDD应该触发job，DStream常见的output操作有：

  1. saveAsTextFiles
  2. saveAsObjectFiles
  3. print
  4. foreachRDD
    等

3 DStreamGraph

DStreamGraph用来保存所有output操作生成DStream DAG。
比如下面是DStream#foreachRDD的代码：

private def foreachRDD(
      foreachFunc: (RDD[T], Time) => Unit,
      displayInnerRDDOps: Boolean): Unit = {
   // 调用了DStream#register()方法
    new ForEachDStream(this,
      context.sparkContext.clean(foreachFunc, false), displayInnerRDDOps).register()
  }

//register方法向DStreamGraph注册当前DStream
// 由于DStream保存了所有的父依赖，因此注册当前DStream
// 就能追溯出整个DStream DAG，相当于注册了DStream DAG
private[streaming] def register(): DStream[T] = {
    ssc.graph.addOutputStream(this)
    this
  }

上面说DStream的output操作相当于触发一个DStream DAG模板的创建，而一个模板对应一种job。 第一节wordcount代码中分别有print和foreachRDD两个output操作，因此DStreamGraph可以理解持有两个DStream DAG，如下图：

《Spark Streaming(1) - 基本原理》

尽管创建出来的DStream DAG是一样的，但是依然会创建出两份RDD DAG，生成两类job，

DStreamGraph还肩负着从根据注册的DStreamDAG创建job的任务，后续JobGenerator就是调用DStreamGraph完成创建job。下面是DstreamGraph创建job的方法：

def generateJobs(time: Time): Seq[Job] = {
    logDebug("Generating jobs for time " + time)
    val jobs = this.synchronized {
      // 对每一个因output操作而注册的DStream DAG生成job
      outputStreams.flatMap { outputStream =>
        val jobOption = outputStream.generateJob(time)
        jobOption.foreach(_.setCallSite(outputStream.creationSite))
        jobOption
      }
    }
    logDebug("Generated " + jobs.length + " jobs for time " + time)
    jobs
  }


这是DStream#generateJob方法，time表示每一次生成job的时间

private[streaming] def generateJob(time: Time): Option[Job] = {
   // getOrCompute将DStream转换成RDD，转换操作是从当前
   // DStream往上游追溯，追溯到源头后在一次往下生成RDD的过程
   // 是一次DFS的过程。
    getOrCompute(time) match {
      case Some(rdd) =>
        val jobFunc = () => {
          val emptyFunc = { (iterator: Iterator[T]) => {} }
         // 创建到RDD后提交job
          context.sparkContext.runJob(rdd, emptyFunc)
        }
        Some(new Job(time, jobFunc))
      case None => None
    }
  }
    原文作者：aaron1993
    原文地址: https://www.jianshu.com/p/b3b7cc54b352
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。
点赞