一、概述

Spark Streaming 用于流式数据的处理。Spark Streaming 支持的数据输入源很多，例如：Kafka、 Flume、Twitter、ZeroMQ 和简单的 TCP 套接字等等。数据输入后可以用 Spark 的高度抽象原语如：map、reduce、join、window 等进行运算。而结果也能保存在很多地方，如 HDFS，数据库等。
和 Spark 基于 RDD 的概念很相似，Spark Streaming 使用离散化流(discretized stream)作为抽象表示，叫作DStream。DStream 是随时间推移而收到的数据的序列。在内部，每个时间区间收到的数据都作为 RDD 存在（微批次），而 DStream 是由这些RDD 所组成的序列(因此得名“离散化”)。所以简单来将，DStream 就是对 RDD 在实时数据处理场景的一种封装。

二、架构

Ø 整体架构图

Ø SparkStreaming 架构图

Master：记录Dstream之间的依赖关系或者血缘关系，并负责任务调度以生成新的RDD
Worker：从网络接收数据，存储并执行RDD计算
Client：负责向Spark Streaming中灌入数据

三、DsStream入门

3.1、WordCountDemo

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
object SparkStreaming01_WordCount {
    def main(args: Array[String]): Unit = {
        // TODO 创建环境对象
        // StreamingContext创建时，需要传递两个参数
        // 第一个参数表示环境配置
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
        // 第二个参数表示批量处理的周期（采集周期）
        val ssc = new StreamingContext(sparkConf, Seconds(3))
        // TODO 逻辑处理
        // 获取端口数据
        val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)
        val words = lines.flatMap(_.split(" "))
        val wordToOne = words.map((_,1))
        val wordToCount: DStream[(String, Int)] = wordToOne.reduceByKey(_+_)
        wordToCount.print()
        // 由于SparkStreaming采集器是长期执行的任务，所以不能直接关闭
        // 如果main方法执行完毕，应用程序也会自动结束。所以不能让main执行完毕
        //ssc.stop()
        // 1. 启动采集器
        ssc.start()
        // 2. 等待采集器的关闭
        ssc.awaitTermination()
    }
}

3.2、WordCount解析

Discretized Stream 是 Spark Streaming 的基础抽象，代表持续性的数据流和经过各种 Spark 原语操作后的结果数据流。在内部实现上，DStream 是一系列连续的 RDD 来表示（Dstream可以看做一组RDDs）。
每个RDD 含有一段时间间隔内的数据。
对数据的操作也是按照RDD 为单位来进行的：
计算过程由 Spark Engine 来完成：

四、DStream 转换

DStream 上的操作与 RDD 的类似，分为Transformations（转换）和Output Operations（输出）两种，此外转换操作中还有一些比较特殊的原语，如：updateStateByKey()、transform()以及各种Window 相关的原语。

4.1、无状态转化操作

无状态转化操作就是把简单的RDD 转化操作应用到每个批次上（只会对当前批次的数据进行统计处理），也就是转化DStream 中的每一个RDD。部分无状态转化操作列在了下表中。注意，针对键值对的DStream 转化操作(比如reduceByKey())要添加 import StreamingContext._ 才能在 Scala 中使用。

需要记住的是，尽管这些函数看起来像作用在整个流上一样，但事实上每个DStream 在内部是由许多RDD（批次）组成，且无状态转化操作是分别应用到每个RDD 上的。
例如：reduceByKey()会归约每个时间区间中的数据，但不会归约不同区间之间的数据。

4.1.2、Transform

Transform 允许 DStream 上执行任意的RDD-to-RDD 函数。即使这些函数并没有在DStream 的 API 中暴露出来（由于DStream的不足，需要使用底层的RDD api进行操作），通过该函数可以方便的扩展 Spark API。该函数每一批次调度一次。其实也就是对 DStream 中的 RDD 应用转换。
DStream功能性不完善时、需要代码周期性的执行。

package com.spark.stream
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
object Transform {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    //创建 StreamingContext
    val ssc = new StreamingContext(sparkConf, Seconds(3))
    //创建 DStream
    val lineDStream: ReceiverInputDStream[String] = ssc.socketTextStream("linux1", 9999)
    //转换为 RDD 操作
    val wordAndCountDStream: DStream[(String, Int)] = lineDStream.transform(rdd =>
    {
      val words: RDD[String] = rdd.flatMap(_.split(" "))  
      val wordAndOne: RDD[(String, Int)] = words.map((_, 1))
      val value: RDD[(String, Int)] = wordAndOne.reduceByKey(_ + _)
      value
    })
    //打印
    wordAndCountDStream.print
    //启动ssc.start()
    ssc.awaitTermination()
  }
}

4.2、有状态转化操作

即是对历史数据的状态进行维护。

4.2.1、UpdateStateByKey

