专栏首页Spark生态圈[spark streaming] DStream 和 DStreamGraph 解析

[spark streaming] DStream 和 DStreamGraph 解析

看 spark streaming 源码解析之前最好先了解spark core的内容。

前言

Spark Streaming 是基于Spark Core将流式计算分解成一系列的小批处理任务来执行。

在Spark Streaming里,总体负责任务的动态调度是JobScheduler,而JobScheduler有两个很重要的成员:JobGeneratorReceiverTrackerJobGenerator 负责将每个 batch 生成具体的 RDD DAG ,而ReceiverTracker负责数据的来源。

Spark Streaming里的DStream可以看成是Spark Core里的RDD的模板,DStreamGraph是RDD DAG的模板。

跟着例子看流程

DStream 也和 RDD 一样有着转换(transformation)和 输出(output)操作,通过 transformation 操作会产生新的DStream,典型的transformation 操作有map(), filter(), reduce(), join()等。RDD的输出操作会触发action,而DStream的输出操作也会新建一个ForeachDStream,用一个函数func来记录所需要做的操作。

下面看一个例子:

val conf = new SparkConf().setMaster("local[2]")
                          .setAppName("NetworkWordCount")
val ssc = new StreamingContext(conf, Seconds(1))
val lines = ssc.socketTextStream("localhost", 9999)
val words = lines.flatMap(_.split(" "))      
val pairs = words.map(word => (word, 1))    
val wordCounts = pairs.reduceByKey(_ + _)   
wordCounts.print()
ssc.start()
ssc.awaitTermination()

在创建 StreamingContext 的时候实创建了 graph: DStreamGraph:

private[streaming] val graph: DStreamGraph = {
    if (isCheckpointPresent) {
      _cp.graph.setContext(this)
      _cp.graph.restoreCheckpointData()
      _cp.graph
    } else {
      require(_batchDur != null, "Batch duration for StreamingContext cannot be null")
      val newGraph = new DStreamGraph()
      newGraph.setBatchDuration(_batchDur)
      newGraph
    }
  }

checkpoint 可用,会优先从 checkpoint 恢复 graph,否则新建一个。graph用来动态的创建RDD DAG,DStreamGraph有两个重要的成员:inputStreamsoutputStreams

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

Spark Streaming记录DStream DAG 的方式就是通过DStreamGraph实例记录所有的outputStreams ,因为outputStream会通过依赖 dependencies 来和parent DStream形成依赖链,通过outputStreams 向前追溯遍历就可以得到所有上游的DStream,另外,DStreamGraph 还会记录所有的inputStreams ,避免每次为查找 input stream 而对 output steam 进行 BFS 的消耗。

继续回到例子,这里通过ssc.socketTextStream 创建了一个ReceiverInputDStream,在其父类 InputDStream 中会将该ReceiverInputDStream添加到inputStream里。

接着调用了flatMap方法:

def flatMap[U: ClassTag](flatMapFunc: T => TraversableOnce[U]): DStream[U] = ssc.withScope {
    new FlatMappedDStream(this, context.sparkContext.clean(flatMapFunc))
  }

--------------------------------------------------------------------

private[streaming]
class FlatMappedDStream[T: ClassTag, U: ClassTag](
    parent: DStream[T],
    flatMapFunc: T => TraversableOnce[U]
  ) extends DStream[U](parent.ssc) {

  override def dependencies: List[DStream[_]] = List(parent)

  override def slideDuration: Duration = parent.slideDuration

  override def compute(validTime: Time): Option[RDD[U]] = {
    parent.getOrCompute(validTime).map(_.flatMap(flatMapFunc))
  }
}

创建了一个 FlatMappedDStream ,而该类的compute方法是在父 DStream(ReceiverInputDStream) 在对应batch时间的RDD上调用了flatMap方法,也就是构造了 rdd.flatMap(func)这样的代码,后面的操作类似,随后形成的是rdd.flatMap(func1).map(func2).reduceByKey(func3).take(),这不就是我们spark core里的东西吗。另外其dependencies是直接指向了其构造参数parent,也就是刚才的ReceiverInputDStream,每个新建的DStream的dependencies都是指向了其父DStream,这样就构成了一个依赖链,也就是形成了DStream DAG。

这里我们再看看最后的 print() 操作:

