Spark Streaming状态管理函数（三）——MapWithState的使用（scala版）

文章目录

关于mapWithState

  须要本身写一个匿名函数func来实现本身想要的功能。若是有初始化的值得须要，可使用initialState(RDD)来初始化key的值。 另外，还能够指定timeout函数，该函数的做用是，若是一个key超过timeout设定的时间没有更新值，那么这个key将会失效。这个控制须要在func中实现，必须使用state.isTimingOut()来判断失效的key值。若是在失效时间以后，这个key又有新的值了，则会从新计算。若是没有使用isTimingOut，则会报错。

注意事项

  下面程序是使用idea编写的，使用的是scala语言，在程序中master(“local[2]”)设置为本地模式（[]中的数指定的是线程数，不能少于2，不然看不到结果。主要是由于spark须要启动一个线程receiver来循环接收数据，一个Executor来接收数据，若是少于2线程不够将不能打印出结果。），在window上运行的。使用的spark版本是2.3.0，在2.x之后的版本，基本采用SparkSession来进行操做。同时，想要运行程序你的服务器上还必需要安装netcat这个软件，使用yum install nc进行安装（注意安全配置好yum源，DNS才能下载安装），使用命令nc -lk 6666开启服务发送数据。最后在运行程序前还须要导入spark、scala相应的依赖包。

示例代码

