在Flink Streaming中按键对记录进行分组并收集到ListBuffer中

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import java.util.ArrayList;
import java.util.List;

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

        // 创建数据流
        DataStream<Record> stream = env.fromElements(
                new Record("key1", "value1"),
                new Record("key2", "value2"),
                new Record("key1", "value3"),
                new Record("key2", "value4")
        );

        // 按键对记录进行分组
        DataStream<Record> keyedStream = stream.keyBy(Record::getKey);

        // 定义滚动窗口，每5秒处理一次数据
        DataStream<List<Record>> windowedStream = keyedStream
                .timeWindow(Time.seconds(5))
                .aggregate(new RecordListAggregator());

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

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

    // 自定义聚合函数，将窗口中的数据收集到ListBuffer中
    public static class RecordListAggregator implements AggregateFunction<Record, List<Record>, List<Record>> {
        @Override
        public List<Record> createAccumulator() {
            return new ArrayList<>();
        }

        @Override
        public List<Record> add(Record value, List<Record> accumulator) {
            accumulator.add(value);
            return accumulator;
        }

        @Override
        public List<Record> getResult(List<Record> accumulator) {
            return accumulator;
        }

        @Override
        public List<Record> merge(List<Record> a, List<Record> b) {
            a.addAll(b);
            return a;
        }
    }

    // 数据记录类
    public static class Record {
        private String key;
        private String value;

        public Record(String key, String value) {
            this.key = key;
            this.value = value;
        }

        public String getKey() {
            return key;
        }

        public String getValue() {
            return value;
        }

        @Override
        public String toString() {
            return "Record{" +
                    "key='" + key + '\'' +
                    ", value='" + value + '\'' +
                    '}';
        }
    }
}