状态相关

状态概述

在 Flink 中，算子任务可以分为无状态和有状态两种情况。

• 无状态的算子任务只需要观察每个独立事件，根据当前输入的数据直接转换输出结果。

  

• 有状态

  • ==首先搞明白什么是状态？==：流式计算的数据往往是转瞬即逝， 真实业务场景不可能说所有的数据都是进来之后就走掉，没有任何东西留下来，那么留下来的其实就是状态(state)。在Flink中，也可以这样理解State：task/operator在某时刻的一个中间结果。
  • 在Apache Flink架构体系中，有状态计算可以说是Flink非常重要的特征之一。有状态计算是指在程序计算过程中，在Flink程序内部，存储计算产生的中间结果，并提供给Functions 或 算子计算使用。

  

  有状态算子处理主要有两个步骤：

  1.获取当前状态

  2.根据具体的业务逻辑进行计算，使用计算后的结果更新当前状态。

有状态计算案例分析：

为什么Flink知道之前已经处理过一次 'a'和'b'?

原因就是 state发挥作用了，它 存储了之前统计后的结果数据，所以Flink 程序知道'a'和'b'词频。

扩展：Apache Flink作为一个计算框架，提供了有状态的计算，封装了一些底层的实现，比如状态的高效存储、Checkpoint和Savepoint持久化备份机制、计算资源扩缩容等问题。因为Flink接管了这些问题，开发者只需调用Flink API，这样可以更加专注于业务逻辑。

思考：为什么流式计算中需要State状态呢？

1. 与批计算相比，==State是流计算特有的==，批计算没有failover机制，要么成功，要么重新计算。
2. 流计算在大多数场景下是==增量计算，数据逐条处理（大多数场景），每次计算是在上一次计算结果之上进行处理的==，这样的机制势必要将上一次的计算结果进行存储（生产模式要持久化；
3. 另外由于机器、网络、脏数据等原因导致的程序错误，在==重启job时候需要从成功的检查点(checkpoint)进行state的恢复==。增量计算，Failover这些机制都需要state的支撑。

代码参考：

xxxxxxxxxx
package com.itbaizhan.flink.scala.state
object WordCountState {
  def main(args: Array[String]): Unit = {
    //导入隐式转换
    import org.apache.flink.streaming.api.scala._
    val env: StreamExecutionEnvironment =
      StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)//设置并行度为1
    env.socketTextStream("node3", 8888)
      .filter(_.trim.length>0)//过滤掉空行
      .flatMap(_.split("\\s+"))//按空格拆分
      .map((_, 1)) //每个单词计数为1
      .keyBy(_._1) .sum(1)//分组统计
      .print()
    //触发执行-execute
    env.execute("WordCountState")
  }
}