Flink流计算状态与容错【键控状态、操作状态、托管状态】

╰+哭是因爲堅強的太久メ 2023-02-20 08:24 22阅读 0赞

## 一.概述 ##

有状态的函数和运算符在处理单个元素/事件的过程中存储数据，从而使状态成为任何类型的更精细操作的关键构建块。

包括：

*  当应用程序搜索某些事件模式时，状态将存储到目前为止遇到的事件序列。
 *  在每分钟/小时/天汇总事件时，状态将保留待处理的汇总。
 *  在数据点流上训练机器学习模型时，状态保持模型参数的当前版本。
 *  当需要管理历史数据时，该状态允许有效访问过去发生的事件。

Flink需要了解状态，以便使用检查点使状态容错并允许流应用程序的保存点。

状态还允许重新缩放Flink应用程序，这意味着Flink负责在并行实例之间重新分配状态。

Flink 的可查询状态功能允许在运行时从Flink外部访问状态。

使用状态时，阅读有关Flink的状态后端的信息也可能会很有用。Flink提供了不同的状态后端，用于指定状态的存储方式和位置。状态可以位于Java的堆或堆外。根据状态后端，Flink还可以管理应用程序的状态，这意味着Flink可以处理内存管理（如有必要，可以落到磁盘上）以允许应用程序保持很大的状态。可以配置状态后端，而无需更改应用程序逻辑。

## 二.工作状态 ##

Flink中有两种基本状态：Keyed State和Operator State。  
1.键控状态  
键控状态始终与键相关，并且只能在KeyedStream中使用函数和运算符。

可以将**键控状态**视为已分区或分片的操作状态，每个键仅具有一个状态分区。每个键状态在逻辑上都绑定到<parallel-operator-instance，key>的唯一组合，并且由于每个键都完全属于键操作符的一个并行实例，因此我们可以简单地将其视为<operator，key >。

Keyed State被进一步组织成所谓的Key Groups。密钥组是Flink可以重新分配密钥状态的原子单位；与定义的最大并行度完全相同的键组数。在执行期间，键控运算符的每个并行实例都使用一个或多个键组的键。  
2.操作状态  
使用运算符状态（或非键状态），每个运算符状态都绑定到一个并行运算符实例。在kafka连接器使用Flink运营状况的一个很好激励的例子。Kafka使用者的每个并行实例都维护一个主题分区和偏移量的映射作为其操作状态。

当更改并行性时，操作状态接口支持在并行操作实例之间重新分配状态。可以有不同的方案来进行此重新分配。

## 三.原始状态和托管状态 ##

键控状态和操作状态以两种形式存在：托管状态和原始状态。

托管状态用Flink运行时控制的数据结构表示，例如内部哈希表或RocksDB。例如“ ValueState”，“ ListState”等。Flink的运行时对状态进行编码，并将其写入检查点。

原始状态是操作保留其自己的数据结构的状态。被检查点时，它们仅将字节序列写入检查点。Flink对状态的数据结构一无所知，只看到原始字节。

所有数据流功能都可以使用托管状态，但是原始状态接口只能在实现运算符时使用。建议使用托管状态（而不是原始状态），因为有了托管状态，Flink可以在更改并行性时自动重新分配状态，并且还可以进行更好的内存管理。

注意：如果管理状态需要自定义序列化逻辑，请参阅相应的指南，以确保将来的兼容性。Flink的默认序列化器不需要特殊处理。

## 四.托管密钥状态 ##

管理密钥的状态接口提供不同类型状态的访问接口，这些状态都作用于当前输入数据的键下。换句话说，这些状态仅在KeyedStream上使用，可以使用stream.keyBy(…)得到KeyedStream。

所有支持的状态类型如下所示：

