Flink流处理

1. Flink架构与核心概念

Apache Flink 是一个有状态的流处理框架，原生支持流处理，批处理被视为流处理的特例。

1.1 核心特性

1.2 运行架构

┌──────────────────────────────────────────────┐
│                JobManager                     │
│  ┌────────────┐  ┌────────────────────────┐  │
│  │ Dispatcher │  │  ResourceManager       │  │
│  └────────────┘  └────────────────────────┘  │
│  ┌────────────────────────────────────────┐  │
│  │  JobMaster (每个Job一个)               │  │
│  │  ├── ExecutionGraph                    │  │
│  │  └── CheckpointCoordinator             │  │
│  └────────────────────────────────────────┘  │
└──────────────────────────────────────────────┘
                    │
     ┌──────────────┼──────────────┐
     ▼              ▼              ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│TaskManager│ │TaskManager│ │TaskManager│
│ Slot 0   │ │ Slot 0   │ │ Slot 0   │
│ Slot 1   │ │ Slot 1   │ │ Slot 1   │
│ Task A   │ │ Task C   │ │ Task E   │
│ Task B   │ │ Task D   │ │ Task F   │
└──────────┘ └──────────┘ └──────────┘

1.3 作业执行层次

$\text{StreamGraph}\to \text{JobGraph}\to \text{ExecutionGraph}\to \text{Physical Execution}$

2. 窗口机制

窗口是流处理中将无限流切割为有限流的核心机制。

2.1 窗口类型

滚动窗口 (Tumbling Window):
|──── 5s ────|──── 5s ────|──── 5s ────|

滑动窗口 (Sliding Window):
|──── 5s ────|
   |──── 5s ────|
      |──── 5s ────|

会话窗口 (Session Window):
|── 3s ──|  gap  |──── 5s ────|  gap  |── 2s ──|

全局窗口 (Global Window):
|────────────────── 无限 ──────────────────|

2.2 窗口API

// 滚动窗口
dataStream.keyBy(t -> t.key)
    .window(TumblingEventTimeWindows.of(Time.seconds(5)));

// 滑动窗口
dataStream.keyBy(t -> t.key)
    .window(SlidingEventTimeWindows.of(Time.seconds(5), Time.seconds(1)));

// 会话窗口
dataStream.keyBy(t -> t.key)
    .window(EventTimeSessionWindows.withGap(Time.minutes(10)));

// 计数窗口
dataStream.keyBy(t -> t.key)
    .countWindow(100);  // 每100条触发

2.3 窗口触发器

3. 水位线（Watermark）

水位线是 Flink 处理事件时间和迟到数据的核心机制。

3.1 水位线定义

水位线是一个时间戳，表示到此时间点之前的所有数据应该已经到达：

$W(t)=\max (\text{eventTime})-\text{allowedLateness}$

事件时间轴:
  e1(3)  e2(5)  e3(4)  e4(8)  e5(7)  e6(10)
   │      │      │      │      │      │
   ▼      ▼      ▼      ▼      ▼      ▼
───┬──┬──┬──┬──┬──┬──┬──┬──┬──┬──→ 时间
   3  4  5  6  7  8  9  10

水位线: W=3  W=4  W=5  W=7  W=8  W=10

3.2 水位线生成策略

// 有序流（单调递增）
WatermarkStrategy.forMonotonousTimestamps()

// 乱序流（允许延迟）
WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5))

// 自定义
WatermarkStrategy.forGenerator(ctx -> new PunctuatedWatermarkGenerator())

有序流水位线：

$W_n=\max ({\text{eventTime}}_n)$

乱序流水位线：

$W_n=\max ({\text{eventTime}}_n)-\Delta$

其中 $\Delta$ 为最大允许乱序时间。

3.3 迟到数据处理

水位线 W=10
│
├── eventTime <= 10 → 正常处理
├── 10 < eventTime <= 10 + allowedLateness → 侧输出（可更新结果）
└── eventTime > 10 + allowedLateness → 丢弃

OutputTag<Event> lateEvents = new OutputTag<Event>("late-events"){};

dataStream.keyBy(t -> t.key)
    .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    .allowedLateness(Time.minutes(1))
    .sideOutputLateData(lateEvents)
    .process(new MyProcessFunction());

// 获取迟到数据
DataStream<Event> lateStream = result.getSideOutput(lateEvents);

4. 状态管理

4.1 状态类型

Keyed State 分类：

4.2 状态后端

// 配置RocksDB状态后端
env.setStateBackend(new EmbeddedRocksDBStateBackend());
env.getCheckpointConfig().setCheckpointStorage("hdfs://checkpoints");

5. Checkpoint与容错

5.1 Checkpoint机制

Flink 基于 Chandy-Lamport 算法的分布式快照实现容错：

Source-1 ──→ Map ──→ Sink
Source-2 ──→ Map ──→ Sink

Checkpoint Barrier注入:
Source-1 ──|B1|──→ Map ──|B1|──→ Sink
Source-2 ──|B1|──→ Map ──|B1|──→ Sink

所有算子收到Barrier后保存状态快照

流程：

1. JobManager 向所有 Source 注入 Checkpoint Barrier
2. Barrier 随数据流向下流动
3. 算子收到 Barrier 后对齐（所有输入的Barrier到齐）
4. 对齐后保存状态快照到持久化存储
5. 向下游转发 Barrier
6. 所有算子完成，Checkpoint 完成

5.2 Checkpoint配置

CheckpointConfig config = env.getCheckpointConfig();

// 开启Checkpoint，间隔1秒
env.enableCheckpointing(1000);

// 模式：精确一次 vs 至少一次
config.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// 超时时间
config.setCheckpointTimeout(60000);

// 最小间隔（防止Checkpoint过于频繁）
config.setMinPauseBetweenCheckpoints(500);

// 并发Checkpoint数
config.setMaxConcurrentCheckpoints(1);

// 保留策略
config.setExternalizedCheckpointCleanup(
    ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

// 允许的Checkpoint失败次数
config.setTolerableCheckpointFailureNumber(3);

5.3 Savepoint

Savepoint 是手动触发的、可迁移的Checkpoint：

# 触发Savepoint
flink savepoint <jobId> [targetDirectory]

# 从Savepoint恢复
flink run -s <savepointPath> -d <jarFile>

# 取消作业并触发Savepoint
flink cancel -s [targetDirectory] <jobId>

5.4 端到端Exactly-Once

实现端到端Exactly-Once需要两阶段提交（2PC）：

1. Checkpoint Barrier 到达 Sink
2. Sink 开启事务 → 写入外部系统
3. 所有算子完成状态快照
4. JobManager 确认 Checkpoint 完成
5. Sink 提交事务

支持Exactly-Once的Sink：Kafka、HDFS（通过两阶段提交）、数据库（通过XA事务）。

6. Flink SQL

-- 创建Kafka源表
CREATE TABLE orders (
    order_id BIGINT,
    user_id BIGINT,
    amount DECIMAL(10, 2),
    order_time TIMESTAMP(3),
    WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'orders',
    'properties.bootstrap.servers' = 'localhost:9092',
    'format' = 'json'
);

-- 窗口聚合查询
SELECT
    window_start,
    window_end,
    user_id,
    SUM(amount) AS total_amount,
    COUNT(*) AS order_count
FROM TABLE(
    HOP(TABLE orders, DESCRIPTOR(order_time), INTERVAL '1' HOUR, INTERVAL '24' HOUR)
)
GROUP BY window_start, window_end, user_id;