<dependencies>
    <!-- Spring Boot Starter -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>

    <!-- Flink 核心依賴 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>1.17.2</version>
        <scope>provided</scope>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>1.17.2</version>
        <scope>provided</scope>
    </dependency>

???????   <!-- 本地執(zhí)行時(shí)需添加 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-runtime</artifactId>
        <version>1.17.2</version>
        <scope>test</scope>
    </dependency>
</dependencies>

// src/main/java/com/example/demo/flink/WordCountJob.java
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;

public class WordCountJob {
    public static void execute() throws Exception {
        // 1. 獲取 Flink 執(zhí)行環(huán)境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. 定義輸入數(shù)據(jù)
        DataStream<String> text = env.fromElements(
            "Spring Boot整合Flink",
            "Flink實(shí)時(shí)流處理",
            "Spring生態(tài)集成"
        );

        // 3. 處理數(shù)據(jù)流
        DataStream<Tuple2<String, Integer>> counts = text
            .flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
                @Override
                public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
                    for (String word : value.split("\\s")) {
                        out.collect(new Tuple2<>(word, 1));
                    }
                }
            })
            .keyBy(value -> value.f0) // 按單詞分組
            .sum(1); // 對(duì)計(jì)數(shù)求和

        // 4. 打印結(jié)果
        counts.print();

        // 5. 啟動(dòng)作業(yè)
        env.execute("WordCountJob");
    }
}

// src/main/java/com/example/demo/DemoApplication.java
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class DemoApplication {
    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
        System.out.println("Spring Boot Application Started...");

        try {
            // 觸發(fā) Flink 作業(yè)
            WordCountJob.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}