第一章.Flink的状态编程

1.状态介绍

一.什么是状态

day06[Flink流处理高阶编程(下)] - 图1

  1.     在流式计算中有些操作一次处理一个独立的事件,比如解析一个事件,有些操作却需要记住多个事件的信息(比如窗口操作),那么需要记住多个事件信息的操作是有状态的
  2.     流式计算分为无状态计算和有状态计算两种情况
  3.     无状态计算观察每个独立事件,并根据最后一个事件输出结果,例如.流处理应用程序从传感器接收水位数据,并在水位超出指定高度时发出警告
  4.     有状态的计算则会基于多个事件输出结果,例如,计算过去一小时的平均水位,就是有状态的计算,所有用于复杂事件处理的状态机,例如,若在一分钟内收到两个相差20cm以上的水位差读数,则发出警告,这就是有状态的计算,所有流与流之间的关联操作,以及流与静态表或动态表的关联操作,都是有状态的计算

二.为什么需要管理状态

  1. 1. 去重
  2.     数据流中的数据有重复,我们想对重复数据去重,需要记录哪些数据已经流入过应用,当新数据流入时,根据已流入过的数据来判断去重
  3. 2. 检测
  4.     检查输入流是否符合某个特定的模式,需要将之前流入的元素以状态的形式缓存下来,比如,判断一个温度传感器数据流中的温度是否在持续上升
  5. 3. 聚合
  6.     对一个时间窗口内的数据进行聚合分析,分析一个小时内水位的情况
  7. 4. 更新机器学习模型
  8.     在线机器学习场景下,需要根据新流入数据不断更新机器学习的模型参数

三.Flink中的状态分类

Flink 包括两种基本类型的状态 Managed State 和Row State

Managed State Row State
状态管理方式 Flink Runtime托管,自动存储,自动恢复,自动伸缩 用户自己管理
状态数据结构 Flink提供多种常用的数据结构,例如ListState,MapState等 字节数组[]
使用场景 绝大数Flink算子 所有算子 
  1.     注意,从具体的使用场景来说,绝大多数的算子都可以通过继承Rich函数类或其他提供好的接口类,在里面使用Managed State,
  2.     Row State一般是在已有算子和Managed State 不够用时,用户自定义算子时使用
  3.     在平常的使用中Managed State已经足够使用,所以重点学习Managed State

对Managed State继续细分,它又有两种类型,Operator State(算子状态),Keyed State(键控状态)

Operator State Keyed State
适用算子类型 可用于所有算子,常用于source,sink,例如FlinkKafkaConsumer 只能用于KeyedStream上的算子
状态分配 一个算子的子任务对应一个状态 一个Key对应一个State:一个算子会处理多个Key,则访问相应的多个State
创建和访问方式 实现CheckpointedFunction或ListCheckpointed(已经过时)接口 重写RichFunction,通过里面的RuntimeContext访问
横向扩展 并发改变时有多重重写分配方式可选:均匀分配和合并后每个得到全量 并发改变,State随着key在实例间迁移
支持的数据结构 ListState和BroadCastState ValueState,ListState,MapState,ReduceState,AggregatingState

2.算子状态的使用

  1.     Operator State可以用在所有算子上,每个算子子任务或者说每个算子实例共享一个状态,流入这个算子子任务的数据可以访问和更新这个状态
  2.     算子子任务之间的状态不能互相访问
  3.     Operator State的实际应用场景不如Keyed State多,它经常被用在Source或Sink等算子上,用来保存流入数据的偏移量或对输出数据做缓存,以保证Flink应用的Exactly-Once语义
  1.     Flink为算子状态提供三种基本数据结构
  2.     1.列表状态，将状态表示为一组数据的列表
  3.     2.联合列表状态，也是将状态表示为数据的列表，它与常规列表状态的区别在于，在发生故障时，或者从保存点（savepoint）启动应用程序时如何恢复，一种是均匀分配，另外一种是将所有的State合并为全量State再分发给每个实例
  4.     3.广播状态（Broadcast state）,是一种特殊的算子状态，如果一个算子有多项任务，而他的每项任务状态又都相同，那么这种特殊情况最适合应用广播状态

