Flink操练（三十四）之自定义键控状态（四）ReducingState

　　一、ReducingState的方法
　　ReducingState是和ReduceFunction配合使用
　　get() 获取状态的值
　　add(IN value)方法添加一个元素，触发reduceFunction计算一次
　　二、ReducingState的描述器
　　ReducingState的描述器和之前ValueState、ListState不同，它得和一个ReduceFunction配合使用。三、统计单词
　　1、KeyedProcessFunction处理类package test; import org.apache.flink.api.common.functions.ReduceFunction; import org.apache.flink.api.common.functions.RichFlatMapFunction; import org.apache.flink.api.common.state.ReducingState; import org.apache.flink.api.common.state.ReducingStateDescriptor; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.configuration.Configuration; import org.apache.flink.util.Collector; /** * @Description: 求和 * @Param: * @return: * @Author: Mr.逗 * @Date: 2021/9/9 */ public class CountSumWithReduceState         extends RichFlatMapFunction, Tuple2> {      private ReducingState reducingState;      /***状态初始化*/     @Override     public void open(Configuration parameters) throws Exception {          ReducingStateDescriptor descriptor = new ReducingStateDescriptor(＂ReducingDescriptor＂, new ReduceFunction() {             @Override             public Long reduce(Long v1, Long v2) throws Exception {                 return v1 + v2;             }         },Long.class);         reducingState = getRuntimeContext().getReducingState(descriptor);     }      @Override     public void flatMap(Tuple2 element, Collector> collector) throws Exception {          //将状态放入         reducingState.add(element.f1);         collector.collect(Tuple2.of(element.f0,reducingState.get()));      }   }
　　2、主体类package test;  import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;  /**  * @program: bigdata_learn  * @description: 测试reduceState  * @author: Mr.逗  * @create: 2021-09-08 17:43  **/ public class TestKeyedReduceStateMain {     public static void main(String[] args) throws  Exception{         //获取执行环境         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();         //StreamExecutionEnvironment.getExecutionEnvironment();         //设置并行度         env.setParallelism(16);         //获取数据源         DataStreamSource> dataStreamSource =                 env.fromElements(                         Tuple2.of(1L, 3L),                         Tuple2.of(1L, 7L),                         Tuple2.of(2L, 4L),                         Tuple2.of(1L, 5L),                         Tuple2.of(2L, 2L),                         Tuple2.of(2L, 6L));           // 输出：         //(1,5.0)         //(2,4.0)         dataStreamSource                 .keyBy(0)                 .flatMap(new CountSumWithReduceState())                 .print();         String name = TestKeyedReduceStateMain.class.getName();         env.execute(name);     }  }
　　3、结果展示
　　四、计算最高温度
　　1、处理类package test; import org.apache.flink.api.common.functions.ReduceFunction; import org.apache.flink.api.common.functions.RichFlatMapFunction; import org.apache.flink.api.common.state.ReducingState; import org.apache.flink.api.common.state.ReducingStateDescriptor; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.configuration.Configuration; import org.apache.flink.util.Collector; /** * @Description: 求最大值 * @Param: * @return: * @Author: Mr.逗 * @Date: 2021/9/9 */ public class CountMaxWithReduceState         extends RichFlatMapFunction, Tuple2> {      private ReducingState reducingState;      /***状态初始化*/     @Override     public void open(Configuration parameters) throws Exception {          ReducingStateDescriptor descriptor = new ReducingStateDescriptor(＂ReducingDescriptor＂, new ReduceFunction() {             @Override             public Long reduce(Long v1, Long v2) throws Exception {                 return v1>=v2?v1:v2;             }         },Long.class);         reducingState = getRuntimeContext().getReducingState(descriptor);     }      @Override     public void flatMap(Tuple2 element, Collector> collector) throws Exception {          //将状态放入         reducingState.add(element.f1);         collector.collect(Tuple2.of(element.f0,reducingState.get()));      }   }
　　2、主体类package test;  import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;  /**  * @program: bigdata_learn  * @description: 测试reduceState  * @author: Mr.逗  * @create: 2021-09-08 17:43  **/ public class TestKeyedReduceStateMain {     public static void main(String[] args) throws  Exception{         //获取执行环境         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();         //StreamExecutionEnvironment.getExecutionEnvironment();         //设置并行度         env.setParallelism(16);         //获取数据源         DataStreamSource> dataStreamSource =                 env.fromElements(                         Tuple2.of(1L, 3L),                         Tuple2.of(1L, 7L),                         Tuple2.of(2L, 4L),                         Tuple2.of(1L, 5L),                         Tuple2.of(2L, 2L),                         Tuple2.of(2L, 6L));           // 输出：         //(1,5.0)         //(2,4.0)         dataStreamSource                 .keyBy(0)                 .flatMap(new CountMaxWithReduceState())                 .print();         String name = TestKeyedReduceStateMain.class.getName();         env.execute(name);     }  }
　　3、结果展示