package spark2x
  
  import org.apache.spark.SparkContext
  import org.apache.spark.sql.SparkSession
  import org.apache.spark.streaming.{Seconds, StreamingContext}
  import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
  /**
    * 类名  UpdateStateByKeyDemo
    * 做者   彭三青
    * 建立时间  2018-12-01 9:35
    * 版本  1.0
    * 描述: $
    */
  
  object UpdateStateByKeyDemo {
    def main(args: Array[String]): Unit = {
      /** 第一步：配置SparkConf：
        * 1，至少2条线程：由于Spark Streaming应用程序在运行的时候，至少有一条
        * 线程Receiver用于不断的循环接收数据，还有一条线程是Executor用于处理接受的数据（少于两条
        * 就没有线程用于处理数据，窗口不会显示数据。而且随着时间的推移，内存和磁盘因为负担太重而崩溃）；
        * 2，对于集群而言，根据已有经验，大概5个左右的Core是性能最佳（通常分配为奇数个Core）
        */
       val spark = SparkSession.builder()
        .master("local[2]")
        .appName("UpdateStateByKeyDemo")
        .getOrCreate()
      val conf: SparkContext = spark.sparkContext
      /**
        * 第二步：建立SparkStreamingContext：
        * 1，这个是SparkStreaming应用程序全部功能的起始点和程序调度的核心
        * SparkStreamingContext的构建能够基于SparkConf参数，也可基于持久化的SparkStreamingContext的内容
        * 来恢复过来（典型的场景是Driver崩溃后从新启动，因为Spark Streaming具备连续7*24小时不间断运行的特征，
        * 全部须要在Driver从新启动后继续上衣系的状态，此时的状态恢复须要基于曾经的Checkpoint）；
        * 2，在一个Spark Streaming应用程序中能够建立若干个SparkStreamingContext对象，使用下一个SparkStreamingContext
        * 以前须要把前面正在运行的SparkStreamingContext对象关闭掉，由此，咱们得到一个重大的启发SparkStreaming框架也只是
        * Spark Core上的一个应用程序而已，只不过Spark Streaming框架箱运行的话须要Spark工程师写业务逻辑处理代码；
        */
      val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
      //报错解决办法作checkpoint,开启checkpoint机制，把checkpoint中的数据放在这里设置的目录中，生产环境下通常放在HDFS中
      ssc.checkpoint("hdfs://SC01:8020/user/tmp/cp-20181201")
  
      /**
        * 第三步：建立Spark Streaming输入数据来源input Stream：
        * 1，数据输入来源能够基于File、HDFS、Flume、Kafka、Socket等
        * 2, 在这里咱们指定数据来源于网络Socket端口，Spark Streaming链接上该端口并在运行的时候一直监听该端口
        * 的数据（固然该端口服务首先必须存在）,而且在后续会根据业务须要不断的有数据产生(固然对于Spark Streaming
        * 应用程序的运行而言，有无数据其处理流程都是同样的)；
        * 3,若是常常在每间隔5秒钟没有数据的话不断的启动空的Job实际上是会形成调度资源的浪费，由于并无数据须要发生计算，因此
        * 实例的企业级生成环境的代码在具体提交Job前会判断是否有数据，若是没有的话就再也不提交Job；
        */
      val line: ReceiverInputDStream[String] = ssc.socketTextStream("SC01", 6666)
  
      /**
        * 第四步：接下来就像对于RDD编程同样基于DStream进行编程！！！缘由是DStream是RDD产生的模板（或者说类），在Spark Streaming具体
        * 发生计算前，其实质是把每一个Batch的DStream的操做翻译成为对RDD的操做！！！
        * 对初始的DStream进行Transformation级别的处理，例如map、filter等高阶函数等的编程，来进行具体的数据计算
        * 进行单词拆分
        */
      val words: DStream[String] = line.flatMap(_.split(" "))
  
      /**
        * 对初始的DStream进行Transformation级别的处理，例如map、filter等高阶函数等的编程，来进行具体的数据计算
        * 单词分组计数实，word => (word, 1) Word ->(word, 1) day -> day(day, 1)
        */
      val pairs: DStream[(String, Int)] = words.map((_, 1))
  
      /**
        * 经过updateStateByKey来以Batch Interval为单位来对历史状态进行更新，
        * 这是功能上的一个很是大的改进，不然的话须要完成一样的目的，就可能须要把数据保存在Redis、
        * Tagyon或者HDFS或者HBase或者数据库中来不断的完成一样一个key的State更新，若是你对性能有极为苛刻的要求，
        * 且数据量特别大的话，能够考虑把数据放在分布式的Redis或者Tachyon内存文件系统中；
        * Spark2.X后mapWithState应该很是稳定了。
        */
      val wordCount: DStream[(String, Int)] = pairs.updateStateByKey((values: Seq[Int], state: Option[Int]) => {
        var newValue = state.getOrElse(0)
        for (value <- values) {
          newValue += value
        }
        Option(newValue)
      })
  
      /**
        * 此处的print并不会直接出发Job的执行，由于如今的一切都是在Spark Streaming框架的控制之下的，对于Spark Streaming
        * 是否触发真正的Job运行是基于设置的Duration时间间隔的
        * 须要注意的是Spark Streaming应用程序要想执行具体的Job，对Dtream就必须有output Stream操做，
        * output Stream有不少类型的函数触发，类print、saveAsTextFile、saveAsHadoopFiles等，最为重要的一个
        * 方法是foraeachRDD,由于Spark Streaming处理的结果通常都会放在Redis、DB、DashBoard等上面，foreachRDD
        * 主要就是用用来完成这些功能的，并且能够随意的自定义具体数据到底放在哪里！！！
        */
      wordCount.print()
  
      /**
        * Spark Streaming执行引擎也就是Driver开始运行，Driver启动的时候是位于一条新的线程中的，固然其内部有消息循环体，用于
        * 接受应用程序自己或者Executor中的消息；
        */
      // 开始提交任务
      ssc.start()
      // 线程等待，等待处理下一批次任务
      ssc.awaitTermination()
    }
  
      /** Iterator[(K, Seq[V], Option[S])]) => Iterator[(K, S)]
        * 在调用updateStateByKey中，须要传入一个用于计算历史批次和当前批次数据的函数
        * 该函数中有几个类型：String, Seq[Int], Option[Int])]
        * String表明元组中每个单词，也就是key
        * Seq[Int]表明当前批次相同key对应的value，好比Seq(1,1,1,1)
        * Option[Int]表明上一批次中相同key对应的累加的结果，有可能有值，有可能没有值。
        * 此时，获取历史批次的数据时，最好用getOrElse方法
        */
    val func = (it: Iterator[(String, Seq[Int], Option[Int])]) => {
      it.map(tup => {
        (tup._1, tup._2.sum + tup._3.getOrElse(0))
      })
    }
  }

运行

  服务器运行nc

  idea端运行编写好的程序
  服务器发送数据

  控制台显示结果

结论

  mapWithState它会按照时间线在每个批次间隔返回以前的发生改变的或者新的key的状态，不发生变化的不返回。同时mapWithState能够不用设置checkpoint，返回的数据量少，性能和效率都比mapWithState好。