java Flink（十一）Flink状态之算子状态

浅浅的花香味﹌ 2022-11-16 10:56 228阅读 0赞

**状态编程：**

流式计算分为无状态和有状态两种情况。无状态的计算观察每个独立事件，并 根据最后一个事件输出结果。有状态的计算则会基于多个事件输出结果。

**算子状态：**

算子状态的作用范围限定为算子任务。这意味着由同一并行任务所处理的所有 数据都可以访问到相同的状态，状态对于同一任务而言是共享的。算子状态不能由 相同或不同算子的另一个任务访问。

package window;
    
    import beans.SensorReading;
    import kafka.KafkaConsumerPro;
    import org.apache.commons.collections.IteratorUtils;
    import org.apache.commons.math3.fitting.leastsquares.EvaluationRmsChecker;
    import org.apache.flink.api.common.functions.AggregateFunction;
    import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.api.common.functions.ReduceFunction;
    import org.apache.flink.api.common.serialization.SimpleStringSchema;
    import org.apache.flink.api.java.tuple.Tuple;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.TimeCharacteristic;
    import org.apache.flink.streaming.api.checkpoint.ListCheckpointed;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
    import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
    import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
    import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
    import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
    import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
    import org.apache.flink.streaming.api.windowing.time.Time;
    import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
    import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
    import org.apache.flink.util.Collector;
    import org.apache.flink.util.OutputTag;
    
    import java.util.Collections;
    import java.util.List;
    
    public class WindowTest1 {
        public static void main(String[] args) throws Exception{
            //获取当前执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            //设置并行度
            env.setParallelism(1);
            //设置时间语义为时间时间
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
            //从文件读取数据 ***demo模拟的文本流处理，其实不该用文本流处理，因为读取文本根本不许要分桶，速度过快了
            //DataStream<String> inputStream = env.readTextFile("D:\\idle\\FlinkTest\\src\\main\\resources");
            //从kafka读取数据
            DataStream<String> inputStream = env.addSource(new FlinkKafkaConsumer011<String>(kafkaConsumerTopic,
                    new SimpleStringSchema(), KafkaConsumerPro.getKafkaConsumerPro()));
    
            //转换成SensorReading类型
            DataStream<SensorReading> dataStream = inputStream.map(new MapFunction<String, SensorReading>() {
                public SensorReading map(String line) throws Exception {
                    String[] fields = line.split(",");
                    return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
                }
            });
    
            //定义一个有状态的map操作，统计当前分区数据个数
            SingleOutputStreamOperator<Integer> resultStream = dataStream.map(new MyCountMapper());
            env.execute();
    
        }
    
        public static class MyCountMapper implements MapFunction<SensorReading, Integer>, ListCheckpointed<Integer> {
            //定义一个本地变量，作为算子状态
            private Integer count = 0;
    
            public Integer map(SensorReading sensorReading) throws Exception {
                count ++;
                return count;
            }
            //状态快照,将状态保存到状态列表里
            public List<Integer> snapshotState(long checkpointId, long timestamp) throws Exception {
                return Collections.singletonList(count);//保存状态
            }
            //状态恢复
            public void restoreState(List<Integer> state) throws Exception {
                for(Integer num:state){
                    count+=num;
                }
            }
        }
    }