2. Flink 基础回顾：架构、DataStream API、时间语义与状态后端

在深入 Flink CEP 与实时风控人群切片之前，必须夯实 Flink 的核心基础。本章将系统回顾 Flink 的运行时架构、DataStream API 编程模型、时间语义与 Watermark 机制，以及状态后端的选择与配置。这些知识点是理解后续复杂事件处理与状态管理的前提。

2.1 Flink 运行时架构

Flink 采用经典的 主从（Master-Worker）架构，由 JobManager 和 TaskManager 两个核心组件构成，配合 客户端（Client） 提交作业，形成完整的分布式执行链路。

2.2 DataStream API 编程模型

DataStream API 是 Flink 的核心抽象，用于构建有状态、无界数据流的处理逻辑。在风控场景中，我们主要使用以下算子构建处理管道：

以下是一个典型的实时风控 DataStream 管道代码片段，展示如何从 Kafka 读取事件、按用户分组并计算窗口内交易次数：

DataStream<TransactionEvent> sourceStream = env.fromSource(
    KafkaSource.<TransactionEvent>builder()
        .setBootstrapServers("localhost:9092")
        .setTopics("transaction-events")
        .setGroupId("risk-group")
        .setDeserializer(new TransactionDeserializer())
        .build(),
    WatermarkStrategy.<TransactionEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
        .withTimestampAssigner((event, timestamp) -> event.getEventTime()),
    "Kafka Source"
);

DataStream<TransactionEvent> filteredStream = sourceStream
    .filter(event -> event.getAmount() > 0);  // 过滤无效交易

DataStream<WindowResult> windowedStream = filteredStream
    .keyBy(TransactionEvent::getUserId)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .aggregate(new CountAggregate(), new WindowResultProcess());

windowedStream.addSink(new RiskResultSink());  // 写入风险结果

2.3 时间语义与 Watermark

Flink 支持三种时间语义：事件时间（Event Time）、摄入时间（Ingestion Time） 和 处理时间（Processing Time）。在实时风控中，事件时间是最常用的语义，因为它能准确反映事件发生的真实顺序，不受系统延迟或重试影响。

Watermark 是事件时间语义的核心机制，用于衡量事件时间的进度，并触发基于事件时间的窗口计算。Watermark 本质上是一个时间戳，表示“在此时间戳之前的事件已经全部到达”。

2.4 状态后端

Flink 的状态后端（State Backend）决定了状态数据的存储位置、持久化方式以及访问性能。在实时风控中，状态后端的选择直接影响作业的吞吐量、延迟和容错能力。

以下是一个典型的生产环境状态后端配置示例：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 设置 RocksDB 状态后端
EmbeddedRocksDBStateBackend rocksDBBackend = new EmbeddedRocksDBStateBackend(true);  // 开启增量 Checkpoint
rocksDBBackend.setDbStoragePath("/data/flink/rocksdb");  // 指定 RocksDB 本地存储路径
env.setStateBackend(rocksDBBackend);

// 设置 Checkpoint 存储路径（HDFS）
env.getCheckpointConfig().setCheckpointStorage("hdfs://namenode:8020/flink/checkpoints");
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setCheckpointInterval(60_000);  // 每分钟一次 Checkpoint
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30_000);
env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);