一.列表状态

  1. package com.atguigu.flink.day06;
  3. import org.apache.flink.api.common.functions.RichMapFunction;
  4. import org.apache.flink.api.common.state.ListState;
  5. import org.apache.flink.api.common.state.ListStateDescriptor;
  6. import org.apache.flink.api.common.typeinfo.Types;
  7. import org.apache.flink.runtime.state.FunctionInitializationContext;
  8. import org.apache.flink.runtime.state.FunctionSnapshotContext;
  9. import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
  10. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  11. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  13. import java.util.ArrayList;
  14. import java.util.List;
  16. /**
  17.  * 列表状态
  18.  */
  20. public class $01_OperatorStateList {
  21.     public static void main(String[] args) throws Exception {
  22.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  23.         env.setParallelism(1);
  24.         DataStreamSource<String> inputDS = env.socketTextStream("hadoop102", 9999);
  25.         //读进来一行数据->切分成一个一个的单词,放到一个list里面
  26.         inputDS.map(new MyMap()).print();
  27.         env.execute();
  29.     }
  30.     public static class MyMap extends RichMapFunction<String,List<String>> implements CheckpointedFunction{
  31.         ListState<String> listState;
  32.         List<String> wordList = new ArrayList<>();
  34.         @Override
  35.         public List<String> map(String value) throws Exception {
  36.             String[] words = value.split(",");
  37.             for (String word : words) {
  38.                 wordList.add(word);
  39.             }
  40.             return wordList;
  41.         }
  43.         /**
  44.          * 对状态做快照(往状态里面存数据)
  45.          * @param context
  46.          * @throws Exception
  47.          */
  48.         @Override
  49.         public void snapshotState(FunctionSnapshotContext context) throws Exception {
  50.           listState.update(wordList);
  51.         }
  53.         /**
  54.          * 从备份恢复状态到本地
  55.          * @param context
  56.          * @throws Exception
  57.          */
  58.         @Override
  59.         public void initializeState(FunctionInitializationContext context) throws Exception {
  60.            listState =  context.getOperatorStateStore().getListState(new ListStateDescriptor<String>("list-state", Types.STRING));
  61.             Iterable<String> listStateIt = listState.get();
  62.             for (String element : listStateIt) {
  63.                 wordList.add(element);
  64.             }
  65.         }
  66.     }
  67. }

二.广播状态

  1.     从版本1.5.0开始, Apache Flink具有一种新的状态,称为广播状态
  2.     广播状态被引入以支持这样的用例:来自一个流的一些数据需要广播到所有下游任务,在那里它被本地存储,并用于处理另一个流上的所有传入元素,作为广播状态自然适合出现的一个例子.我们可以想象一个低吞吐量流,其中包含一个规则,我们希望根据来自另一个流的所有元素对这些规则进行评估,考虑到上述类型的用例,广播状态与其他算子状态的区别在于
  3.     1.它是一个map格式
  4.     2.它只对输入有广播流和无广播流的特定算子可用
  5.     3.这样的算子可以具有不同名称的多个广播状态
  1. package com.atguigu.flink.day06;
  3. import org.apache.flink.api.common.functions.RichMapFunction;
  4. import org.apache.flink.api.common.state.*;
  5. import org.apache.flink.api.common.typeinfo.Types;
  6. import org.apache.flink.runtime.state.FunctionInitializationContext;
  7. import org.apache.flink.runtime.state.FunctionSnapshotContext;
  8. import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
  9. import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
  10. import org.apache.flink.streaming.api.datastream.BroadcastStream;
  11. import org.apache.flink.streaming.api.datastream.DataStreamSource;
  12. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  13. import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
  14. import org.apache.flink.util.Collector;
  16. import java.util.ArrayList;
  17. import java.util.List;
  19. /**
  20.  * 列表状态
  21.  */
  23. public class $02_OperatorStateBroadCast {
  24.     public static void main(String[] args) throws Exception {
  25.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  26.         env.setParallelism(1);
  27.         DataStreamSource<String> inputDS = env.socketTextStream("hadoop162", 9999);
  28.         DataStreamSource<String> controlDS = env.socketTextStream("hadoop162", 8888);
  29.         //TODO 1.将流广播出去
  30.         MapStateDescriptor<String, String> broadcastStateDesc = new MapStateDescriptor<>("broadcast-state", Types.STRING, Types.STRING);
  31.         BroadcastStream<String> controlBS = controlDS.broadcast(broadcastStateDesc);
  32.         //TODO 2.连接主流和广播流
  33.         BroadcastConnectedStream<String, String> inputControlBCS = inputDS.connect(controlBS);
  34.         //选择做一个keyby,因为connect最好结合keyby使用
  35.         //TODO 3.调用算子处理
  36.         inputControlBCS
  37.                 .process(new BroadcastProcessFunction<String, String, Object>() {
  38.                     /**
  39.                      * 处理主流的数据
  40.                      * @param value
  41.                      * @param ctx
  42.                      * @param out
  43.                      * @throws Exception
  44.                      */
  45.                     @Override
  46.                     public void processElement(String value, ReadOnlyContext ctx, Collector<Object> out) throws Exception {
  47.                         //TODO 5.获取广播状态,进行相应处理
  48.                         ReadOnlyBroadcastState<String, String> broadcastState = ctx.getBroadcastState(broadcastStateDesc);
  49.                         String b = broadcastState.get("a");
  50.                         if("1".equals(b)){
  51.                             out.collect("走1的逻辑");
  52.                         }else if("0".equals(b)){
  53.                             out.collect("走0的逻辑");
  54.                         }else{
  55.                             out.collect("非法格式");
  56.                         }
  57.                     }
  59.                     /**
  60.                      * 处理广播流的数据
  61.                      * @param value
  62.                      * @param ctx
  63.                      * @param out
  64.                      * @throws Exception
  65.                      */
  66.                     @Override
  67.                     public void processBroadcastElement(String value, Context ctx, Collector<Object> out) throws Exception {
  68.                       //TODO 4.处理广播流,给广播状态赋值
  69.                         BroadcastState<String, String> broadcastState = ctx.getBroadcastState(broadcastStateDesc);
  70.                         //给广播状态赋值
  71.                         broadcastState.put("a",value);
  72.                     }
  73.                 })
  74.                 .print();
  75.         env.execute();
  78.     }
  79. }

