1.3 Flink 环境搭建

<properties>
    <maven.compiler.release>17</maven.compiler.release>
    <flink.version>1.20.0</flink.version>
</properties>

<dependencies>
    <!-- Flink 核心 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>${flink.version}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients</artifactId>
        <version>${flink.version}</version>
    </dependency>

    <!-- Kafka 连接器 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka</artifactId>
        <version>3.3.0-1.20</version>
    </dependency>

    <!-- CEP -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-cep</artifactId>
        <version>${flink.version}</version>
    </dependency>

    <!-- RocksDB 状态后端（用到 EmbeddedRocksDBStateBackend 时必须显式引入） -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-statebackend-rocksdb</artifactId>
        <version>${flink.version}</version>
    </dependency>

    <!-- 测试 -->
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-test-utils</artifactId>
        <version>${flink.version}</version>
        <scope>test</scope>
    </dependency>
</dependencies>

import java.time.Duration;

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.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.util.Collector;

public class WordCount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置并行度（默认 = CPU 核心数）
        env.setParallelism(2);

        // Source：从 socket 读取（开发测试用）
        DataStream<String> text = env.socketTextStream("localhost", 9999);

        // Transformation：分词 + 计数
        DataStream<Tuple2<String, Integer>> wordCounts = text
            .flatMap(new Tokenizer())
            .keyBy(value -> value.f0)
            .window(TumblingProcessingTimeWindows.of(Duration.ofSeconds(5)))
            .sum(1);

        // Sink：打印到控制台
        wordCounts.print();

        // 启动作业
        env.execute("Socket WordCount");
    }

    static class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
            for (String word : value.split(" ")) {
                out.collect(new Tuple2<>(word.toLowerCase(), 1));
            }
        }
    }
}

# 启动 netcat 发送数据
nc -lk 9999

# 另开终端运行作业
mvn compile exec:java -Dexec.mainClass="WordCount"

# 在 nc 终端输入：
hello world
hello flink
hello java

# 控制台输出：
# (hello, 3)
# (world, 1)
# (flink, 1)
# (java, 1)

import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.test.util.MiniClusterWithClientResource;

public class FlinkTest {
    @ClassRule
    public static MiniClusterWithClientResource cluster =
        new MiniClusterWithClientResource(
            new MiniClusterResourceConfiguration.Builder()
                .setNumberSlotsPerTaskManager(2)
                .setNumberTaskManagers(1)
                .build());
}

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 并行度
env.setParallelism(4);

// 时间语义：由每个 Source 的 WatermarkStrategy 决定，没有全局开关

// Checkpoint 配置（第 5.3 节详细展开）
env.enableCheckpointing(60000);  // 每 60 秒 Checkpoint，语义默认 EXACTLY_ONCE

// 重启策略
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Duration.ofSeconds(10)));

// 状态后端
env.setStateBackend(new HashMapStateBackend());
env.getCheckpointConfig().setCheckpointStorage("file:///tmp/flink-checkpoints");

http://localhost:8081

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4);
env.enableCheckpointing(30000);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10000);
env.getCheckpointConfig().setCheckpointTimeout(60000);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(5, Duration.ofSeconds(30)));
env.setStateBackend(new EmbeddedRocksDBStateBackend());
env.getCheckpointConfig().setCheckpointStorage("file:///var/flink/checkpoints");