标签：方法 参数 利用 多个 dep 技术 ons key 功能
在Flink架构体系中，有状态计算可以说是Flink非常重要的特征之一。有状态计算是指在程序计算过程中，在Flink程序内部，存储计算产生的中间结果，并提供给Functions 或 孙子计算结果使用。如图所示：
状态数据可以维系在本地存储中，这里的存储可以是 Flink 的堆内存或者堆外内存，也可以借助第三方的存储介质，例如：Flink中已经实现的RocksDB，当然用户也可以自己实现相应的缓存系统去存储状态信息，以完成更加复杂的计算逻辑。和状态计算不同的是，无状态计算不会存储计算过程中产生的结果，也不会将结果用于下一步计算过程中，程序只会在当前的计算流程中实行计算，计算完成就输出结果，然后下一条数据接入，然后处理。
无状态计算实现的复杂度相对较低，实现起来比较容易，但是无法完成提到的比较复杂的业务场景，例如：
在 Flink 中根据数据集是否根据 Key 进行分区，将状态分为 Keyed State 和 Operator State(Non-Keyed State) 两种类型。
Keyed State
表示和key相关的一种state ,只能用于 KeyedStream 类型数据集对应的Functions和Operators之上。Keyed State 是 Operator State 的特例，区别在于 Keyed State 事先按照 key 对数据集进行了分区，每个 Key State 仅对应一个 Operator 和 Key 的组合。 Keyed State 可以通过 Key Group 进行管理，主要用于当算子并行度发生变化时，自动重新分布 Keyed State 数据。
Operator State
与 Keyed State 不同的是，Operator State 只和并行的算子实例绑定，和数据元素中的 Key 无关，每个算子实例中持有所有数据元素中的一部分状态数据。 Operator State 支持当算子实例并行度发生变化时自动重新分配状态数据。
同时在Flink中 Keyed State 和 Operator State 均具有两种形式，其中一种为托管状态（Managered State）形式，由Flink Runtime 中控制和管理状态数据，并将状态数据转换称为内存Hash tables 或 Recks DB 的对象存储，然后将这些状态数据通过内部接口持久化到 Checkpoints 中，任务异常时可以通过这些状态数据恢复任务。另外一种是原生状态（Row State）形式，由算子自己管理数据结构，当触发 Checkpoints 过程中，Flink并不知道状态数据内部的数据结构，只是将数据转换成 bytes 数据存储在 Checkpoints 中，当从 Checkpoints 恢复任务时，算子自己在反序列化出状态的数据结构。
Notes: Flink中推荐用户使用 Managered State 管理状态数据，主要原因是：Manager State 能够更好的支持状态数据的重平衡以及更加完善的内存管理。
Flink 有以下Managered Keyed State 类型可以使用，每种状态都有相应的的使用场景，用户可以根据实际需求选择使用。
ValueState[T]: 与 Key 对应单个值的状态，例如统计 user_id 对应的交易次数，每次用户交易都会在 count 状态值上进行更新。 ValueState 对应的更新方法是 update(T) , 取值是 T value() ;ListState[T]: 与 Key 对应元素列表的状态，状态中存放元素的 List 列表。例如定义 ListValue存储用户经常访问的 IP 地址。在 ListState 中添加元素使用 add(T) , addAll(List[T]) 两个方法。获取元素使用 Iterable<T> get() 方法，更新元素使用 update(List[T])方法;ReducingState[T]: 定义与 Key 相关的数据元素单个聚合值的状态，用户存储经过指定 ReduceFunction 计算之后的指标，因此，ReduceState 需要指定ReduceFunction 完成状态数据的聚合。ReducingState 添加元素使用 add(T)方法，获取元素使用 T get() ;AggregeateState[IN,OUT]: 定义 与key相关的数据元素单个聚合值的状态，用于维护数据经过指定 AggregateFunction 计算之后的指标。和ReducingState相比，AggregeateState 的输入输出类型不一定相同，但ReducingState 输入/出 类型必须保持一致。和ListState相似，AggregatingState 需要指定AggregateFunction完成状态数据的聚合操作。AggregatringState添加元素使用 add(IN) 方法， 获取元素使用 OUT get() 方法;MapState<UK, UV>：这会保留一个映射列表。您可以将键值对放入状态并检索Iterable所有当前存储的映射。使用put(UK, UV)或 添加映射putAll(Map[UK,UV])(Map<UK, UV>)。可以使用来检索与用户键关联的值get(UK)。对于映射，键和值可迭代视图可以使用被检索entries()，keys()并values()分别。Stateful Function定义
示例：
在RichFlatMapFunction 中定义 ValueState,已完成最小值的获取：
inputStream.keyBy(_._1).flatMap(
// (String,Long,Int) 输入类型
// (String,Long,Long) 输出类型
new RichFlatMapFunction[(Int,Long) , (Int,Long,Long)] {
private var leastValueState:ValueState[Long] = _
// 定义状态名称
private var leastValueStateDesc:ValueStateDescriptor[Long] = _
override def open(parameters: Configuration): Unit = {
// 指定状态类型
leastValueStateDesc = new ValueStateDescriptor[Long]("leastValueState" , classOf[Long])
// 通过 getRuntimeContext.getState 拿到状态
leastValueState = getRuntimeContext.getState(leastValueStateDesc)
}
override def flatMap(value: (Int, Long), out: Collector[(Int, Long, Long)]): Unit = {
// 通过 value 拿到最小值
val leastValue: Long = leastValueState.value()
// 如果前一个指标大于最小值，则直接输出数据元素和最小值
if ( leastValue != 0L && value._2 > leastValue){
out.collect((value._1 , value._2 , leastValue))
}else{
// 如果当前指标小于最小值，则更新状态中的最小值
leastValueState.update(value._2)
// 将当前数据中的指标作为最小值输出
out.collect(value._1 , value._2 , value._2)
}
}
}).print()State生命周期
对于任何类型 Keyed State 都可以设定状态生命周期（TTL），以确保能够在规定时间内即时清理状态数据。状态生命周期功能可通过 StateTtlConfig 配置然后将 StateTtlConfig 配置传入StateDescriptor 中的 enableTimeToLive 方法中即可。Keyed State 配置实例如下所示：
val config: StateTtlConfig = StateTtlConfig
// 指定TTL时长为 5s
.newBuilder(Time.seconds(5))
// 指定TTL 刷新只对创建和写入操作有效
.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
// 指定状态可见性不返回过期数据
.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
.build()
leastValueStateDesc.enableTimeToLive(config)在StateTtlConfig中除了通过 newBuilder() 方法中设定过期时间的参数是必须的之外，其他的参数都是可选的或使用默认值。其中 setUpdateType方法中传入的类型有两种：
另外，可以通过 setStateVisibility 方法设定状态的可见性，根据过期数据是否被清理来确定是否返回状态数据:
Scala DataStream API中使用状态
直接上代码片段：
inputStream.keyBy(_._1)
// 指定输入参数类型和状态参数类型
.mapWithState((in:(Int,Long) , count : Option[Int]) =>
// 判断count 类型是否非空
count match {
// 输出 key , count 并在原来 count 数据上累加
case Some(c) => ((in._1 , c) , Some(c + in._2))
// 如果状态为空，则将指标填入
case None => ((in._1 , 0) , Some(in._2))
}
)Operator State 是一种 non-keyed-state ,与并行的操作算子实例相关联，例如在 Kafka Connector 中，每个 Kafka 消费端算子实例都对应到 Kafka 的一个分区中，维护Topic分区和 Offsets 偏移量作为算子的 Operator State. 在Flink中可以实现 CheckpointedFunction 或者 ListCheckpoint<T extends Serializable>两个接口来定义操作 Managered Operator State 的函数。
通过 CheckpointedFunction 接口操作Operator State
CheckpointedFunction 接口定义如图：
@PublicEvolving
@SuppressWarnings("deprecation")
public interface CheckpointedFunction {
/**
* This method is called when a snapshot for a checkpoint is requested. This acts as a hook to the function to
* ensure that all state is exposed by means previously offered through {@link FunctionInitializationContext} when
* the Function was initialized, or offered now by {@link FunctionSnapshotContext} itself.
*
* @param context the context for drawing a snapshot of the operator
* @throws Exception
*/
void snapshotState(FunctionSnapshotContext context) throws Exception;
/**
* This method is called when the parallel function instance is created during distributed
* execution. Functions typically set up their state storing data structures in this method.
*
* @param context the context for initializing the operator
* @throws Exception
*/
void initializeState(FunctionInitializationContext context) throws Exception;
}在每个独立的算子中，Managered Operator State 都是以 List 形式存储的，算子和算子之间的状态数据相互独立，List存储比较适合于状态数据的重新分布，Flink目前支持Manager Operator State 两种重要分布策略，分别是 Event-split Redistribution 和 Union Redistribution。
标签：方法 参数 利用 多个 dep 技术 ons key 功能
原文地址：https://www.cnblogs.com/sun-iot/p/12089562.html