----
def print(num: Int): Unit = ssc.withScope {
    def foreachFunc: (RDD[T], Time) => Unit = {
      (rdd: RDD[T], time: Time) => {
        val firstNum = rdd.take(num + 1)
        // scalastyle:off println
        println("-------------------------------------------")
        println(s"Time: $time")
        println("-------------------------------------------")
        firstNum.take(num).foreach(println)
        if (firstNum.length > num) println("...")
        println()
        // scalastyle:on println
      }
    }
    foreachRDD(context.sparkContext.clean(foreachFunc), displayInnerRDDOps = false)
  }
----
private def foreachRDD(
      foreachFunc: (RDD[T], Time) => Unit,
      displayInnerRDDOps: Boolean): Unit = {
    new ForEachDStream(this,
      context.sparkContext.clean(foreachFunc, false), displayInnerRDDOps).register()
  }
----
#ForEachDStream
override def generateJob(time: Time): Option[Job] = {
    parent.getOrCompute(time) match {
      case Some(rdd) =>
        val jobFunc = () => createRDDWithLocalProperties(time, displayInnerRDDOps) {
          foreachFunc(rdd, time)
        }
        Some(new Job(time, jobFunc))
      case None => None
    }
  }

在print() 方法里构建了一个foreachFunc方法:对一个rdd进行了take操作并打印(spark core中的action操作)。随后创建了ForEachDStream实例并调用了register()方法:

 private[streaming] def register(): DStream[T] = {
    ssc.graph.addOutputStream(this)
    this
  }

将 OutputStream 添加到DStreamGraphoutputStreams 里。可以看到刚才构建的 foreachFunc 方法最终用在了ForEachDStream实例的generateJob方法里,并创建了一个Streaming 中的Job,在job中的run方法中会调用这个方法,也就是会触发action操作。

注意这里Spark Streaming的Job和Spark Core里的Job是不一样的,Streaming的Job执行的是前面构造的方法,方法里面是Core里的Job,方法可以定义多个core里的Job,也可以一个core里的job都没有。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • [spark] DAGScheduler划分stage源码解析

    Spark Application只有遇到action操作时才会真正的提交任务并进行计算,DAGScheduler 会根据各个RDD之间的依赖关系形成一个DAG...

    UFO
  • [spark] Standalone模式下Master、WorKer启动流程

    而Standalone 作为spark自带cluster manager,需要启动Master和Worker守护进程,本文将从源码角度解析两者的启动流程。Mas...

    UFO
  • [spark] 从spark-submit开始解析整个任务调度流程

    spark应用程序可以以Client模式和Cluster启动,区别在于Client模式下的Driver是在执行spark-submit命令节点上启动的,而Clu...

    UFO
  • 【哈工大】Dynamic ReLU:自适应参数化ReLU及Keras代码(调参记录23)Cifar10~95.47%

    本文在调参记录21的基础上,增加卷积核的个数,也就是增加深度神经网络的宽度,继续尝试深度残差网络+自适应参数化ReLU激活函数在Cifar10数据集上的效果。

    用户7368967
  • 使用腾讯云 GPU 学习深度学习系列之五:文字的识别与定位

    本系列文章主要介绍如何使用腾讯云GPU服务器进行深度学习运算,前面主要介绍原理部分,后期则以实践为主。

    集智
  • 数据分析 ——— pandas基础(三)

    接着之前的文章,在这里我们来看一些利用pandas处理文本数据,利用索引,loc, iloc,ix,属性选取数据

    andrew_a
  • 【哈工大】Dynamic ReLU:自适应参数化ReLU及Keras代码(调参记录24)Cifar10~95.80%

    本文在调参记录23的基础上,增加卷积核的个数,最少是64个,最多是256个,继续测试深度残差网络+自适应参数化ReLU激活函数在cifar10数据集上的效果。

    用户7368967
  • 【哈工大】Dynamic ReLU:自适应参数化ReLU及Keras代码(调参记录22)Cifar10~95.25%

    本文在调参记录21的基础上,将残差模块的个数,从60个增加到120个,测试深度残差网络+自适应参数化ReLU激活函数在Cifar10图像集上的效果。

    用户7368967
  • 【哈工大】Dynamic ReLU:自适应参数化ReLU及Keras代码(调参记录21)Cifar10~95.12%

    本文在调参记录20的基础上,将残差模块的个数,从27个增加到60个,继续测试深度残差网络ResNet+自适应参数化ReLU激活函数在Cifar10数据集上的表现...

    用户7368967
  • 使用Sonar分析Ant/Maven/Gradle项目

    目前Java平台三大构建工具是:Apache Ant、Apache Maven、Gradle,当然还有其他几个高效、快速的编译工具,如:Google Ba...

    donghui

扫码关注云+社区

领取腾讯云代金券