一.Flink中的状态
1.1 概述
在Flink中，算子任务可以分为有状态和无状态两种状态。
无状态的算子任务只需要观察每个独立事件，根据当前输入的数据直接转换输出结果。例如Map、Filter、FlatMap都是属于无状态算子。
而有状态的算子任务，就是除了当前数据外，还需要一些其他的数据来得到计算结果。这里的其他数据就是所谓的“状态”。例如聚合函数、窗口函数都属于有状态算子。
1.2 状态的分类
1.2.1 托管状态（Managed State）和原始状态（Raw State）
Flink的状态有两种，托管状态（Managed State）和原始状态（Raw State）。托管状态就是由Flink统一管理的，状态的存储访问、故障恢复和重组等一系列问题都由Flink实现，我们只要调接口就可以；而原始状态则是自定义的，所有的状态具体管理则需要自行实现。
一般使用托管状态即可。后面的所有内容也仅是基于托管状态的。
1.2.2 算子状态（Operator State）和按键分区状态（Keyed State）
在Flink中，一个算子任务会按照并行度分为多个并行子任务执行，而不同的子任务会占据不同的任务槽（task slot）。由于不同的slot在计算资源上是物理隔离的，所以Flink能管理的状态在并行任务间是无法共享的，每个状态只能针对当前子任务的实例有效。
而很多有状态的操作（比如聚合、窗口）都是要先做keyBy进行按键分区的。按键分区之后，任务所进行的所有计算都应该只针对当前key有效，所以状态也应该按照key彼此隔离。在这种情况下，状态的访问方式又会有所不同。
基于这样的想法，又可以将托管状态分为两类：算子状态和按键分区状态。经过KeyBy操作后的状态则被称为"按键分区状态（Keyed State）"，否则就是“算子状态（Operator State）”。
算子状态
算子状态的状态作用范围为当前算子任务实例-即每个task(分区)间状态不共享。
算子状态可以用在所有算子上，使用时与本地变量没什么区别，在使用时需实现checkpoint接口。假如使用新的Source架构，则需要继承SourceReaderBase抽象类。
按键分区状态
按键分区状态只有在KeyBy后才能使用，因为状态是根据输入流中定义的键(Key)来维护和访问的。每个Key分区间状态不共享。
二.按键分区状态（Keyed State）
按键分区状态（Keyed State）顾名思义，是任务按照键（key）来访问和维护的状态。它的特点非常鲜明，就是以key为作用范围进行隔离。
2.1 值状态（ValueState）
顾名思义，状态中只保存一个“值”（value）。ValueState<T>本身是一个接口，源码如下：
@PublicEvolving
public interface ValueState<T> extends State {
T value() throws IOException;
void update(T var1) throws IOException;
}这里的T是泛型，表示值状态数据类型。
对值的操作主要有以下：
// 获取当前状态值
T value()
// 更新/覆盖状态值
update(T value)在具体使用时，为了让运行时上下文清楚到底是哪个状态，我们还需要创建一个“状态描述器”（StateDescriptor）来提供状态的基本信息。例如源码中，ValueState的状态描述器构造方法如下：
public ValueStateDescriptor(String name, Class<T> typeClass) {
super(name, typeClass, null);
}
这里需要传入状态的名称和类型——这跟我们声明一个变量时做的事情完全一样。
案例：检测每种传感器的水位值，如果连续的两个水位值差值超过10，就输出报警。
public class KeyedValueStateDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
SingleOutputStreamOperator<WaterSensor> sensorDS = env
.socketTextStream("xxx.xxx.xxx.xxx", 1234)
.map(new MyMapFunctionImpl())
.assignTimestampsAndWatermarks(WatermarkStrategy
.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3)) // 设置最大等待时间为3s
.withTimestampAssigner((SerializableTimestampAssigner<WaterSensor>) (waterSensor, l) -> waterSensor.getTs() * 1000L)
);
// 对传感器做KeyBy
SingleOutputStreamOperator<String> process = sensorDS.keyBy(r -> r.getId())
.process(new KeyedProcessFunction<String, WaterSensor, String>() {
// 定义状态，用于每组当前的水位线
ValueState<Integer> lastVcState;
// 必须在open方法中，初始化状态
@Override
public void open(Configuration parameters) throws Exception {
// 初始化值状态，需传入值状态描述器(唯一的名称，值的类型)
lastVcState = getRuntimeContext().getState(new ValueStateDescriptor<>("lastVc", Types.INT));
}
@Override
public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
// 1.取出上一条水位线
int lastVc = lastVcState.value() == null ? value.getVc() : lastVcState.value();
// 2.判断是否超过10
if (Math.abs(value.getVc() - lastVc) > 10) {
out.collect("传感器：" + value.getId() + ",上一次水位线：" + lastVc + ",当前水位线:" + value.getVc() + ",触发报警(相差超过10)!!!");
}
// 更新当前状态
lastVcState.update(value.getVc());
}
});
process.print();
env.execute();
}输入：
[root@VM-55-24-centos ~]# nc -lk 1234
s1,1,1
s1,2,13
s1,5,9
s1,6,22
s2,9,10
s2,10,23
输出：
传感器：s1,上一次水位线：1,当前水位线:13,触发报警(相差超过10)!!!
传感器：s1,上一次水位线：9,当前水位线:22,触发报警(相差超过10)!!!
传感器：s2,上一次水位线：10,当前水位线:23,触发报警(相差超过10)!!!
如果不使用状态存储，则需要定义HashMap存储每个Key的水位线，没有状态高效。
2.2 列表状态（ListState）
将需要保存的数据，以列表（List）的形式组织起来。在ListState<T>接口中同样有一个类型参数T，表示列表中数据的类型。ListState也提供了一系列的方法来操作状态，使用方式与一般的List非常相似。
对 List 状态的操作主要有以下：
// 获取当前的列表状态，返回的是一个可迭代类型Iterable<T>
Iterable<T> get()
// 传入一个列表values，直接对状态进行覆盖
update(List<T> values)
// 向列表中添加多个元素，以列表values形式传入
addAll(List<T> values)
类似地，ListState的状态描述器就叫作ListStateDescriptor，用法跟ValueStateDescriptor完全一致。
案例:针对每种传感器输出最高的3个水位值。
public class KeyedListStateDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
SingleOutputStreamOperator<WaterSensor> sensorDS = env
.socketTextStream("xxx.xxx.xxx.xxx", 1234)
.map(new MyMapFunctionImpl())
.assignTimestampsAndWatermarks(WatermarkStrategy
.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3)) // 设置最大等待时间为3s
.withTimestampAssigner((SerializableTimestampAssigner<WaterSensor>) (waterSensor, l) -> waterSensor.getTs() * 1000L)
);
// 对传感器做KeyBy
SingleOutputStreamOperator<String> process = sensorDS.keyBy(r -> r.getId())
.process(new KeyedProcessFunction<String, WaterSensor, String>() {
// 定义 ListState
ListState<Integer> vcListState;
// 初始化 ListState
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
vcListState = getRuntimeContext().getListState(new ListStateDescriptor<>("vcListState",Types.INT));
}
@Override
public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
// 将当前水位线存入 ListState
vcListState.add(value.getVc());
// 将 ListState (迭代器)中的值取出，拷贝到 List 中，
List<Integer> vcList = new ArrayList<Integer>();
for (Integer vc : vcListState.get()) {
vcList.add(vc);
}
// 排序
vcList.sort(((o1, o2) -> o2 - o1));
// 取前三
if (vcList.size() > 3) {
vcList.remove(3);
}
out.collect("当前传感器:"+value.getId()+",最大的3个水位线为:"+vcList.toString());
// 更新 ListState
vcListState.update(vcList);
}
});
process.print();
env.execute();
}
}
输入：
[root@VM-55-24-centos ~]# nc -lk 1234
s1,1,1
s1,4,4
s1,5,3
s1,6,6
s2,5,6
s3,6,5
s2,4,7
输出：
当前传感器:s1,最大的3个水位线为:[1]
当前传感器:s1,最大的3个水位线为:[4, 1]
当前传感器:s1,最大的3个水位线为:[4, 3, 1]
当前传感器:s1,最大的3个水位线为:[6, 4, 3]
当前传感器:s2,最大的3个水位线为:[6]
当前传感器:s3,最大的3个水位线为:[5]
当前传感器:s2,最大的3个水位线为:[7, 6]2.3 Map状态（MapState）
把一些键值对（key-value）作为状态整体保存起来，可以认为就是一组key-value映射的列表。使用与Map非常类似。
对Map状态的操作主要有以下：
// 根据key查询mapState中的value
UV get(UK key)
// 向mapState中put一个键值对
put(UK key, UV value)
// 向mapState中put多个键值对
putAll(Map<UK, UV> map)
// 将指定key对应的键值对删除
remove(UK key)
// 判断是否存在指定的key
boolean contains(UK key)
// 获取映射状态中所有的键值对
Iterable<Map.Entry<UK, UV>> entries()
// 获取映射状态中所有的键（key），返回一个可迭代Iterable类型
Iterable<UK> keys()
// 获取映射状态中所有的值（value），返回一个可迭代Iterable类型
Iterable<UV> values()
// 判断映射是否为空
boolean isEmpty()
案例：统计每种传感器每种水位值出现的次数。
public class KeyedMapStateDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
SingleOutputStreamOperator<WaterSensor> sensorDS = env
.socketTextStream("xxx.xxx.xxx.xxx", 1234)
.map(new MyMapFunctionImpl())
.assignTimestampsAndWatermarks(WatermarkStrategy
.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3)) // 设置最大等待时间为3s
.withTimestampAssigner((SerializableTimestampAssigner<WaterSensor>) (waterSensor, l) -> waterSensor.getTs() * 1000L)
);
// 对传感器做KeyBy
SingleOutputStreamOperator<String> process = sensorDS.keyBy(r -> r.getId())
.process(new KeyedProcessFunction<String, WaterSensor, String>() {
// 定义Map状态，键为vc(Integer),值为count(Integer)
MapState<Integer,Integer> vcCountMapState;
// 初始化Map状态
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
vcCountMapState = getRuntimeContext().getMapState(new MapStateDescriptor<Integer,Integer>("vcCountMapState",Types.INT,Types.INT));
}
@Override
public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
// 判断map状态中是否存在该vc，存在则count+1，否则put进map状态
Integer vc = value.getVc();
if(vcCountMapState.contains(value.getVc())){
Integer vcCount = vcCountMapState.get(vc);
vcCountMapState.put(vc , ++vcCount);
}else{
vcCountMapState.put(vc , 1);
}
StringBuilder outStr = new StringBuilder();
outStr.append("传感器:"+value.getId()+",下的所有水位线及出现次数：\n");
// 遍历该key下的所有键值
for (Map.Entry<Integer, Integer> entry : vcCountMapState.entries()) {
outStr.append("vc="+entry.getKey()+",count="+entry.getValue()+"\n");
}
outStr.append("------------------------------------------------------");
out.collect(outStr.toString());
}
});
process.print();
env.execute();
}
}输入：
[root@VM-55-27-centos ~]# nc -lk 1234
s1,1,1
s1,2,2
s1,3,1
s2,1,1
s1,4,1
输出：
传感器:s1,下的所有水位线及出现次数：
vc=1,count=1
------------------------------------------------------
传感器:s1,下的所有水位线及出现次数：
vc=1,count=1
vc=2,count=1
------------------------------------------------------
传感器:s1,下的所有水位线及出现次数：
vc=1,count=2
vc=2,count=1
------------------------------------------------------
传感器:s2,下的所有水位线及出现次数：
vc=1,count=1
------------------------------------------------------
传感器:s1,下的所有水位线及出现次数：
vc=1,count=3
vc=2,count=1
------------------------------------------------------2.4 归约状态（ReducingState）
类似于值状态（Value），不过需要对添加进来的所有数据进行归约，将归约聚合之后的值作为状态保存下来。
与之前不同的是，在归约状态描述器中需要传入ReduceFunction实现具体的归约逻辑。
对归约状态的操作主要有以下：
// 把新数据和之前的状态进行归约，并用得到的结果更新状态。
add(IN)
// 获取归约状态中的值
OUT get()案例：计算每种传感器的水位和。
public class KeyedReducingStateDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
SingleOutputStreamOperator<WaterSensor> sensorDS = env
.socketTextStream("xxx.xxx.xxx.xxx", 1234)
.map(new MyMapFunctionImpl())
.assignTimestampsAndWatermarks(WatermarkStrategy
.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3)) // 设置最大等待时间为3s
.withTimestampAssigner((SerializableTimestampAssigner<WaterSensor>) (waterSensor, l) -> waterSensor.getTs() * 1000L)
);
// 对传感器做KeyBy
SingleOutputStreamOperator<String> process = sensorDS.keyBy(r -> r.getId())
.process(new KeyedProcessFunction<String, WaterSensor, String>() {
// 定义Reducing状态
ReducingState<Integer> vcSumReducingState;
// 初始化Reducing状态(需要传入ReduceFunction实现具体的归约逻辑)
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
vcSumReducingState = getRuntimeContext().getReducingState(
new ReducingStateDescriptor<>(
"vcSumReducingState",
// 归约逻辑 （两数相加）
(v1, v2) -> v1+v2,
Types.INT
)
);
}
@Override
public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
vcSumReducingState.add(value.getVc());
out.collect("传感器:"+value.getId()+",水位线总值为:"+vcSumReducingState.get());
}
});
process.print();
env.execute();
}
}输入：
[root@VM-55-24-centos ~]# nc -lk 1234
s1,1,1
s1,2,2
s1,3,3
s2,4,4
s2,5,5
s1,6,6
s3,7,7
输出：
传感器:s1,水位线总值为:1
传感器:s1,水位线总值为:3
传感器:s1,水位线总值为:6
传感器:s2,水位线总值为:4
传感器:s2,水位线总值为:9
传感器:s1,水位线总值为:12
传感器:s3,水位线总值为:72.5 聚合状态（AggregatingState）
与归约状态非常类似，聚合状态也是一个值，用来保存添加进来的所有数据的聚合结果。并且允许输入、输出、中间累加器类型可以不一致。
对聚合状态的操作主要有以下：
// 向聚合状态中添加元素
add(IN)
// 从聚合状态中获取结果
OUT get()案例：计算每种传感器的平均水位。
public class KeyedAggregatingStateDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
SingleOutputStreamOperator<WaterSensor> sensorDS = env
.socketTextStream("xxx.xxx.xxx", 1234)
.map(new MyMapFunctionImpl())
.assignTimestampsAndWatermarks(WatermarkStrategy
.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3)) // 设置最大等待时间为3s
.withTimestampAssigner((SerializableTimestampAssigner<WaterSensor>) (waterSensor, l) -> waterSensor.getTs() * 1000L)
);
// 对传感器做KeyBy
SingleOutputStreamOperator<String> process = sensorDS.keyBy(r -> r.getId())
.process(new KeyedProcessFunction<String, WaterSensor, String>() {
// 定义聚合状态
AggregatingState<Integer,Double> vcAvgAggState;
// 初始化聚合状态
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
vcAvgAggState =
getRuntimeContext()
.getAggregatingState(new AggregatingStateDescriptor<>(
"vcAvgAggState",
// 聚合逻辑
new AggregateFunction<Integer, Tuple2<Integer, Integer>, Double>() {
// 初始化累加器
@Override
public Tuple2<Integer, Integer> createAccumulator() {
return Tuple2.of(0, 0);
}
// 累加逻辑 （水位相加，次数+1）
@Override
public Tuple2<Integer, Integer> add(Integer integer, Tuple2<Integer, Integer> accumulator) {
return Tuple2.of(accumulator.f0 + integer , accumulator.f1 + 1);
}
// 结果 水位 / 次数
@Override
public Double getResult(Tuple2<Integer, Integer> accumulator) {
return (accumulator.f0 * 1D) / accumulator.f1;
}
@Override
public Tuple2<Integer, Integer> merge(Tuple2<Integer, Integer> integerIntegerTuple2, Tuple2<Integer, Integer> acc1) {
return null;
}
},
Types.TUPLE(Types.INT, Types.INT)
));
}
@Override
public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
vcAvgAggState.add(value.getVc());
out.collect("传感器:"+value.getId()+",平均水位为:"+vcAvgAggState.get());
}
});
process.print();
env.execute();
}
}输入：
[root@VM-55-24-centos ~]# nc -lk 1234
s1,1,1
s1,2,2
s1,4,4
s2,5,5
s2,6,6
s1,7,7输出：
传感器:s1,平均水位为:1.0
传感器:s1,平均水位为:1.5
传感器:s1,平均水位为:2.3333333333333335
传感器:s2,平均水位为:5.0
传感器:s2,平均水位为:5.5
传感器:s1,平均水位为:3.52.6 状态生存时间（TTL）
随着Flink程序的运行，状态所消耗的存储空间也会随之增长，如果不限制则可能会导致存储空间耗尽。可以使用 .clear() 方法清除状态，但是不够灵活。
可以在状态描述器中通过.enableTimeToLive()方法启动TTL功能，并创建一个StateTtlConfig配置对象。
StateTtlConfig ttlConfig = StateTtlConfig
.newBuilder(Time.seconds(10))
.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
.build();
ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("my state", String.class);
// 开启TTL
stateDescriptor.enableTimeToLive(ttlConfig);
主要的配置项：
示例代码：
public class StateTTLDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
SingleOutputStreamOperator<WaterSensor> sensorDS = env
.socketTextStream("xxx.xxx.xxx.xxx", 1234)
.map(new MyMapFunctionImpl())
.assignTimestampsAndWatermarks(WatermarkStrategy
.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3)) // 设置最大等待时间为3s
.withTimestampAssigner((SerializableTimestampAssigner<WaterSensor>) (waterSensor, l) -> waterSensor.getTs() * 1000L)
);
SingleOutputStreamOperator<String> process = sensorDS.keyBy(r -> r.getId())
.process(new KeyedProcessFunction<String, WaterSensor, String>() {
ValueState<Integer> lastVcState;
@Override
public void open(Configuration parameters) throws Exception {
// 创建 StateTtlConfig
StateTtlConfig stateTtlConfig = StateTtlConfig
.newBuilder(Time.seconds(10)) // 状态存活时间为10s
.updateTtlOnCreateAndWrite() // 创建/更新状态时重置存活时间
.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) // 不返回过期状态
.build();
// 启用 TTL
ValueStateDescriptor<Integer> stateDescriptor = new ValueStateDescriptor<>("lastVc", Types.INT);
stateDescriptor.enableTimeToLive(stateTtlConfig);
lastVcState = getRuntimeContext().getState(stateDescriptor);
}
@Override
public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
out.collect("传感器:"+value.getId()+",当前状态值:"+lastVcState.value());
lastVcState.update(value.getVc());
}
});
process.print();
env.execute();
}
}输入：
[root@VM-55-24-centos ~]# nc -lk 1234
s1,1,1
...间隔10s
s1,2,2
s1,3,3
s1,4,4
s1,5,5输出：
传感器:s1,当前状态值:null
传感器:s1,当前状态值:1
传感器:s1,当前状态值:null
传感器:s1,当前状态值:3
传感器:s1,当前状态值:4三.算子状态（Operator State）
算子状态（Operator State）就是一个算子并行实例上定义的状态，作用范围被限定为当前算子任务。（每个算子子任务共享一个算子状态，子任务间不共享）
算子状态的实际应用场景不如Keyed State多，一般用在Source或Sink等与外部系统连接的算子上，一般使用不多。
当算子并行度发生变化时，算子状态也支持在并行的算子子任务实例间做重新分配，根据状态的类型不同，重组分配的方案也会不同。
算子状态也支持不同的结构类型，主要有三种：ListState、UnionListState和BroadcastState。
3.1 列表状态（ListState）
与Keyed State中的ListState一样，将状态表示为一组数据的列表。
与Keyed State中的列表状态的区别是，在算子状态的上下文中，不会按键（key）分别处理状态，所以每一个并行子任务上只会保留一个“列表”（list），也就是当前并行子任务上所有状态项的集合。列表中的状态项就是可以重新分配的最细粒度，彼此之间完全独立。
当算子并行度进行缩放调整时，算子的状态列表将会被全部收集收集起来，再通过轮询的方式重新依次分配给新的所有并行任务。
算子状态中不会存在“键组”（key group）这样的结构，所以为了方便重组分配，就把它直接定义成了“列表”（list）。这也就解释了，为什么算子状态中没有最简单的值状态（ValueState）。
案例：在map算子中计算数据的个数。
/**
* 在map算子中计算数据的个数
*/
public class OperatorListStateDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 全局算子并行度为2
env.setParallelism(2);
env
.socketTextStream("xxx.xxx.xxx.xxx", 1234)
.map(new MyCountMapFunction())
.print();
env.execute();
}
// 实现 CheckPointedFunction 接口
public static class MyCountMapFunction implements MapFunction<String,Long>, CheckpointedFunction {
// 定义本地变量
private Long count = 0L;
// 定义算子状态
private ListState<Long> state;
// Map算子逻辑
@Override
public Long map(String s) throws Exception {
return ++count;
}
/**
* 状态快照：用于将本地变量持久化至算子状态中，,开启checkpoint时才会调用
* @param context the context for drawing a snapshot of the operator
* @throws Exception
*/
@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
System.out.println("调用了snapshotState方法...");
// 清空状态
state.clear();
// 将本地变量存入状态中
state.add(count);
}
/**、
* 初始化本地变量：程序启动和恢复时，从状态中把数据添加到本地变量，每个子任务调用一次
* @param context the context for initializing the operator
* @throws Exception
*/
@Override
public void initializeState(FunctionInitializationContext context) throws Exception {
System.out.println("调用了initializeState方法...");
// 从上下文中获取算子状态
state = context
.getOperatorStateStore()
.getListState(new ListStateDescriptor<Long>("list-state", Types.LONG));
// 从算子状态中将数据拷贝至本地变量
if (context.isRestored()) { // 判断是否初始化成功
for (Long v : state.get()) {
count += v;
}
}
}
}
}输入：
[root@VM-55-24-centos ~]# nc -lk 1234
a
b
c
d
e
f
g
输出：
调用了initializeState方法...
调用了initializeState方法...
1> 1
2> 1
1> 2
2> 2
1> 3
2> 3
1> 43.2 联合列表状态（UnionListState）
与ListState类似，联合列表状态也会将状态表示为一个列表。它与常规列表状态的区别在于，算子并行度进行缩放调整时对于状态的分配方式不同。
在并行度进行缩放调整时，联合列表与普通列表不同，联合列表会将所有并行子任务的列表状态收集起来，并直接向所有并行子任务广播完整的列表。如果列表中状态项太多则不推荐使用联合里欸包状态。
使用上也与ListState类似，只需要在实现CheckpointedFunction类的initializeState方法时，通过上下文获取算子状态使用 .getUnionListState() 即可，其他与ListState无异。
state = context
.getOperatorStateStore()
.getUnionListState(new ListStateDescriptor<>("list-state", Types.LONG));3.3 广播状态（BroadcastState）
有时我们希望算子并行子任务都保持同一份“全局”状态，用来做统一的配置和规则设定。这时所有分区的所有数据都会访问到同一个状态，状态就像被“广播”到所有分区一样，这种特殊的算子状态，就叫作广播状态（BroadcastState）。
在并行度进行缩放操作时，由于是全局状态，也不会造成影响。
案例：水位超过指定的阈值发送告警，阈值可以动态修改。
/**
* 水位超过指定的阈值发送告警，阈值可以动态修改。
*/
public class OperatoBroadcastStateDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(2);
// 数据流
SingleOutputStreamOperator<WaterSensor> sensorDS = env
.socketTextStream("xxx.xxx.xxx.xxx", 1234)
.map(new MyMapFunctionImpl());
// 配置流：用于广播配置(阈值配置将发往这条流)
DataStreamSource<String> configDS = env.socketTextStream("xxx.xxx.xxx.xxx", 4321);
// 将配置流进行广播
MapStateDescriptor<String, Integer> broadcastMapState = new MapStateDescriptor<>("broadcast-state", Types.STRING, Types.INT);
BroadcastStream<String> configBS = configDS.broadcast(broadcastMapState);
// 将数据流和广播后的配置流使用connect进行连接
BroadcastConnectedStream<WaterSensor, String> sensorBCS = sensorDS.connect(configBS);
// 调用process
sensorBCS
.process(new BroadcastProcessFunction<WaterSensor, String, String>() {
/**
* 数据流的处理逻辑，可以通过上下文读取广播状态(只读)
* @param value The stream element.
* @param ctx
* @param out
* @throws Exception
*/
@Override
public void processElement(WaterSensor value, BroadcastProcessFunction<WaterSensor, String, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {
// 通过上下文获取广播状态的值(阈值)
ReadOnlyBroadcastState<String, Integer> broadcastState = ctx.getBroadcastState(broadcastMapState);
// 未从广播状态中读到值则设置默认值
Integer threshold = broadcastState.get("threshold") != null ? broadcastState.get("threshold"): 0;
if(value.getVc() > threshold){
out.collect("传感器:"+ value.getId()+",当前水位为:"+ value.getVc()+",触发了阈值:"+threshold);
}
}
/**
* 配置广播流的处理逻辑，可以通过上下文可以往广播状态写入值
* @param value The stream element.
* @param ctx
* @param out
* @throws Exception
*/
@Override
public void processBroadcastElement(String value, BroadcastProcessFunction<WaterSensor, String, String>.Context ctx, Collector<String> out) throws Exception {
// 读取流中的阈值，写入广播状态中
BroadcastState<String, Integer> broadcastState = ctx.getBroadcastState(broadcastMapState);
broadcastState.put("threshold" , Integer.valueOf(value));
}
})
.print();
env.execute();
}
}输入：
// 数据流：
[root@VM-55-24-centos ~]# nc -lk 1234
s1,1,1
s1,2,2
输出：
2> 传感器:s1,当前水位为:1,触发了阈值:0
1> 传感器:s1,当前水位为:2,触发了阈值:0输入：
// 广播配置流：
[root@VM-55-24-centos ~]# nc -lk 4321
10
// 数据流：
[root@VM-55-24-centos ~]# nc -lk 1234
s1,7,7
s1,11,11
输出：
1> 传感器:s1,当前水位为:11,触发了阈值:10应用场景：MySQL定义中一张配置表，定义一条配置流读取MySQL中的binlog，配置表如有修改，就将相应的配置广播出去，更改数据库即可实现线上程序动态配置。
四.状态后端（State Backends）
在Flink中，状态的存储、访问以及维护，都是由一个可插拔的组件决定的，这个组件就叫作状态后端（state backend）。状态后端主要负责管理本地状态的存储方式和位置。
4.1 状态后端的分类（HashMapStateBackend/RocksDB）
状态后端是一个“开箱即用”的组件，可以在不改变应用程序逻辑的情况下独立配置。Flink中提供了两类不同的状态后端，一种是“哈希表状态后端”（HashMapStateBackend），另一种是“内嵌RocksDB状态后端”（EmbeddedRocksDBStateBackend）。如果没有特别配置，系统默认的状态后端是HashMapStateBackend。
4.1.1 哈希表状态后端（HashMapStateBackend）
HashMapStateBackend是把状态存放在内存里。具体实现上，哈希表状态后端在内部会直接把状态当作对象（objects），保存在Taskmanager的JVM堆上。普通的状态，以及窗口中收集的数据和触发器，都会以键值对的形式存储起来，所以底层是一个哈希表（HashMap），这种状态后端也因此得名。
4.1.2 内嵌RocksDB状态后端（EmbeddedRocksDBStateBackend）
RocksDB是一种内嵌的key-value存储介质，可以把数据持久化到本地硬盘。配置EmbeddedRocksDBStateBackend后，会将处理中的数据全部放入RocksDB数据库中，RocksDB默认存储在TaskManager的本地数据目录里。
RocksDB的状态数据被存储为序列化的字节数组，读写操作需要序列化/反序列化，因此状态的访问性能要差一些。另外，因为做了序列化，key的比较也会按照字节进行，而不是直接调用.hashCode()和.equals()方法。
EmbeddedRocksDBStateBackend始终执行的是异步快照(快照时不会阻塞任务)，所以不会因为保存检查点而阻塞数据的处理；而且它还提供了增量式保存检查点的机制，这在很多情况下可以大大提升保存效率。
4.2 如何选择正确的状态后端
- 由此可以看出，虽然HashMapStateBackend的读写速度快，但是使用的是Taskmanager的JVM堆内存，如果存储的状态较大，则可能会将Taskmanager的内存耗尽。
- EmbeddedRocksDBStateBackend则存在Taskmanager的本地磁盘中，可以存储大的状态，不过牺牲了一定的读写速度。
4.3 状态后端的配置
在默认配置下，应用程序使用的默认状态后端是由集群配置文件flink-conf.yaml中指定的配置名称为state.backend，可修改为hashmap或rocksdb。除此之外，还可以在提交作业时通过参数设置状态后端、以及在代码中指定。
4.3.1 配置默认的状态后端
在flink-conf.yaml中，可以使用state.backend来配置默认状态后端。
配置项的可能值为hashmap，这样配置的就是HashMapStateBackend；如果配置项的值是rocksdb，这样配置的就是EmbeddedRocksDBStateBackend。
# 默认状态后端
state.backend: hashmap
4.2.2 为每个作业（Per-job/Application）单独配置状态后端
- 通过执行环境设置 hashMapStateBackend
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 设置状态后端为HashMap
HashMapStateBackend hashMapStateBackend = new HashMapStateBackend();
env.setStateBackend(hashMapStateBackend);- 通过执行环境设置 EmbeddedRocksDBStateBackend
在IDE使用EmbeddedRocksDBStateBackend则需要导入以下依赖：
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-statebackend-rocksdb</artifactId>
<version>${flink.version}</version>
</dependency>设置 EmbeddedRocksDBStateBackend 状态后端
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 设置状态后端为RocksDB
EmbeddedRocksDBStateBackend embeddedRocksDBStateBackend = new EmbeddedRocksDBStateBackend();
env.setStateBackend(embeddedRocksDBStateBackend);4.2.3 提交参数设置状态后端
[root@VM-55-24-centos flink-1.17.0]#
bin/flink run -m localhost:1234 -D state.backend=rocksdb -c com.xxx.wc.SocketStreamWordCount ./FlinkTutorial-1.0-SNAPSHOT.jar-D ：指定状态后端