在Spring Boot中获取Flink聚合结果：无界数据源的挑战与策略

// 假设在Spring Boot中动态构建并提交Flink作业
public List<Tuple2<String, Long>> getAggregatedDataFromKafka(
    String topic, long startOffset, long endOffset, int partition) throws Exception {

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    // 生产环境建议使用RemoteEnvironment或Standalone模式
    env.setRuntimeMode(RuntimeMode.BATCH); // 对于有界查询，建议设置为BATCH模式

    // 构建Kafka源，指定起始和结束偏移量
    KafkaSource<String> source = KafkaSource.<String>builder()
        .setBootstrapServers("localhost:9092")
        .setTopics(topic)
        .setStartingOffsets(OffsetsInitializer.forSpecificOffsets(
            new HashMap<TopicPartition, Long>() {{
                put(new TopicPartition(topic, partition), startOffset);
            }}
        ))
        // 设置结束偏移量，将其变为有界源
        .setBoundedStopOffsets(OffsetsInitializer.forSpecificOffsets(
            new HashMap<TopicPartition, Long>() {{
                put(new TopicPartition(topic, partition), endOffset);
            }}
        ))
        .setValueOnlyDeserializer(new SimpleStringSchema())
        .build();

    DataStream<String> kafkaStream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Bounded Source");

    // 示例：简单词频统计
    DataStream<Tuple2<String, Long>> aggregatedStream = kafkaStream
        .flatMap((String value, Collector<Tuple2<String, Long>>) out -> {
            for (String word : value.split(" ")) {
                if (!word.isEmpty()) {
                    out.collect(new Tuple2<>(word, 1L));
                }
            }
        })
        .keyBy(0)
        .sum(1);

    // 将结果收集到List中 (适用于小规模数据，且会阻塞API调用)
    List<Tuple2<String, Long>> results = new ArrayList<>();
    try (CloseableIterator<Tuple2<String, Long>> it = aggregatedStream.executeAndCollect()) {
        it.forEachRemaining(results::add);
    }

    return results;
}

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.apache.flink.util.Collector;

public class FlinkRedisSinkJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(60000); // 启用检查点，每60秒一次

        // 假设从Kafka读取数据
        KafkaSource<String> source = KafkaSource.<String>builder()
            .setBootstrapServers("localhost:9092")
            .setTopics("my-topic")
            .setStartingOffsets(OffsetsInitializer.earliest()) // 无界源，从最早的可用偏移量开始
            .setValueOnlyDeserializer(new SimpleStringSchema())
            .build();

        DataStream<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

        // 示例：统计每个单词的出现次数
        DataStream<Tuple2<String, Long>> wordCounts = stream
            .flatMap((String value, Collector<Tuple2<String, Long>>) out -> {
                for (String word : value.split(" ")) {
                    if (!word.isEmpty()) {
                        out.collect(new Tuple2<>(word, 1L));
                    }
                }
            })
            .keyBy(0)
            .window(TumblingEventTimeWindows.of(Time.seconds(5))) // 5秒滚动窗口聚合
            .sum(1);