3.键控状态的使用

一.介绍

键控状态是根据输入数据流中定义的键(key)来维护和访问的

Flink为每一个键值维护一个状态实例,并将具有相同键的所有数据,都分区到同一个算子任务中,这个任务会维护和处理这个key对应的状态,当任务处理一条数据时,它会自动将当前状态的访问权限限定为当前数据的key,因此具有相同key的所有数据都会访问相同的状态

KeyedState很类似于一个分布式的key-value map数据结构,只能用于KeyedStream(KeyBy算子之后)

  1. package com.atguigu.flink.day06;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
  5. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  6. import org.apache.flink.api.common.functions.MapFunction;
  7. import org.apache.flink.api.common.state.*;
  8. import org.apache.flink.api.common.typeinfo.Types;
  9. import org.apache.flink.configuration.Configuration;
  10. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  11. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  12. import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
  13. import org.apache.flink.util.Collector;
  15. import java.time.Duration;
  16. import java.util.ArrayList;
  17. import java.util.HashMap;
  19. public class $03_KeyedStateDemo1 {
  20.     public static void main(String[] args) throws Exception {
  21.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  22.         env.setParallelism(2);
  23.         SingleOutputStreamOperator<WaterSensor> sensorDS = env
  24.                 .socketTextStream("hadoop162", 9999)
  25.                 .map(new MapFunction<String, WaterSensor>() {
  26.                     @Override
  27.                     public WaterSensor map(String value) throws Exception {
  28.                         String[] line = value.split(",");
  29.                         return new WaterSensor(
  30.                                 line[0],
  31.                                 Long.parseLong(line[1]),
  32.                                 Integer.parseInt(line[2])
  33.                         );
  34.                     }
  35.                 })
  36.                 .assignTimestampsAndWatermarks(
  37.                         WatermarkStrategy
  38.                                 .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(4))
  39.                                 .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
  40.                                     @Override
  41.                                     public long extractTimestamp(WaterSensor element, long recordTimestamp) {
  42.                                         return element.getTs() * 1000;
  43.                                     }
  44.                                 })
  45.                 );
  46.         sensorDS
  47.                 .keyBy(sensor-> sensor.getId())
  48.                 .process(new KeyedProcessFunction<String, WaterSensor, String>() {
  49.                     //TODO 1.声明状态
  50.                     ValueState<String> valueState;
  51.                     ListState<Integer> listState;
  52.                     MapState<String,Long> mapState;
  53.                     @Override
  54.                     public void open(Configuration parameters) throws Exception {
  55.                         //TODO 2.在open方法里,通过运行上下文,初始化状态
  56.                          valueState = getRuntimeContext().getState(new ValueStateDescriptor<String>("value-state", Types.STRING));
  57.                          listState = getRuntimeContext().getListState(new ListStateDescriptor<Integer>("list-state", Types.INT));
  58.                          mapState = getRuntimeContext().getMapState(new MapStateDescriptor<String,Long>("map-state", Types.STRING,Types.LONG));
  59.                     }
  61.                     @Override
  62.                     public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
  63.                         //TODO 3.使用状态
  64.                         String value1 = valueState.value();//取出状态值
  65.                         valueState.update("haha");//更新状态值
  66.                         valueState.clear();//清空状态值
  67.                         //ListState的API
  68.                         Iterable<Integer> integers = listState.get();//取出状态值
  69.                         listState.update(new ArrayList<Integer>());//更新状态值
  70.                         listState.add(1);//添加单个值到List中
  71.                         listState.addAll(new ArrayList<Integer>());//添加整个list,不会覆盖
  72.                         listState.clear();//清空状态
  73.                         //MapState的API(参考HashMap操作)
  74.                         mapState.put("1",1L);
  75.                         mapState.get("1");
  76.                         mapState.remove("1");
  77.                         mapState.contains("1");
  78.                         mapState.putAll(new HashMap<>());
  79.                         mapState.keys();
  80.                         mapState.values();
  81.                         mapState.clear();
  83.                     }
  84.                 })
  85.                 .print();
  86.         env.execute();
  87.     }
  88. }
  1. package com.atguigu.flink.day06;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
  5. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  6. import org.apache.flink.api.common.functions.MapFunction;
  7. import org.apache.flink.api.common.state.*;
  8. import org.apache.flink.api.common.typeinfo.Types;
  9. import org.apache.flink.configuration.Configuration;
  10. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  11. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  12. import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
  13. import org.apache.flink.util.Collector;
  15. import java.time.Duration;
  16. import java.util.ArrayList;
  17. import java.util.HashMap;
  19. /**
  20.  * 键控状态注意:
  21.  *  总体原则:各组管各自的---->包括更新,清空
  22.  *     存数据的时候,不同组互不影响
  23.  *     clear的时候,哪个组调用的,只会清空自己组的
  24.  */
  26. public class $04_KeyedStateDemo2 {
  27.     public static void main(String[] args) throws Exception {
  28.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  29.         env.setParallelism(1);
  30.         SingleOutputStreamOperator<WaterSensor> sensorDS = env
  31.                 .socketTextStream("hadoop162", 9999)
  32.                 .map(new MapFunction<String, WaterSensor>() {
  33.                     @Override
  34.                     public WaterSensor map(String value) throws Exception {
  35.                         String[] line = value.split(",");
  36.                         return new WaterSensor(
  37.                                 line[0],
  38.                                 Long.parseLong(line[1]),
  39.                                 Integer.parseInt(line[2])
  40.                         );
  41.                     }
  42.                 })
  43.                 .assignTimestampsAndWatermarks(
  44.                         WatermarkStrategy
  45.                                 .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(4))
  46.                                 .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
  47.                                     @Override
  48.                                     public long extractTimestamp(WaterSensor element, long recordTimestamp) {
  49.                                         return element.getTs() * 1000L;
  50.                                     }
  51.                                 })
  52.                 );
  53.         sensorDS
  54.                 .keyBy(sensor-> sensor.getId())
  55.                 .process(new KeyedProcessFunction<String, WaterSensor, String>() {
  56.                     ValueState<Integer> lastVC;
  57.                     @Override
  58.                     public void open(Configuration parameters) throws Exception {
  59.                        lastVC = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("last-vc",Types.INT));
  60.                     }
  61.                     @Override
  62.                     public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
  63.                        ctx.timerService().registerEventTimeTimer(5000L);
  64.                         System.out.println("上一条的水位值="+lastVC.value());
  65.                         lastVC.update(value.getVc());
  66.                         System.out.println("当前水位值="+value.getVc());
  67.                     }
  68.                     @Override
  69.                     public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
  70.                         System.out.println("我被"+ ctx.getCurrentKey() + "触发了");
  71.                         lastVC.clear();
  72.                     }
  75.                 })
  76.                 .print();
  77.         env.execute();
  78.     }
  79. }

