我是最近学习Apache Flink的超级新手，我想构建一个程序，实现：
当我对一串数字进行boardcast时，例如：Mqtt.fx中的3940351236363752（配置代码未显示，工作正常），我下面写的java程序可以排除离群值，这里是12和52，然后计算平均数，这是我的代码：

public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<String> stream = env.addSource(new MqttConsumer());

        DataStream<Integer> dataStream = stream.flatMap(new FlatMapFunction<String, Integer>() {
            @Override
            public void flatMap(String value, Collector<Integer> out) throws Exception {
                String[] str = value.split(" ");
                int temperature = Integer.parseInt(str[1]);
                out.collect(temperature);
            }
        }).setParallelism(1);

        

        DataStream<Integer> filteredDataStream = dataStream
                .filter(temperature -> temperature != findMaxTemperature(dataStream) && temperature != findMinTemperature(dataStream))
                .setParallelism(1);

        DataStream<Double> averageDataStream = filteredDataStream
                .map(temperature -> (double) temperature)
                .reduce(( t1,  t2) ->  t1 + t2)
                .map(sum -> sum / filteredDataStream.count())
                .setParallelism(1);

        averageDataStream.print();
        env.execute();
    }

    private static int findMaxTemperature(DataStream<Integer> dataStream) throws Exception {
        return dataStream.max(1).collect().get(0).getField(0);
    }

    private static int findMinTemperature(DataStream<Integer> dataStream) throws Exception {
        return dataStream.min(1).collect().get(0).getField(0);
    }

我还输入了这些包：

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

但是，在reduce下，有一个错误：The method reduce((<no type> t1, <no type> t2) -> {}) is undefined for the type SingleOutputStreamOperator<Double>
在min和max下，它说：The method max(int) is undefined for the type DataStream<Integer>。
这似乎是一些类型转换的问题，但我试图修改它，我也试图使用keyBy()，这是说是弃用，和maxBy()，这也不工作。有人能帮我吗？