UpdateStateByKey 原语用于记录历史记录，有时，我们需要在DStream 中跨批次维护状态(例如流计算中累加wordcount)。针对这种情况，updateStateByKey()为我们提供了对一个状态变量的访问，用于键值对形式的DStream。给定一个由(键，事件)对构成的DStream，并传递一个指定如何根据新的事件更新每个键对应状态的函数，它可以构建出一个新的DStream，其内部数据为(键，状态) 对。
updateStateByKey() 的结果会是一个新的DStream，其内部的RDD 序列是由每个时间区间对应的(键，状态)对组成的。
updateStateByKey 操作使得我们可以在用新信息进行更新时保持任意的状态。为使用这个功能，需要做下面两步：
1. 定义状态，状态可以是一个任意的数据类型。
2. 定义状态更新函数，用此函数阐明如何使用之前的状态和来自输入流的新值对状态进行更新。
使用updateStateByKey 需要对检查点目录进行配置，会使用检查点来保存状态。更新版的wordcount

package com.spark.stream
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Duration, Seconds, StreamingContext}
/**
  * SparkStreaming：将数据分为一个周期一个周期进行区分，每一个周期的数据分装为 DStream，再发给Executor 执行
  * 所以SparkStreaming并不是一个真正的流式处理，而是一个微批
  */
object StreamWordsCount {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("Save").setMaster("local[*]")
    /**
      * 指定采集周期，只会处理当前周期内的数据
      */
    val ctx = new StreamingContext(conf,Seconds(5))
    /**
      * 有状态编程时需要设置State状态的checkpoint，否则会有报错
      */
    ctx.checkpoint("/streamdata/")
    val data: ReceiverInputDStream[String] = ctx.socketTextStream("localhost",7777)
    /*val unit: DStream[(String, Iterable[Int])] = data.flatMap(_.split(" ")).map((_,1)).groupByKey()
    unit.map(data => (data._1,data._2.size)).print()*/
    /**
      * 无状态编程计算 wordcount
      */
    //    data.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).print()
    /**
      * 有状态编程，就是把之前的数据保存下来，而不是每次计算完就丢弃(无状态)
      */
    /**
      * The updateStateByKey operation allows you to maintain arbitrary state while continuously updating it with new information. To use this, you will have to do two steps.
      * 根据key对数据的状态进行更新
      * Define the state - The state can be an arbitrary data type.
      * Define the state update function - Specify with a function how to update the state using the previous state and the new values from an input stream.
      */
    val mapDStream: DStream[(String, Int)] = data.flatMap(_.split(" ")).map((_,1))
    val stateDStream: DStream[(String, Int)] = mapDStream.updateStateByKey {
      case (seq:Seq[Int], buffer:Option[Int]) => {//第一个值表示相同key的value值，第二个值表示缓冲区的值
        val sum = buffer.getOrElse(0) + seq.sum
        Option(sum)
      }
    }
    stateDStream.print()
    //启动采集器
    ctx.start()
    //Driver等待采集器的执行
    ctx.awaitTermination()
  }
}

4.2.2、WindowOperations

Window Operations 可以设置窗口的大小和滑动窗口的间隔来动态的获取当前Steaming 的允许状态。所有基于窗口的操作都需要两个参数，分别为窗口时长以及滑动步长。

• 窗口时长：计算内容的时间范围；
• 滑动步长：隔多久触发一次计算。

注意：这两者都必须为采集周期大小的整数倍。

object WorldCount {
def main(args: Array[String]) { 
  val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount") 
  val ssc = new StreamingContext(conf, Seconds(3)) //指定采集周期
    ssc.checkpoint("./ck")
  // Create a DStream that will connect to hostname:port, like localhost:9999 
    val lines = ssc.socketTextStream("linux1", 9999)
  // Split each line into words
  val words = lines.flatMap(_.split(" "))
  // Count each word in each batch
  val pairs = words.map(word => (word, 1))
  val wordCounts = pairs.reduceByKeyAndWindow(
    (a:Int,b:Int) => (a + b),Seconds(12), Seconds(6)
  )
  // Print the first ten elements of each RDD generated in this DStream to the console
  wordCounts.print()
  ssc.start()   
  ssc.awaitTermination() // Wait for the computation to terminate
 }
}

关于Window 的操作还有如下方法：
（1） window(windowLength, slideInterval): 基于对源DStream 窗化的批次进行计算返回一个新的Dstream；
（2） countByWindow(windowLength, slideInterval): 返回一个滑动窗口计数流中的元素个数；
（3） reduceByWindow(func, windowLength, slideInterval): 通过使用自定义函数整合滑动区间流元素来创建一个新的单元素流；
（4） reduceByKeyAndWindow(func, windowLength, slideInterval, [numTasks]): 当在一个(K,V) 对的DStream 上调用此函数，会返回一个新(K,V)对的DStream，此处通过对滑动窗口中批次数据使用reduce 函数来整合每个key 的value 值。
（5） reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, [numTasks]): 这个函数是上述函数的变化版本，每个窗口的reduce 值都是通过用前一个窗的reduce 值来递增计算。通过reduce 进入到滑动窗口数据并”反向 reduce”离开窗口的旧数据来实现这个操作。一个例子是随着窗口滑动对keys 的“加”“减”计数。通过前边介绍可以想到，这个函数只适用于“可逆的reduce 函数”，也就是这些 reduce 函数有相应的”反 reduce”函数(以参数invFunc 形式传入)。如前述函数，reduce 任务的数量通过可选参数来配置。

