如何在Apache Flink中用Java实现DataStream的平均运算

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-java</artifactId>
    <version>${flink.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
</dependency>

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

public class AverageCalculation {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 创建数据流
        DataStream<Integer> dataStream = env.fromElements(1, 2, 3, 4, 5);

        // 计算平均值
        DataStream<Double> averageStream = dataStream.map(new MapFunction<Integer, Tuple2<Integer, Integer>>() {
            @Override
            public Tuple2<Integer, Integer> map(Integer value) throws Exception {
                return new Tuple2<>(value, 1);
            }
        }).keyBy(0).reduce((Tuple2<Integer, Integer> value1, Tuple2<Integer, Integer> value2) ->
                new Tuple2<>(value1.f0 + value2.f0, value1.f1 + value2.f1)).map(new MapFunction<Tuple2<Integer, Integer>, Double>() {
            @Override
            public Double map(Tuple2<Integer, Integer> value) throws Exception {
                return (double) value.f0 / value.f1;
            }
        });

        // 打印结果
        averageStream.print();

        // 执行任务
        env.execute("Average Calculation");
    }
}

java -classpath <classpath> AverageCalculation