Flink之KeyedState

前面的文章中介绍过Operator State,这里介绍一下Keyed State.
在使用Operator State时必须要实现CheckpointFunction接口,而Keyed State则不需要,在使用keyBy(...)分组分组后,调用的函数必须是实现RichFuntion接口的函数才可以使用Keyed State.同样使用Keyed State也必须开启Checkpoint.

• 需求
  将接收到的Socket数据源中的字符串进行拼接
  在命令行开启socket命令:

  nc -lk 8888
  

• 业务代码

  public class FlinkKeyedState {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度为1,便于观察
        env.setParallelism(1);
        // 开启Checkpoint, 8秒一个周期并开启一次性语义
        env.enableCheckpointing(8000, CheckpointingMode.EXACTLY_ONCE);
        // 指定checkpoint持久化路径
        env.getCheckpointConfig().setCheckpointStorage("file:///Users/xxx/data/testData/checkpoint");
        // 开启Task级别故障自动failover,通过fixedDelayRestart设置Task重启上限和重启间隔,这里设置的重启次数为2次,一旦Task重启次数超过这个次数,整个job也会停止
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, Time.seconds(5)));
        // 获取Socket数据源
        DataStreamSource<String> socketSource = env.socketTextStream("localhost", 8888);
        // 将数据进行分组,将分组key给一个常量值
        SingleOutputStreamOperator<String> map = socketSource.keyBy(s -> "1")
                // 使用Keyed State的算子必须实现RichFunction接口,如RichMapFunction,ProcessFunction等
                .map(new RichMapFunction<String, String>() {
                    ListState<String> listState;
  
                    // open方法可以理解为和Operator State中的initializeState方法一样,需要在这个方法中构造和获取状态存储器
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // 获取上下文
                        RuntimeContext ctx = getRuntimeContext();
                        // 获取ListState,不同于Operator State的是在这里有更多的选择,如ListState,MapState等
                        listState = ctx.getListState(new ListStateDescriptor<>("demo", String.class));
                    }
  
                    // 在map方法中正常编写业务逻辑
                    @Override
                    public String map(String s) throws Exception {
                        // 模拟Task失败
                        if (s.equals("k") && RandomUtils.nextInt(0, 5) == 3) {
                            throw new Exception("Task 异常");
                        }
                        // 将数据添加到状态存储器中
                        listState.add(s);
  
                        Iterable<String> strings = listState.get();
                        StringBuilder builder = new StringBuilder();
                        for (String string : strings) {
                            builder.append(string);
                        }
                        return builder.toString();
                    }
                });
        map.print();
        env.execute("Keyed State");
    }
  }
  

  API的使用大概就这些内容,不过在使用Keyed Sate时首先要对keyBy的特性有所了解,才能得到最终想要的结果数据,如使用keyBy时上下游之间的数据分发模式、所设置的默认并行度、上下游算子的并行度是否一致等问题,这些都是需要注意的,然后根据实际业务需求开发对应的逻辑就可以了.