二.ReduceState

需求:计算每个传感器的水位和

  1. package com.atguigu.flink.day06;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
  5. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  6. import org.apache.flink.api.common.functions.MapFunction;
  7. import org.apache.flink.api.common.functions.ReduceFunction;
  8. import org.apache.flink.api.common.state.ReducingState;
  9. import org.apache.flink.api.common.state.ReducingStateDescriptor;
  10. import org.apache.flink.api.common.typeinfo.Types;
  11. import org.apache.flink.configuration.Configuration;
  12. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  13. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  14. import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
  15. import org.apache.flink.util.Collector;
  17. import java.time.Duration;
  19. /**
  20.  * 需求:计算各水位器的水位和
  21.  */
  23. public class $06_ReduceState {
  24.     public static void main(String[] args) throws Exception {
  25.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  26.         env.setParallelism(1);
  27.         SingleOutputStreamOperator<WaterSensor> sensorDS = env
  28.                 .socketTextStream("hadoop162", 9999)
  29.                 .map(new MapFunction<String, WaterSensor>() {
  30.                     @Override
  31.                     public WaterSensor map(String value) throws Exception {
  32.                         String[] line = value.split(",");
  33.                         return new WaterSensor(
  34.                                 line[0],
  35.                                 Long.parseLong(line[1]),
  36.                                 Integer.parseInt(line[2])
  37.                         );
  38.                     }
  39.                 })
  40.                 .assignTimestampsAndWatermarks(
  41.                         WatermarkStrategy
  42.                                 .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(4))
  43.                                 .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
  44.                                     @Override
  45.                                     public long extractTimestamp(WaterSensor element, long recordTimestamp) {
  46.                                         return element.getTs() * 1000L;
  47.                                     }
  48.                                 })
  49.                 );
  50.         sensorDS
  51.                 .keyBy(sensor-> sensor.getId())
  52.                 .process(new KeyedProcessFunction<String, WaterSensor, String>() {
  53.                     ReducingState<Integer> vcSumState;
  54.                     @Override
  55.                     public void open(Configuration parameters) throws Exception {
  57.                         vcSumState = getRuntimeContext().getReducingState(
  58.                                 new ReducingStateDescriptor<Integer>(
  59.                                         "vcSumState",
  60.                                         new ReduceFunction<Integer>(){
  61.                                             @Override
  62.                                             public Integer reduce(Integer value1, Integer value2) throws Exception {
  63.                                                 return value1 + value2;
  64.                                             }
  65.                                             },
  66.                                         Types.INT)
  69.                         );
  71.                     }
  73.                     @Override
  74.                     public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
  75.                          vcSumState.add(value.getVc());
  76.                          out.collect(vcSumState.get().toString());
  77.                     }
  81.                 })
  82.                 .print();
  83.         env.execute();
  84.     }
  85. }