*  ValueState：保存一个可以更新和检索的值（每个值都对应到当前的输入数据的key，因此算子接收到的每个key都可能对应一个值）。这个值可以通过update(T)进行更新，通过T value()进行检索。
 *  ListState：保存一个元素的列表可以往这个列表中追加数据，并在当前的列表上进行检索。 add(T)或者addAll(List)进行添加元素，通过Iterable get()获得整个列表。update(List)覆盖当前的列表。
 *  ReducingState：保存一个单值，表示添加到状态的所有值的聚合。接口与ListState类似，但使用add(T)增加的元素，会使用提供的ReduceFunction进行聚合。
 *  AggregatingState<IN, OUT>：保留一个单值，表示添加到状态的所有值的聚合。和ReducingState相反的是，聚合类型可能与添加到状态的元素的类型不同。接口与ListState类似，但使用add(IN)添加的元素会用指定的AggregateFunction进行聚合。
 *  FoldingState<T, ACC>：与ReducingState相反，聚合类型可能与添加到状态的元素类型不同。接口与ListState类似，但使用add（T）添加的元素会用指定的FoldFunction折叠成聚合值。
 *  MapState<UK, UV>：维护了一个映射列表可以添加键值对到状态中，也可以从中获取。当前反映所有映射的迭代器使用put(UK，UV)或者putAll(Map<UK，UV>)添加映射，使用get(UK)获取映射。检索特定键使用entries()，keys()，values()分别检索映射，键和值的可迭代视图。

所有类型的状态还有一个clear()方法，清除当前键下一个状态数据，也就是当前输入元素的键。

注意 FoldingState和FoldingStateDescriptor从Flink 1.4开始就已经被启用，将会在未来被删除。作为替代请使用AggregatingState和AggregatingStateDescriptor。

请始终保持不变，将其存储在内存中，还可能在磁盘或其他位置。另外需要牢记的是从状态中获取的值由输入元素所代表的键。因此，在不同的键上调用同一个接口，可能得到不同的值。

必须创建一个StateDescriptor，才能得到对应的状态句柄。这保存了状态名称【可以创建多个状态，并且它们必须具有唯一的名称以便可以引用它们】状态所持有值的类型，并且可能包含用户指定的函数，例如ReduceFunction，根据不同的状态类型，可以创建ValueStateDescriptor，ListStateDescriptor， ReducingStateDescriptor，FoldingStateDescriptor或MapStateDescriptor。

状态通过RuntimeContext进行访问，因此只能在丰富的功能中使用。RichFunction中RuntimeContext提供如下方法：

*  ValueState getState(ValueStateDescriptor)
 *  ReducingState getReducingState(ReducingStateDescriptor)
 *  ListState getListState(ListStateDescriptor)
 *  AggregatingState<IN, OUT> getAggregatingState(AggregatingStateDescriptor<IN, ACC, OUT>)
 *  FoldingState<T, ACC> getFoldingState(FoldingStateDescriptor<T, ACC>)
 *  MapState<UK, UV> getMapState(MapStateDescriptor<UK, UV>)  
    下面是一个FlatMapFunction的例子，展示了如何将这些部分组合起来：

class CountWindowAverage extends RichFlatMapFunction[(Long, Long), (Long, Long)] { 
    
      private var sum: ValueState[(Long, Long)] = _
    
      override def flatMap(input: (Long, Long), out: Collector[(Long, Long)]): Unit = { 
    
        // access the state value
        val tmpCurrentSum = sum.value
    
        // If it hasn't been used before, it will be null
        val currentSum = if (tmpCurrentSum != null) { 
          tmpCurrentSum
        } else { 
          (0L, 0L)
        }
    
        // update the count
        val newSum = (currentSum._1 + 1, currentSum._2 + input._2)
    
        // update the state
        sum.update(newSum)
    
        // if the count reaches 2, emit the average and clear the state
        if (newSum._1 >= 2) { 
          out.collect((input._1, newSum._2 / newSum._1))
          sum.clear()
        }
      }
    
      override def open(parameters: Configuration): Unit = { 
        sum = getRuntimeContext.getState(
          new ValueStateDescriptor[(Long, Long)]("average", createTypeInformation[(Long, Long)])
        )
      }
    }
    
    
    object ExampleCountWindowAverage extends App { 
      val env = StreamExecutionEnvironment.getExecutionEnvironment
    
      env.fromCollection(List(
        (1L, 3L),
        (1L, 5L),
        (1L, 7L),
        (1L, 4L),
        (1L, 2L)
      )).keyBy(_._1)
        .flatMap(new CountWindowAverage())
        .print()
      // the printed output will be (1,4) and (1,5)
    
      env.execute("ExampleManagedState")
    }

把元组的第一个元素当作键（在示例中都key都是“ 1”）。该函数将出现的次数以及总和存储在“ ValueState”中。一旦出现次数达到2，则将逐步发送到下游，并清除状态重新开始。请注意，我们会为每个不同的key（元组中第一个元素）保存一个单独的值。