五、DStream 输出

输出操作指定了对流数据经转化操作得到的数据所要执行的操作(例如把结果推入外部数据库或输出到屏幕上)。与RDD 中的惰性求值类似，如果一个DStream 及其派生出的DStream 都没有被执行输出操作，那么这些DStream 就都不会被求值。如果StreamingContext 中没有设定输出操作，整个context 就都不会启动。
输出操作如下：

• print()：在运行流程序的驱动结点上打印DStream 中每一批次数据的最开始10 个元素。这用于开发和调试。在Python API 中，同样的操作叫print()。
• saveAsTextFiles(prefix, [suffix])：以text 文件形式存储这个DStream 的内容。每一批次的存储文件名基于参数中的prefix 和suffix。”prefix-Time_IN_MS[.suffix]”。
• saveAsObjectFiles(prefix, [suffix])：以Java 对象序列化的方式将Stream 中的数据保存为SequenceFiles . 每一批次的存储文件名基于参数中的为”prefix-TIME_IN_MS[.suffix]”. Python 中目前不可用。
• saveAsHadoopFiles(prefix, [suffix])：将Stream 中的数据保存为Hadoop files. 每一批次的存储文件名基于参数中的为”prefix-TIME_IN_MS[.suffix]”。Python API 中目前不可用。
• foreachRDD(func)：这是最通用的输出操作，即将函数func 用于产生于stream 的每一个RDD。其中参数传入的函数func 应该实现将每一个RDD 中数据推送到外部系统，如将RDD 存入文件或者通过网络将其写入数据库。

通用的输出操作foreachRDD()，它用来对DStream 中的RDD 运行任意计算。这和transform() 有些类似，都可以让我们访问任意RDD。在foreachRDD()中，可以重用我们在Spark 中实现的所有行动操作。比如，常见的用例之一是把数据写到诸如MySQL 的外部数据库中。
注意：
1) 连接不能写在driver 层面（序列化）
2) 如果写在foreach 则每个RDD 中的每一条数据都创建，得不偿失；
3) 增加foreachPartition，在分区创建（获取）。

六、优雅关闭

流式任务需要7*24 小时执行，但是有时涉及到升级代码需要主动停止程序，但是分布式程序，没办法做到一个个进程去杀死，所有配置优雅的关闭就显得至关重要了。
使用外部文件系统来控制内部程序关闭。

import java.net.URI
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext, StreamingContextState}
object SparkTest {
  def createSSC(): StreamingContext = {
    val update: (Seq[Int], Option[Int]) => Some[Int] = (values: Seq[Int], status: Option[Int]) => {
      //当前批次内容的计算
      val sum: Int = values.sum
      //取出状态信息中上一次状态
      val lastStatu: Int = status.getOrElse(0)
      Some(sum + lastStatu)
    }
    val sparkConf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("SparkTest")
    //设置优雅的关闭sparkConf.set("spark.streaming.stopGracefullyOnShutdown", "true")
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    ssc.checkpoint("./ck")
    val line: ReceiverInputDStream[String] = ssc.socketTextStream("linux1", 9999)
    val word: DStream[String] = line.flatMap(_.split(" "))
    val wordAndOne: DStream[(String, Int)] = word.map((_, 1))
    val wordAndCount: DStream[(String, Int)] = wordAndOne.updateStateByKey(update)
    wordAndCount.print()
    ssc
  }
  def main(args: Array[String]): Unit = {
    val ssc: StreamingContext = StreamingContext.getActiveOrCreate("./ck", () => createSSC())
    new Thread(new MonitorStop(ssc)).start()
    ssc.start()
    ssc.awaitTermination()
  }
}
class MonitorStop(ssc: StreamingContext) extends Runnable {
  override def run(): Unit = {
    val fs: FileSystem = FileSystem.get(new URI("hdfs://linux1:9000"), null, "atguigu")
    while (true) {
      try
        Thread.sleep(5000) catch {
        case e: InterruptedException => e.printStackTrace()
      }
      val state: StreamingContextState = ssc.getState
      val bool: Boolean = fs.exists(new Path("hdfs://linux1:9000/stopSpark"))
      if (bool) {
        if (state == StreamingContextState.ACTIVE) {
          ssc.stop(stopSparkContext = true, stopGracefully = true)
          System.exit(0)
        }
      }
    }
  }
}