三.AggregatingState

需求:求每个传感器的水位均值

  1. package com.atguigu.flink.day06;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
  5. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  6. import org.apache.flink.api.common.functions.AggregateFunction;
  7. import org.apache.flink.api.common.functions.MapFunction;
  9. import org.apache.flink.api.common.state.AggregatingState;
  10. import org.apache.flink.api.common.state.AggregatingStateDescriptor;
  11. import org.apache.flink.api.common.typeinfo.Types;
  12. import org.apache.flink.configuration.Configuration;
  13. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  14. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  15. import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
  16. import org.apache.flink.util.Collector;
  17. import org.apache.flink.api.java.tuple.Tuple2;
  19. import java.time.Duration;
  21. /**
  22.  * 需求:计算各水位器的水位均值
  23.  */
  25. public class $07_AggregatingState {
  26.     public static void main(String[] args) throws Exception {
  27.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  28.         env.setParallelism(1);
  29.         SingleOutputStreamOperator<WaterSensor> sensorDS = env
  30.                 .socketTextStream("hadoop162", 9999)
  31.                 .map(new MapFunction<String, WaterSensor>() {
  32.                     @Override
  33.                     public WaterSensor map(String value) throws Exception {
  34.                         String[] line = value.split(",");
  35.                         return new WaterSensor(
  36.                                 line[0],
  37.                                 Long.parseLong(line[1]),
  38.                                 Integer.parseInt(line[2])
  39.                         );
  40.                     }
  41.                 })
  42.                 .assignTimestampsAndWatermarks(
  43.                         WatermarkStrategy
  44.                                 .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(4))
  45.                                 .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
  46.                                     @Override
  47.                                     public long extractTimestamp(WaterSensor element, long recordTimestamp) {
  48.                                         return element.getTs() * 1000L;
  49.                                     }
  50.                                 })
  51.                 );
  52.         sensorDS
  53.                 .keyBy(sensor-> sensor.getId())
  54.                 .process(new KeyedProcessFunction<String, WaterSensor, String>() {
  55.                     AggregatingState<Integer,Double> vcSumAndCountState;
  56.                     @Override
  57.                     public void open(Configuration parameters) throws Exception {
  58.                         vcSumAndCountState = getRuntimeContext().getAggregatingState(
  59.                                 new AggregatingStateDescriptor<Integer, Tuple2<Integer,Integer>, Double>(
  60.                                         "vcSumAndCountState",
  61.                                         new AggregateFunction<Integer, Tuple2<Integer,Integer>, Double>() {
  62.                                             @Override
  63.                                             public Tuple2<Integer, Integer> createAccumulator() {
  64.                                                 return Tuple2.of(0,0);
  65.                                             }
  67.                                             @Override
  68.                                             public Tuple2<Integer, Integer> add(Integer value, Tuple2<Integer, Integer> accumulator) {
  69.                                                 //将水位加上去,将数量加1
  70.                                                 Integer lastSum = accumulator.f0;
  71.                                                 Integer lastCount = accumulator.f1;
  72.                                                 //将新的结果返回,封装成Tuple2返回
  73.                                                 return Tuple2.of(lastSum + value,lastCount+1);
  74.                                             }
  76.                                             @Override
  77.                                             public Double getResult(Tuple2<Integer, Integer> accumulator) {
  78.                                                 return accumulator.f0 * 1D / accumulator.f1;
  79.                                             }
  81.                                             @Override
  82.                                             public Tuple2<Integer, Integer> merge(Tuple2<Integer, Integer> a, Tuple2<Integer, Integer> b) {
  83.                                                 System.out.println("我merge了......");
  84.                                                 return null;
  85.                                             }
  86.                                         },
  87.                                         Types.TUPLE(Types.INT, Types.INT)
  88.                                 )
  89.                         );
  91.                     }
  93.                     @Override
  94.                     public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
  95.                           //将水位值放进状态里
  96.                         vcSumAndCountState.add(value.getVc());
  97.                         out.collect(vcSumAndCountState.get().toString());
  98.                     }
  102.                 })
  103.                 .print();
  104.         env.execute();
  105.     }
  106. }

4.状态练习

需求:监控水位传感器的水位值,如果水位值在5秒之内(event time)连续上升,则报警

