如何使用Flink 1.9 LAST_VALUE？

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
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.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

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

        // 配置Kafka消费者
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "flink-consumer");

        // 创建Kafka消费者
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("topic", new SimpleStringSchema(), properties);

        // 添加Kafka消费者到执行环境
        DataStream<String> stream = env.addSource(consumer);

        // 解析数据流
        DataStream<Tuple2<String, Integer>> parsedStream = stream.map(line -> {
            String[] parts = line.split(",");
            return new Tuple2<>(parts[0], Integer.parseInt(parts[1]));
        });

        // 使用LAST_VALUE函数获取最新值
        DataStream<Tuple2<String, Integer>> result = parsedStream
                .keyBy(0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .aggregate(new LastValueAggregate());

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

        // 执行任务
        env.execute("Flink Last Value Example");
    }

    // 自定义AggregateFunction实现最新值的获取
    public static class LastValueAggregate implements AggregateFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>> {
        @Override
        public Tuple2<String, Integer> createAccumulator() {
            return new Tuple2<>("", 0);
        }

        @Override
        public Tuple2<String, Integer> add(Tuple2<String, Integer> value, Tuple2<String, Integer> accumulator) {
            return value;
        }

        @Override
        public Tuple2<String, Integer> getResult(Tuple2<String, Integer> accumulator) {
            return accumulator;
        }

        @Override
        public Tuple2<String, Integer> merge(Tuple2<String, Integer> a, Tuple2<String, Integer> b) {
            return b;
        }
    }
}