SparkStreaming使用mapWithState时，设置timeout()无法生效问题，解决方案和原因过程分析！

前言

当我在测试SparkStreaming的状态操作mapWithState算子时，当我们设置timeout(3s)的时候，3s过后数据还是不会过期，不对此key进行操作，等到30s左右才会清除过期的数据。

百度了很久，关于timeout的资料很少，更没有解决这个问题的文章，所以说，百度也不是万能的，有时候还是需要靠自己。

所以我就在周末研究了一下，然后将结果整理了出来，希望能帮助大家更全面的理解Spark状态计算。

mapWithState

按理说Spark Streaming实时处理，数据就像流水，每个批次之间的数据都是独立的，处理完就处理完了，不留下任何状态。但是免不了一些有状态的操作，例如统计从流启动到现在，某个单词出现了多少次，所以状态操作就出现了。

状态操作分为updateStateByKey和mapWithState，两者有着很大的区别。简单的来说，前者每次输出的都是全量状态，后者输出的是增量状态。

过期原理

过期这一块估计很多人开始都理解错了，我刚开始理解就是数据从出现，经过多少秒之后就会过期。其实不是，这里的过期指的是空闲时间。

注释大概是这个意思：timeout()传入一个时间间隔参数，如果一个key在大于此间隔没有此key的数据流入，则被认为是空闲的，就会单独调用一次mapWithState中的func来清除这些空闲数据状态。

先写结论

使用了timeout()之后，需要使用以下代码来在间隔内清除失效key。

stream.checkpoint(Seconds(6))

checkpoint的时候，会开启全面扫描，才会对state中的失效key进行清理。

测试

   val conf = new SparkConf().setMaster("local[2]").setAppName("state")
    val ssc = new StreamingContext(conf, Seconds(3))
    ssc.checkpoint("./tmp")
    val streams: DStream[(String, Int)] = ssc.socketTextStream("localhost", 9999)
      .map(x => (x, 1))

    val result = streams.mapWithState(StateSpec.function((k: String, v: Option[Int], state: State[Int]) => {
        val count = state.getOption().getOrElse(0)
        println(k)
        println(v)
        var sum = 0
        if (!state.isTimingOut()) {
          sum = count + v.get
          state.update(sum)
        } else {
          println("timeout")
        }
        Option(sum)
      })
      .timeout(Seconds(3))
    )
    // 这行代码是触发清除机制的关键
    // result.checkpoint(Seconds(6))
    result.print()
    ssc.start()
    ssc.awaitTermination()

使用上面的代码进行测试，设置过期时间为3s。但是3s过后发现key并没有过期，也不会被清除，大概30S之后被清除。

在9999端口输入一个tom后，不再进行任何操作。测试结果如下：

tom
Some(1)
-------------------------------------------
Time: 1618228587000 ms
-------------------------------------------
Some(1)


tom
None
timeout
-------------------------------------------
Time: 1618228614000 ms
-------------------------------------------
Some(0)

从测试结果可以看出，从输入到清除大概是27s。

我们现在将注释的代码放开，每6s进行checkpoint一次，输入tom：

tom
Some(1)
-------------------------------------------
Time: 1618228497000 ms
-------------------------------------------
Some(1)

tom
None
timeout
-------------------------------------------
Time: 1618228506000 ms
-------------------------------------------
Some(0)

从生成到清除用了9秒，正好是过期时间 + 下一个窗口时间，触发了checkpoint。