提示:使用状态和定时器完成

  1. package com.atguigu.flink.day06;
  3. import com.atguigu.flink.day02.pojo.WaterSensor;
  4. import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
  5. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  6. import org.apache.flink.api.common.functions.MapFunction;
  7. import org.apache.flink.api.common.state.ValueState;
  8. import org.apache.flink.api.common.state.ValueStateDescriptor;
  9. import org.apache.flink.api.common.typeinfo.Types;
  10. import org.apache.flink.configuration.Configuration;
  11. import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
  12. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  13. import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
  14. import org.apache.flink.util.Collector;
  16. import java.time.Duration;
  18. /**
  19.  *
  20.  *需求:监控水位传感器的水位值,如果水位值在5秒之内(event time)连续上升,则报警
  21.  *
  22.  */
  24. public class $04_KeyedStateDemo3 {
  25.     public static void main(String[] args) throws Exception {
  26.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  27.         env.setParallelism(1);
  28.         SingleOutputStreamOperator<WaterSensor> sensorDS = env
  29.                 .socketTextStream("hadoop162", 9999)
  30.                 .map(new MapFunction<String, WaterSensor>() {
  31.                     @Override
  32.                     public WaterSensor map(String value) throws Exception {
  33.                         String[] line = value.split(",");
  34.                         return new WaterSensor(
  35.                                 line[0],
  36.                                 Long.parseLong(line[1]),
  37.                                 Integer.parseInt(line[2])
  38.                         );
  39.                     }
  40.                 })
  41.                 .assignTimestampsAndWatermarks(
  42.                         WatermarkStrategy
  43.                                 .<WaterSensor>forMonotonousTimestamps()
  44.                                 .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
  45.                                     @Override
  46.                                     public long extractTimestamp(WaterSensor element, long recordTimestamp) {
  47.                                         return element.getTs() * 1000L;
  48.                                     }
  49.                                 })
  50.                 );
  51.         sensorDS
  52.                 .keyBy(sensor-> sensor.getId())
  53.                 .process(new KeyedProcessFunction<String, WaterSensor, String>() {
  54.                     ValueState<Long> timeTsState;
  55.                     ValueState<Integer> lastVCState;
  56.                     @Override
  57.                     public void open(Configuration parameters) throws Exception {
  58.                         timeTsState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("timerTsState",Types.LONG));
  59.                         lastVCState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("lastVcState",Types.INT));
  60.                     }
  61.                     @Override
  62.                     public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
  63.                        if(timeTsState.value()==null){
  64.                            //说明没有注册过,可以注册
  65.                            timeTsState.update(ctx.timestamp() + 5000L);
  66.                            ctx.timerService().registerEventTimeTimer(timeTsState.value());
  67.                        }
  68.                        //判断是上升还是下降
  69.                         //如果是上升,啥也不干
  70.                         if(value.getVc() < (lastVCState.value() == null ? 0 : lastVCState.value())){
  71.                             //如果是下降=>删除定时器,重新注册5s后的定时器
  72.                             ctx.timerService().deleteEventTimeTimer(timeTsState.value());
  73.                             timeTsState.update(ctx.timestamp() + 5000L);
  74.                             ctx.timerService().registerEventTimeTimer(timeTsState.value());
  75.                         }
  76.                         //无论是上升还是下降,都必须将自己的水位值存下来
  77.                         lastVCState.update(value.getVc());
  78.                     }
  79.                     @Override
  80.                     public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
  81.                         out.collect(ctx.getCurrentKey()+"告警:检测到连续5s水位上涨,赶紧跑!");
  82.                         timeTsState.clear();
  83.                     }
  86.                 })
  87.                 .print();
  88.         env.execute();
  89.     }
  90. }

5.状态后端

状态后端主要负责两件事

  • 本地(taskmanager)的状态管理
  • 将检查点(checkpoint)状态写入远程存储

一.1.13之后版本的状态后端

  1. 本地状态位置
    1. 内存(TaskManager)
    2. 一个数据库 RocksDB
      1. 内嵌的,不需要我们单独安装
      2. 是一个k,v类型的数据库
      3. 存储的是序列化之后的数据, 读—>反序列化 写—>序列化
      4. 使用的是磁盘(数据先写入内存,刷写到磁盘,和HBase类似)
  1. checkpoint位置(对状态的备份)
    1. 内存(JobManager)
    2. 持久化的文件系统: HDFS

注意:使用RocksDB需要先导入依赖

  1. <dependency>
  2.     <groupId>org.apache.flink</groupId>
  3.     <artifactId>flink-statebackend-rocksdb_${scala.binary.version}</artifactId>
  4.     <version>${flink.version}</version>
  5. </dependency>

二.1.13之前的状态后端

1.13之前的状态后端分为三种,MemoryStateBackend(默认),FsStateBackend,RockedDBStateBackend

本地状态存哪里 checkpoint存哪里(备份) 使用场景
Memory TaskManager的内存 JobManager的内存 1.本地测试2.几乎无状态的作业3.不推荐在生产环境下使用
Fs TaskManager的内存 HDFS 1.常规使用状态的作业,例如分钟级别窗口聚合,join等2.需要开启HA作业3.可以应用在生产环境中
RocksDB RocksDB HDFS 1.超大状态的作业,例如天级的窗口聚合2.需要开启HA的作业3,对读写状态性能要求不高的作业4,可以使用在生产环境

