乐趣区

关于spark:第五篇SparkStreaming编程指南2

作者：

jiezi

在

第四篇|Spark-Streaming编程指南(1)对Spark Streaming执行机制、Transformations与Output Operations、Spark Streaming数据源(Sources)、Spark Streaming 数据汇(Sinks)进行了探讨。本文将连续上篇内容，次要包含以下内容：

  • 有状态的计算
  • 基于工夫的窗口操作
  • 长久化
  • 检查点Checkpoint
  • 应用DataFrames & SQL解决流数据

有状态的计算

updateStateByKey

上一篇文章中介绍了常见的无状态的转换操作，比方在WordCount的例子中，输入的后果只与以后batch interval的数据无关，不会依赖于上一个batch interval的计算结果。spark Streaming也提供了有状态的操作： updateStateByKey，该算子会保护一个状态，同时进行信息更新 。该操作会读取上一个batch interval的计算结果，而后将其后果作用到以后的batch interval数据统计中。其源码如下：

def updateStateByKey[S: ClassTag](
      updateFunc: (Seq[V], Option[S]) => Option[S]
    ): DStream[(K, S)] = ssc.withScope {
    updateStateByKey(updateFunc, defaultPartitioner())
  }

该算子只能在key–value对的DStream上应用，须要接管一个状态更新函数 updateFunc作为参数。应用案例如下：

object StateWordCount {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setMaster("local[2]")
      .setAppName(StateWordCount.getClass.getSimpleName)
    val ssc = new StreamingContext(conf, Seconds(5))
    // 必须开启checkpoint,否则会报错
    ssc.checkpoint("file:///e:/checkpoint")
    val lines = ssc.socketTextStream("localhost", 9999)

    // 状态更新函数
    def updateFunc(newValues: Seq[Int], stateValue: Option[Int]): Option[Int] = {

      var oldvalue = stateValue.getOrElse(0) // 获取状态值
      // 遍历以后数据，并更新状态
      for (newValue <- newValues) {
        oldvalue += newValue
      }
      // 返回最新的状态
      Option(oldvalue)
    }

    val count = lines.flatMap(_.split(" "))
      .map(w => (w, 1))
      .updateStateByKey(updateFunc)
    count.print()
    ssc.start()
    ssc.awaitTermination()
  }

}

尖叫提醒：下面的代码必须要开启checkpoint，否则会报错：

Exception in thread “main” java.lang.IllegalArgumentException: requirement failed: The checkpoint directory has not been set. Please set it by StreamingContext.checkpoint()

updateStateByKey毛病

运行下面的代码会发现一个景象：即使没有数据源输出，Spark也会为新的batch interval更新状态，即如果没有数据源输出，则会一直地输入之前的计算状态后果。

updateStateByKey能够在指定的批次距离内返回之前的全副历史数据，包含新增的，扭转的和没有扭转的。因为updateStateByKey在应用的时候肯定要做checkpoint，当数据量过大的时候，checkpoint会占据宏大的数据量，会影响性能，效率不高。

mapwithState

mapwithState是Spark提供的另外一个有状态的算子，该操作克服了updateStateByKey的毛病，从Spark 1.5开始引入。源码如下：

def mapWithState[StateType: ClassTag, MappedType: ClassTag](
      spec: StateSpec[K, V, StateType, MappedType]
    ): MapWithStateDStream[K, V, StateType, MappedType] = {
    new MapWithStateDStreamImpl[K, V, StateType, MappedType](
      self,
      spec.asInstanceOf[StateSpecImpl[K, V, StateType, MappedType]]
    )
  }

mapWithState只返回发生变化的key的值，对于没有发生变化的Key，则不返回。这样做能够只关怀那些曾经产生的变动的key，对于没有数据输出，则不会返回那些没有变动的key 的数据。这样的话，即便数据量很大，checkpint也不会updateBykey那样，占用太多的存储，效率比拟高（生产环境中倡议应用）。

