8.1 Flink中的状态

image.png

  • 由一个任务维护、并且用来计算某个结果的所有数据，都属于这个任务的状态。
  • 可以认为状态就是一个本地变量，可以被任务的业务逻辑访问。
  • Flink会进行状态管理，包括状态一致性、故障处理、扩缩容处理、以及高效存储和访问，以便开发人员可以专注于应用程序的逻辑。
  • 在Flink中，状态始终与特定算子相关联。为了使运行时的Flink了解算子的状态，算子需要预先注册其状态。总的来说，Flink中有两种类型的状态：
    • 算子状态（Operator State）：算子状态的作用范围限定为算子任务。
    • 键控状态（KeyedState）：根据输入数据流中定义的键（key）来维护和访问。

8.2 算子状态（Operator State）

image.png

  • 算子状态的作用范围限定为算子任务，由同一个任务所处理的所有数据都可以访问到相同的状态；不同任务之间，即便是同一个算子的并行任务，也不能访问到对方的状态。

    8.2.1 算子状态数据结构

  • 列表状态（List state）：将状态表示为一组数据的列表。

  • 联合列表状态（Union list state）：与常规列表状态的区别在于，在发送故障时，或者从保存点（savepoint）启动应用程序时如何恢复。
  • 广播状态（Broadcast state）：如果一个算子有多项任务，这些任务的状态又需要共享，那么就可使用广播状态。

    算子状态为什么没有单值类型？这是因为在扩缩容时，一个算子的状态可能需要被拆分，也可能需要合并，用List更容易处理。

8.2.2 算子状态的使用

  1. public static void main(String[] args) throws Exception {
  2.     StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  3.     // 设置时间语义
  4.     env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
  6.     // 获取Socket文本流
  7.     DataStreamSource<String> inputStream = env.socketTextStream("localhost", 7777);
  9.     // 转换成SensorReading
  10.     SingleOutputStreamOperator<SensorReading> dataStream = inputStream.map(line -> {
  11.         String[] fields = line.split(",");
  12.         return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
  13.     });
  15.     // 定义一个有状态的map操作，统计当前分区数据个数
  16.     SingleOutputStreamOperator<Integer> resultStream = dataStream.map(new MyCountMapper());
  18.     resultStream.print();
  20.     env.execute();
  21. }
  23. public static class MyCountMapper implements MapFunction<SensorReading, Integer>, ListCheckpointed<Integer> {
  25.     // 状态字段
  26.     private Integer count = 0;
  28.     @Override
  29.     public Integer map(SensorReading value) throws Exception {
  30.         this.count++;
  31.         return this.count;
  32.     }
  34.     // 保存状态时调用的方法
  35.     @Override
  36.     public List<Integer> snapshotState(long checkpointId, long timestamp) throws Exception {
  37.         return Collections.singletonList(this.count);
  38.     }
  40.     // 恢复状态时调用的方法
  41.     @Override
  42.     public void restoreState(List<Integer> state) throws Exception {
  43.         // 恢复时，可能别的分区的state也合并到了一起，因此state中可能有多个值，所以需要遍历相加。
  44.         for (Integer num : state) {
  45.             count += num;
  46.         }
  47.     }
  48. }

8.3 键控状态（Keyed State）

image.png

  • 键控状态是根据输入数据流中定义的键（key）来维护和访问的，其范围粒度比算子状态更小。
  • Flink为每个key维护了一个状态实例，并将具有相同key的所有数据，都分区到同一个算子任务重，这个任务会维护和处理这个key对应的状态。

  • 当任务处理一条数据时，它会自动将状态的访问范围限定为当前数据的key。

    8.3.1 键控状态数据结构

  • 值状态（Value state）：将状态表示为单个的值。

  • 列表状态（List state）：将状态表示为一组数据的列表。
  • 映射状态（Map state）：将状态表示为一组key-Value对。
  • 聚合状态（Reducing state & Aggregation State）：将状态表示为一个用于聚合操作的列表。

    8.3.2 键控状态的使用

    image.png

8.4 状态后端（State Backend）

  • 每传入一条数据，有状态的算子任务都会读取和更新状态。
  • 由于有效的任务访问对于处理数据的低延迟至关重要，因此每个并行任务都会在本地（内存）维护其状态，以确保快速的状态访问。
  • 状态的存储、访问以及维护，由一个可插入的组件决定，这个组件就叫做状态后端（state backend）。

  • 状态后端主要负责两件事情：本地的状态管理，以及将检查点（checkpoint）状态写入远程存储。

    8.4.1 选择一个状态后端

  • MemoryStateBackend

    • 内存级的状态后端，会将键控状态作为内存中的对象进行管理，将它们存储在TaskManager的JVM堆上，而将checkpoint存储在JobManager的内存中。
    • 特点：快速、低延迟，但不稳定。
  • FsStateBackend
    • 将checkpoint存到远程的持久化文件系统（FileSystem）上（一般是HDFS），而对于本地状态，跟MemoryStateBackend一样，也会存在TaskManager的JVM堆上。
    • 特点：同时拥有内存级的本地访问速度，和更好的容错保证。
  • RocksDBStateBackend
    • 将所有状态序列化后，存入本地的RocksDB中存储。
    • 特点：适用于状态大到超过内存上限的场景。

      状态后端可以在配置文件中做全局设置，也可以在每个程序的执行环境中单独配置。