三.代码示例

  1. package com.atguigu.flink.day06;
  2. import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
  3. import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
  4. import org.apache.flink.runtime.state.filesystem.FsStateBackend;
  5. import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
  6. import org.apache.flink.runtime.state.memory.MemoryStateBackend;
  7. import org.apache.flink.runtime.state.storage.JobManagerCheckpointStorage;
  8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  9. /**
  10.  * 状态后端
  11.  */
  13. public class $08_StateBackend {
  14.     public static void main(String[] args) throws Exception {
  15.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  16.         env.setParallelism(1);
  17.         //开启checkpoint
  18.         env.enableCheckpointing(5000L);
  19.         /**
  20.          * TODO 1.13版本开始的写法
  21.          */
  22.         //1.指定本地状态的类型
  23.         env.setStateBackend(new HashMapStateBackend());//本地状态存在内存
  24.         env.setStateBackend(new EmbeddedRocksDBStateBackend());//本地状态存在RocksDB
  25.         //2.指定CheckPoint的存储位置
  26.         env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());//checkPoint存在JM内存
  27.         env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop162:8020/flink/ck");//checkPoint存在HDFS
  29.         /**
  30.          * TODO 1.13版本之前的写法
  31.          */
  32.         //1.Memory
  33.         env.setStateBackend(new MemoryStateBackend());
  34.         //2.FS
  35.         env.setStateBackend(new FsStateBackend("hdfs://hadoop162:8020/flink/ck"));
  36.         //3.RocksDB
  37.         env.setStateBackend(new RocksDBStateBackend("hdfs://hadoop162:8020/flink/ck"));
  40.     }
  41. }

第二章.Flink的容错机制

1.状态一致性

  1.     1.一致性级别
  2.     在流处理中,一致性可以分为3个级别:
  3.     at most once(最多一次):这其实是没有正确性保障搞得委婉说法--故障发生之后,计数结果可能丢失
  4.     at least once(至少一次):这表示计数结果可能大于正确值,但绝不会小于正确值,也就是说,计数程序在发生故障时可能多算,但是绝不会少算
  5.     exactly once(严格一次):这指的是系统保证在发生故障后的计数结果与正确值一致,既不多算也不少算
  6.     Flink的一个重大价值在于,它即保证了exactly-once,又具有低延迟和高吞吐的处理能力
  1.     2.端到端的状态一致性
  2.     目前我们看到的一致性保证都是由流处理器保证的,也就是说都是在Flink流处理器内部保证的,而在真实应用中,流处理器应用除了流处理器以外还包含了数据源(例如Kafka)和输出到持久化系统,整个端到端的一致性级别取决于所有组件中一致性最弱的组件
  3.     Source端:需要外部源可重设数据的读取位置,Kafka Source具有这种特性,读取数据的时候可以指定offset
  4.     Flink内部:依赖checkpoint机制
  5.     Sink端:需要保证故障恢复时,数据不会重复写入外部系统,有两种实现方式
  6.         幂等写入:所谓幂等操作,是说一个操作,可以重复执行很多次,但只导致一次结果更改,也就是说,后面再重复执行就不起作用
  7.         事务性写入
  8.         需要构建事务来写入外部系统,构建的事务对应着checkpoint,等到checkpoint真正完成的时候,才把所有的结果写入sink系统中,对于事务性写入,具体又有两种实现方式,预写日志和两阶段提交
Sink\Source 不可重置 可重置
任意(Any) at most once at least once
幂等 at most once Exactly-once(短暂恢复时会出现暂时不一致)
预写日志(WAL) at most once at least once
两阶段提交(2PC) at most once Exactly-once

2.checkPoint与Barrier

  1.     Checkpoint 使 Flink 的状态具有良好的容错性，通过 checkpoint 机制，Flink 可以对作业的状态和计算位置进行恢复。
  2.     Flink 的 checkpoint 机制会和持久化存储进行交互，读写流与状态。一般需要：
  3.     1.一个能够回放一段时间内数据的持久化数据源，例如持久化消息队列（例如 Apache Kafka、RabbitMQ、 Amazon Kinesis、 Google PubSub 等）或文件系统（例如 HDFS、 S3、 GFS、 NFS、 Ceph 等）。
  4.     2.存放状态的持久化存储，通常为分布式文件系统（比如 HDFS、 S3、 GFS、 NFS、 Ceph 等）。
  5.     流的barrier是Flink的Checkpoint中的一个核心概念. 多个barrier被插入到数据流中, 然后作为数据流的一部分随着数据流动(有点类似于Watermark).这些barrier不会跨越流中的数据.
  6.     每个barrier会把数据流分成两部分: 一部分数据进入当前的快照 , 另一部分数据进入下一个快照 . 每个barrier携带着快照的id. barrier 不会暂停数据的流动, 所以非常轻量级.  在流中, 同一时间可以有来源于多个不同快照的多个barrier, 这个意味着可以并发的出现不同的快照.

day06[Flink流处理高阶编程(下)] - 图2

3.Flink的检查点制作过程