object StatefulNetworkWordCount {
  def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf()
      .setAppName("StatefulNetworkWordCount")
      .setMaster("local[2]")

    val ssc = new StreamingContext(sparkConf, Seconds(5))
    ssc.checkpoint("file:///e:/checkpoint")

    val lines = ssc.socketTextStream("localhost", 9999)
    val words = lines.flatMap(_.split(" "))
    val wordDstream = words.map(x => (x, 1))
    /**
      * word：以后key的值
      * one：以后key对应的value值
      * state：状态值
      */
    val mappingFunc = (batchTime: Time, word: String, one: Option[Int], state: State[Int]) => {
      val sum = one.getOrElse(0) + state.getOption.getOrElse(0)
      println(s">>> batchTime = $batchTime")
      println(s">>> word      = $word")
      println(s">>> one     = $one")
      println(s">>> state     = $state")
      val output = (word, sum)
      state.update(sum) //更新以后key的状态值
      Some(output) //返回后果
    }
    // 通过StateSpec.function构建StateSpec
    val spec = StateSpec.function(mappingFunc)
    val stateDstream = wordDstream.mapWithState(spec)
    stateDstream.print()
    ssc.start()
    ssc.awaitTermination()
  }
}

基于工夫的窗口操作

Spark Streaming提供了两种类型的窗口操作，别离是滚动窗口和滑动窗口。具体分析如下：

滚动窗口(Tumbling Windows)

滚动窗口的示意图如下：滚动窗口只须要传入一个固定的工夫距离，滚动窗口是不存在重叠的。

源码如下：

/**
   * @param windowDuration:窗口的长度; 必须是batch interval的整数倍.
   */
  def window(windowDuration: Duration): DStream[T] = window(windowDuration, this.slideDuration)

滑动窗口(Sliding Windows)

滑动窗口的示意图如下：滑动窗口只须要传入两个参数，一个为窗口的长度，一个是滑动工夫距离。能够看出：滑动窗口是存在重叠的。

源码如下：

/**
   * @param windowDuration 窗口长度;必须是batching interval的整数倍
   *                       
   * @param slideDuration  滑动距离;必须是batching interval的整数倍
   */
  def window(windowDuration: Duration, slideDuration: Duration): DStream[T] = ssc.withScope {
    new WindowedDStream(this, windowDuration, slideDuration)
  }

窗口操作

  • window(windowLength, slideInterval)

    • 解释

      基于源DStream产生的窗口化的批数据，计算失去一个新的Dstream

    • 源码

        def window(windowDuration: Duration): DStream[T] = window(windowDuration, this.slideDuration)
  def window(windowDuration: Duration, slideDuration: Duration): DStream[T] = ssc.withScope {
    new WindowedDStream(this, windowDuration, slideDuration)
  }

  • countByWindow(windowLength, slideInterval)

    • 解释

    返回一个滑动窗口的元素个数

    • 源码

      /**
   * @param windowDuration window长度，必须是batch interval的倍数 
   * @param slideDuration  滑动的工夫距离，必须是batch interval的倍数
   */
  def countByWindow(
      windowDuration: Duration,
      slideDuration: Duration): DStream[Long] = ssc.withScope {
    this.map(_ => 1L).reduceByWindow(_ + _, _ - _, windowDuration, slideDuration)
  }
