Flink DataStream之min()与minBy()，max()与max之间的区别

- 日理万妓 2022-01-22 08:41 278阅读 0赞

### Flink DataStream之min()与minBy()，max()与max之间的区别 ###

min():获取的最小值，指定的field是最小，但不是最小的那条记录，后面的示例会清晰的显示。

minBy():获取的最小值，同时也是最小值的那条记录。

max()与maxBy()的区别也是一样。

那么下面来看示例：

public class MinOrMinByOperator {
    
        public static void main(String[] args) throws Exception {
    
            StreamExecutionEnvironment sEnv = StreamExecutionEnvironment.getExecutionEnvironment();
            sEnv.setParallelism(1);
    
            Properties p = new Properties();
            p.setProperty("bootstrap.servers", "localhost:9092");
    
            DataStreamSource<String> source = sEnv.addSource(new FlinkKafkaConsumer010<String>("student", new SimpleStringSchema(), p));
    
            source.print();
    
            WindowedStream<Student, String, TimeWindow> windowedStream = source
                    .map(new MapFunction<String, Student>() {
                        @Override
                        public Student map(String value) throws Exception {
                            return new Gson().fromJson(value, Student.class);
                        }
                    })
                    .keyBy(new KeySelector<Student, String>() {
                        @Override
                        public String getKey(Student value) throws Exception {
                            return value.sid;
                        }
                    })
                    .timeWindow(Time.minutes(1));
    
            SingleOutputStreamOperator<Student> min = windowedStream.min("age");
            min.print("min  :");
    
            SingleOutputStreamOperator<Student> minBy = windowedStream.minBy("age");
            minBy.print("minBy:");
    
            sEnv.execute("MinOrMinByOperator");
        }
    }

模拟几条数据，输出结果如下

![image][]

[image]: /images/20220122/816a8115d705466ab78cf654347d1c26.png