1.Checkpoint Coordinator 向所有 source 节点 trigger Checkpoint. 然后Source Task会在数据流中安插CheckPoint barrier

day06[Flink流处理高阶编程(下)] - 图3

2.source 节点向下游广播 barrier，这个 barrier 就是实现 Chandy-Lamport 分布式快照算法的核心，下游的 task 只有收到所有进来的 barrier 才会执行相应的 Checkpoint(barrier对齐, 但是新版本有一种新的: barrier)

day06[Flink流处理高阶编程(下)] - 图4

3.当 task 完成 state 备份后，会将备份数据的地址（state handle）通知给 Checkpoint coordinator。

day06[Flink流处理高阶编程(下)] - 图5

4.下游的 sink 节点收集齐上游两个 input 的 barrier 之后，会执行本地快照，这里特地展示了 RocksDB incremental Checkpoint 的流程，首先 RocksDB 会全量刷数据到磁盘上（红色大三角表示），然后 Flink 框架会从中选择没有上传的文件进行持久化备份（紫色小三角）

day06[Flink流处理高阶编程(下)] - 图6

5.同样的，sink 节点在完成自己的 Checkpoint 之后，会将 state handle 返回通知 Coordinator

day06[Flink流处理高阶编程(下)] - 图7

6.最后，当 Checkpoint coordinator 收集齐所有 task 的 state handle，就认为这一次的 Checkpoint 全局完成了，向持久化存储中再备份一个 Checkpoint meta 文件。

day06[Flink流处理高阶编程(下)] - 图8

4.严格一次语义(barrier对齐)

day06[Flink流处理高阶编程(下)] - 图9

5.至少一次语义:barrier不对齐

day06[Flink流处理高阶编程(下)] - 图10

6.SavePoint

  1. Flink 还提供了可以自定义的镜像保存功能，就是保存点（savepoints）
  2. 原则上，创建保存点使用的算法与检查点完全相同，因此保存点可以认为就是具有一些额外元数据的检查点
  3. Flink不会自动创建保存点，因此用户（或外部调度程序）必须明确地触发创建操作
  4. 保存点是一个强大的功能。除了故障恢复外，保存点可以用于：有计划的手动备份，更新应用程序，版本迁移，暂停和重启应用，等等
Savepoint Checkpoint
Savepoint是由命令触发, 由用户创建和删除 Checkpoint被保存在用户指定的外部路径中, flink自动触发
保存点存储在标准格式存储中，并且可以升级作业版本并可以更改其配置。 当作业失败或被取消时，将保留外部存储的检查点
用户必须提供用于还原作业状态的保存点的路径。 用户必须提供用于还原作业状态的检查点的路径。 如果是flink的自动重启, 则flink会自动找到最后一个完整的状态

7.Kafka+Flink+Kafka 实现端到端严格一次

day06[Flink流处理高阶编程(下)] - 图11

8.checkPoint config

  1. package com.atguigu.flink.day06;
  4. import org.apache.flink.api.common.restartstrategy.RestartStrategies;
  5. import org.apache.flink.api.common.time.Time;
  6. import org.apache.flink.streaming.api.CheckpointingMode;
  7. import org.apache.flink.streaming.api.environment.CheckpointConfig;
  8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  11. /**
  12.  * CheckPoint 配置
  13.  */
  15. public class $09_CheckPointConfig {
  16.     public static void main(String[] args) throws Exception {
  17.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  18.         env.setParallelism(1);
  19.         //TODO checkpoint配置
  20.         //1.开启checkpoint:生产上建议分钟级,3~10分
  21.         env.enableCheckpointing(5000L);
  22.         CheckpointConfig ckConfig = env.getCheckpointConfig();
  23.         //2.指定一致性级别:默认就是精准一次
  24.         ckConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
  25.         //3.设置两个checkpoint之间的最小间隔(上一次的结束到下一次的开始)
  26.         ckConfig.setMinPauseBetweenCheckpoints(3000L);
  27.         //4.设置超时时间,如果超时了,那么就失败了
  28.         ckConfig.setCheckpointTimeout(5000L);
  29.         //5.设置最大失败的次数
  30.         ckConfig.setTolerableCheckpointFailureNumber(3);
  31.         //6.设置job被cancel时,也会保留checkpoint
  32.         ckConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
  34.         //TODO Task FailOver:Task重试策略
  35.         /**
  36.          * 固定延迟重启策略:第一个参数:重试次数 第二个参数:重试的间隔
  37.          */
  38.         env.setRestartStrategy(RestartStrategies.fixedDelayRestart(5,3000L));
  39.         /**
  40.          * 失败率重试策略:
  41.          *  第一个参数:在指定时间范围内最大失败次数
  42.          *  第二个参数:指定的时间范围
  43.          *  第三个参数:重试的间隔
  44.          */
  45.         env.setRestartStrategy(RestartStrategies.failureRateRestart(5, Time.seconds(5),Time.seconds(3)));
  47.     }
  48. }