```

  • reduceByWindow(func, windowLength, slideInterval)

    • 解释

    返回一个单元素流。利用函数func汇集滑动工夫距离的流的元素创立这个单元素流。函数func必须满足结合律，从而能够反对并行计算

    • 源码

        def reduceByWindow(
      reduceFunc: (T, T) => T,
      windowDuration: Duration,
      slideDuration: Duration
    ): DStream[T] = ssc.withScope {
    this.reduce(reduceFunc).window(windowDuration, slideDuration).reduce(reduceFunc)
  }

  • reduceByKeyAndWindow(func, windowLength, slideInterval, [numTasks])

    • 解释

    利用到一个(K,V)键值对组成的DStream上时，会返回一个由(K,V)键值对组成的新的DStream。每一个key的值均由给定的reduce函数(func函数)进行聚合计算。留神：在默认状况下，这个算子利用了Spark默认的并发工作数去分组。能够通过numTasks参数的设置来指定不同的工作数

    • 源码

        def reduceByKeyAndWindow(
      reduceFunc: (V, V) => V,
      windowDuration: Duration,
      slideDuration: Duration
    ): DStream[(K, V)] = ssc.withScope {
    reduceByKeyAndWindow(reduceFunc, windowDuration, slideDuration, defaultPartitioner())
  }

  • reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, [numTasks])

    • 解释

    更加高效的reduceByKeyAndWindow，每个窗口的reduce值，是基于先前窗口的reduce值进行增量计算失去的；它会对进入滑动窗口的新数据进行reduce操作，并对来到窗口的老数据进行逆向reduce操作。然而，只能用于可逆reduce函数，即那些reduce函数都有一个对应的逆向reduce函数（以InvFunc参数传入）留神：必须开启 checkpointing

    • 源码

      def reduceByKeyAndWindow(
      reduceFunc: (V, V) => V,
      invReduceFunc: (V, V) => V,
      windowDuration: Duration,
      slideDuration: Duration,
      partitioner: Partitioner,
      filterFunc: ((K, V)) => Boolean
    ): DStream[(K, V)] = ssc.withScope {

    val cleanedReduceFunc = ssc.sc.clean(reduceFunc)
    val cleanedInvReduceFunc = ssc.sc.clean(invReduceFunc)
    val cleanedFilterFunc = if (filterFunc != null) Some(ssc.sc.clean(filterFunc)) else None
    new ReducedWindowedDStream[K, V](
      self, cleanedReduceFunc, cleanedInvReduceFunc, cleanedFilterFunc,
      windowDuration, slideDuration, partitioner
    )
  }

  • countByValueAndWindow(windowLength, slideInterval, [numTasks])

    • 解释

      当利用到一个(K,V)键值对组成的DStream上，返回一个由(K,V)键值对组成的新的DStream。每个key的对应的value值都是它们在滑动窗口中呈现的频率

    • 源码

      def countByValueAndWindow(
      windowDuration: Duration,
      slideDuration: Duration,
      numPartitions: Int = ssc.sc.defaultParallelism)
      (implicit ord: Ordering[T] = null)
      : DStream[(T, Long)] = ssc.withScope {
    this.map((_, 1L)).reduceByKeyAndWindow(
      (x: Long, y: Long) => x + y,
      (x: Long, y: Long) => x - y,
      windowDuration,
      slideDuration,
      numPartitions,
      (x: (T, Long)) => x._2 != 0L
    )
  }

应用案例

val lines = ssc.socketTextStream("localhost", 9999)

    val count = lines.flatMap(_.split(" "))
      .map(w => (w, 1))
      .reduceByKeyAndWindow((w1: Int, w2: Int) => w1 + w2, Seconds(30), Seconds(10))
      .print()
//滚动窗口

/*    lines.window(Seconds(20))
      .flatMap(_.split(" "))
      .map((_, 1))
      .reduceByKey(_ + _)
      .print()*/

长久化

长久化是晋升Spark利用性能的一种形式，在第二篇|Spark core编程指南一文中解说了RDD长久化的应用形式。其实，DStream也是反对长久化的，同样是应用persist()与cache()办法，长久化通常在有状态的算子中应用，比方窗口操作，默认状况下，尽管没有显性地调用长久化办法，然而底层曾经帮用户做了长久化操作，通过上面的源码能够看出。

private[streaming]
class WindowedDStream[T: ClassTag](
    parent: DStream[T],
    _windowDuration: Duration,
    _slideDuration: Duration)
  extends DStream[T](parent.ssc) {
  // 省略代码...
  // Persist parent level by default, as those RDDs are going to be obviously reused.
  parent.persist(StorageLevel.MEMORY_ONLY_SER)
}

留神：与RDD的长久化不同，DStream的默认持久性级别将数据序列化在内存中，通过上面的源码能够看出：

/** 给定一个持打算级别 */
  def persist(level: StorageLevel): DStream[T] = {
    if (this.isInitialized) {
      throw new UnsupportedOperationException(
        "Cannot change storage level of a DStream after streaming context has started")
    }
    this.storageLevel = level
    this
  }

  /** 默认的长久化级别为(MEMORY_ONLY_SER) */
  def persist(): DStream[T] = persist(StorageLevel.MEMORY_ONLY_SER)
  def cache(): DStream[T] = persist()

从下面的源码能够看出persist()与cache()的次要区别是：

  • cache()办法底层调用的是persist()办法

  • persist()办法有两个重载的办法

    • 无参数的persist()，默认是内存
    • perisist(level: StorageLevel),能够抉择与RDD长久化雷同的长久化级别

检查点Checkpoint

简介

流应用程序通常是24/7运行的，因而必须对与利用程序逻辑无关的故障（例如系统故障，JVM解体等）具备弹性的容错能力。为此，Spark Streaming须要将足够的信息checkpoint到容错存储系统(比方HDFS)，以便能够从故障中复原。检查点包含两种类型：

  • 元数据检查点

    元数据检查点能够保障从Driver程序失败中复原。即如果运行drive的节点失败时，能够查看最近的checkpoin数据获取最新的状态。典型的应用程序元数据包含：

    • 配置 ：用于创立流应用程序的配置。
    • DStream操作 ：定义流应用程序的DStream操作。
    • 未实现的batch ：以后运行batch对应的job在队列中排队，还没有计算到该batch的数据。
  • 数据检查点

    将生成的RDD保留到牢靠的存储中。在某些有状态转换中，须要合并多个批次中的数据，所以须要开启检查点。在此类转换中，生成的RDD依赖于先前批次的RDD，这导致依赖链的长度随工夫一直减少。为了防止复原工夫无限度的减少（与依赖链成比例），有状态转换的两头RDD定期 checkpoint到牢靠的存储（例如HDFS），以切断依赖链，性能相似于长久化，只须要从以后的状态复原，而不须要从新计算整个lineage。

总而言之，从Driver程序故障中复原时，次要须要元数据检查点。而如果应用有状态转换，则须要数据或RDD检查点。

什么时候启用检查点

必须为具备以下类型的应用程序启用检查点：

  • 应用了有状态转换转换操作

    如果在应用程序中应用updateStateByKey或reduceByKeyAndWindow，则必须提供检查点目录以容许定期进行RDD检查点。

  • 从运行应用程序的Driver程序故障中复原

    元数据检查点用于复原进度信息。

留神，没有前述状态转换的简略流应用程序能够在不启用检查点的状况下运行。在这种状况下，从驱动程序故障中复原也将是局部的（某些失落但未解决的数据可能会失落）。这通常是能够承受的，并且许多都以这种形式运行Spark Streaming应用程序。预计未来会改善对非Hadoop环境的反对。

如何配置检查点

能够通过具备容错的、牢靠的文件系统（例如HDFS，S3等）中设置目录来启用检查点，将检查点信息保留到该目录中。开启检查点，须要开启上面的两个配置：

  • streamingContext.checkpoint(<dir>)：配置检查点的目录，比方HDFS门路
  • dstream.checkpoint(<duration>)：检查点的频率

其中配置检查点的工夫距离是可选的。如果不设置，会依据DStream的类型抉择一个默认值。对于MapWithStateDStream，默认的检查点距离是batch interval的10倍。对于其余的DStream，默认的检查点距离是10S，或者是batch interval的间隔时间。须要留神的是：checkpoint的频率必须是 batch interval的整数倍，否则会报错。

此外，如果要使应用程序从Driver程序故障中复原，则须要应用上面的形式创立StreamingContext：

def createStreamingContext (conf: SparkConf,checkpointPath: String):
StreamingContext = {
val ssc = new StreamingContext( <ConfInfo> )
// .... other code ...
ssc.checkPoint(checkpointDirectory)
ssc
}
#创立一个新的StreamingContext或者从最近的checkpoint获取
val context = StreamingContext.getOrCreate(checkpointDirectory,
createStreamingContext _)
#启动
context.start()
context.awaitTermination()
  • 程序首次启动时，它将创立一个新的StreamingContext，而后调用start（）。
  • 失败后重新启动程序时，它将依据检查点目录中的检查点数据从新创立StreamingContext。

留神：

RDD的检查点须要将数据保留到牢靠存储上，由此带来一些老本开销。这可能会导致RDD取得检查点的那些批次的解决工夫减少。因而，须要设置一个正当的检查点的距离。在batch interval较小时(例如1秒），每个batch interval都进行检查点可能会大大降低吞吐量。相同，检查点工夫距离太长会导致 lineage和工作规模减少，这可能会产生不利影响。对于须要RDD检查点的有状态转换，默认距离为batch interval的倍数，至多应为10秒。能够应用 dstream.checkpoint(checkpointInterval)进行配置。通常，DStream的5-10个batch interval的检查点距离是一个较好的抉择。

检查点和长久化之间的区别

  • 长久化

    • 当咱们将RDD放弃在DISK_ONLY存储级别时，RDD将存储在一个地位，该RDD的后续应用将不会从新计算lineage。
    • 在调用persist（）之后，Spark会记住RDD的lineage，即便它没有调用它。
    • 作业运行实现后，将革除缓存并销毁文件。

  • 检查点

    • 检查点将RDD存储在HDFS中，将会删除lineage血缘关系。
    • 在实现作业运行后，与持打算不同，不会删除检查点文件。
    • 当checkpoint一个RDD时，将导致双重计算。即该操作在实现理论的计算工作之前，首先会调用长久化办法，而后再将其写入检查点目录。

应用DataFrames & SQL解决流数据

在Spark Streaming利用中，能够轻松地对流数据应用DataFrames和SQL操作。应用案例如下：

object SqlStreaming {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName(SqlStreaming.getClass.getSimpleName)
      .setMaster("local[4]")
    val ssc = new StreamingContext(conf, Seconds(5))
    val lines = ssc.socketTextStream("localhost", 9999)
    val words = lines.flatMap(_.split(" "))

    words.foreachRDD { rdd =>
      // 调用SparkSession单例办法,如果曾经创立了，则间接返回
      val spark = SparkSessionSingleton.getInstance(rdd.sparkContext.getConf)
      import spark.implicits._

      val wordsDataFrame = rdd.toDF("word")
      wordsDataFrame.show()

      wordsDataFrame.createOrReplaceTempView("words")

      val wordCountsDataFrame =
        spark.sql("select word, count(*) as total from words group by word")
      wordCountsDataFrame.show()

    }


    ssc.start()
    ssc.awaitTermination()
  }
}
/** SparkSession单例 */
object SparkSessionSingleton {

  @transient private var instance: SparkSession = _

  def getInstance(sparkConf: SparkConf): SparkSession = {
    if (instance == null) {
      instance = SparkSession
        .builder
        .config(sparkConf)
        .getOrCreate()
    }
    instance
  }
}

总结

本文是Spark Streaming编程指南的第二篇分享，次要包含有状态的计算、基于工夫的窗口操作、检查点等内容。下一篇将分享Spark MLLib机器学习。

关注公众号大数据技术与数仓，及时理解最新动静

评论

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

这个站点使用 Akismet 来减少垃圾评论。了解你的评论数据如何被处理。

更多文章