# FlinkTutorial

**Repository Path**: BruceCat/flink-tutorial

## Basic Information

- **Project Name**: FlinkTutorial
- **Description**: Flink学习
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-04-18
- **Last Updated**: 2026-04-21

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

请你拓展和丰富下面的Flink项目解读文档。补充一些更丰富的工作流程图

# Flink 1.17 入门教程 - 项目知识点总结

> 本项目是尚硅谷Flink教程的代码示例，涵盖了Flink核心知识点。本文档用大白话 + 举例的方式讲解每个知识点，让你零基础也能看懂！

---

## 📐 项目工程架构图

```
┌─────────────────────────────────────────────────────────────────────────────────────┐
│                                    Flink 应用程序                                    │
├─────────────────────────────────────────────────────────────────────────────────────┤
│                                                                                     │
│  ┌──────────────┐    ┌──────────────┐    ┌──────────────┐    ┌──────────────┐      │
│  │   Data Source │ -> │ Transformation│ -> │    Window    │ -> │     Sink     │      │
│  │   (数据源)    │    │   (转换算子)  │    │   (窗口)     │    │   (输出)     │      │
│  └──────────────┘    └──────────────┘    └──────────────┘    └──────────────┘      │
│         │                    │                    │                   │             │
│         v                    v                    v                   v             │
│  ┌────────────────────────────────────────────────────────────────────────────┐    │
│  │                        StreamExecutionEnvironment                          │    │
│  │                          (Flink执行环境)                                     │    │
│  └────────────────────────────────────────────────────────────────────────────┘    │
│                                                                                     │
└─────────────────────────────────────────────────────────────────────────────────────┘
                                          │
                                          ▼
┌─────────────────────────────────────────────────────────────────────────────────────┐
│                                    核心知识点                                        │
├─────────────┬─────────────┬─────────────┬─────────────┬─────────────┬─────────────┤
│   执行环境    │    数据源    │  转换算子    │    窗口     │   水印      │   状态      │
├─────────────┼─────────────┼─────────────┼─────────────┼─────────────┼─────────────┤
│  检查点      │    Sink     │   合流      │  定时器     │  算子链     │   分区      │
├─────────────┼─────────────┼─────────────┼─────────────┼─────────────┼─────────────┤
│ Table API   │  双流Join    │ Process函数 │  侧输出流   │  富函数     │  流批一体   │
└─────────────┴─────────────┴─────────────┴─────────────┴─────────────┴─────────────┘
```

---

## 📚 知识点详解

### 1. 执行环境 (ExecutionEnvironment)

大白话：就像你要开一家餐厅，首先得租个场地、买些设备。Flink也要先准备一个"执行环境"才能开始干活。

举例：去银行办业务，得先取号、排队，这个"取号机"就是执行环境。

代码示例：

```java
// 方式一：自动识别本地还是集群（最常用）
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 方式二：本地环境，IDE里调试用
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();

// 方式三：远程集群环境，连接别人的集群
StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment(
    "hadoop102", 8081, "/xxx.jar");

// 配置Web UI，方便在浏览器看
Configuration conf = new Configuration();
conf.set(RestOptions.BIND_PORT, "8082");
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
```

流批一体（重要！）：

```java
// 以前流处理和批处理是两套API，现在一套API搞定
env.setRuntimeMode(RuntimeExecutionMode.STREAMING);  // 流处理模式
env.setRuntimeMode(RuntimeExecutionMode.BATCH);      // 批处理模式
// 也可以不写，提交时用参数指定：-Dexecution.runtime-mode=BATCH
```

---

### 2. 数据源 (Source)

大白话：数据源就是Flink的"原材料供应商"，没有数据，Flink只能干瞪眼。

举例：工厂生产需要原材料，厨师做菜需要食材，Flink处理数据需要数据源。

| 数据源类型 | 说明                   | 示例代码                                  |
| ---------- | ---------------------- | ----------------------------------------- |
| 集合数据   | 测试用，从内存读取     | `env.fromElements(1,2,3)`                 |
| Socket流   | 监听网络端口收数据     | `env.socketTextStream("hadoop102", 7777)` |
| Kafka      | 消息队列，适合生产环境 | `KafkaSource.builder()...build()`         |
| 文件       | 读取文件内容           | `env.readTextFile("input/word.txt")`      |
| 数据生成器 | 自动生成测试数据       | `DataGeneratorSource`                     |

代码示例 - Kafka源：

```java
KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
    .setBootstrapServers("hadoop102:9092")  // Kafka地址
    .setGroupId("atguigu")                    // 消费者组
    .setTopics("topic_1")                     // 消费的Topic
    .setValueOnlyDeserializer(new SimpleStringSchema())  // 反序列化
    .setStartingOffsets(OffsetsInitializer.latest())       // 从最新开始消费
    .build();

env.fromSource(kafkaSource, WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(3)), "kafkasource")
    .print();
```

---

### 3. 转换算子 (Transformation)

大白话：转换算子就是对数据进行"加工"的工具，把原始数据变成我们想要的样子。

举例：菜市场买回来的菜，需要择菜、洗菜、切菜才能下锅。数据也是同样的道理。

| 算子        | 一句话解释                     | 图示                                |
| ----------- | ------------------------------ | ----------------------------------- |
| Map     | 一进一出，每个输入对应一个输出 | `1 -> 1`                            |
| FlatMap | 一进多出，把一条数据拆成多条   | `"hello world" -> "hello", "world"` |
| Filter  | 过滤，符合条件的留下           | `1,2,3,4 -> 保留偶数`               |
| KeyBy   | 分组，把相同key的数据放一起    | `s1,s1,s2 -> s1组, s2组`            |

代码示例：

```java
// Map: 把WaterSensor对象转换成它的id
sensorDS.map(sensor -> sensor.getId())

// FlatMap: 把一句话拆成单词
socketDS.flatMap((value, out) -> {
    String[] words = value.split(" ");
    for (String word : words) {
        out.collect(word);
    }
})

// Filter: 只保留水位值大于5的
sensorDS.filter(sensor -> sensor.getVc() > 5)

// KeyBy: 按传感器id分组
sensorDS.keyBy(sensor -> sensor.getId())
```

---

### 4. 窗口 (Window)

大白话：窗口就像把一条流动的河流分成一段一段的，然后对每一段进行分析。比如统计每分钟经过多少辆车。

举例：

• 滚动窗口：每60秒统计一次，不重叠。像按格子收钱，每个格子满了就结算。

• 滑动窗口：每30秒统计一次，但窗口长度60秒，会重叠。像移动的望远镜。

• 会话窗口：来了几辆车算一波，车走了超过5分钟没车，才算这一波结束。


窗口分类：

```
时间窗口（按时间分）：
├── 滚动窗口：每10秒一个窗口，不重叠
│   └── [0-10s], [10-20s], [20-30s]...
│
├── 滑动窗口：每10秒一个窗口，窗口长度30秒，会重叠
│   └── [0-30s], [10-40s], [20-50s]...
│
└── 会话窗口：两次数据间隔超过5秒，就开新窗口

计数窗口（按数据条数分）：
├── 滚动计数窗口：每5条数据一个窗口
│   └── 1-5条, 6-10条, 11-15条...
│
└── 滑动计数窗口：每5条一个窗口，步长2条
    └── 1-5条, 3-7条, 5-9条...
```

窗口函数：

| 函数类型                        | 说明                             | 适用场景               |
| ------------------------------- | -------------------------------- | ---------------------- |
| 增量聚合 (Reduce/Aggregate) | 来一条算一条，窗口结束时输出结果 | 数据量大，需要实时更新 |
| 全窗口函数 (Process)        | 数据都存起来，窗口结束时一次性算 | 需要取窗口首尾、排行   |

代码示例：

```java
// 时间滚动窗口，10秒
sensorKS.window(TumblingProcessingTimeWindows.of(Time.seconds(10)))

// 时间滑动窗口，窗口10秒，滑动2秒
sensorKS.window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(2)))

// 会话窗口，间隔5秒
sensorKS.window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)))

// 计数滚动窗口，5条数据
sensorKS.countWindow(5)

// 窗口函数
sensorWS.reduce((value1, value2) -> { /* 增量聚合 */ })
sensorWS.aggregate(new AggregateFunction<...>())  // 增量聚合
sensorWS.process(new ProcessWindowFunction<...>())  // 全窗口
```

---

### 5. 水印 (Watermark) - 重点难点！

大白话：水印是用来解决"乱序"问题的。想象一下考试交卷，老师收到卷子的顺序不是按学号来的（水印就是学号），老师需要根据学号（水印）来判断有没有收齐试卷。

举例：

• 你是监考老师，学生交卷有先有后（数据乱序）

• 学生试卷上写着交卷时间（水印 = 事件时间）

• 你发现9:05的试卷来了，但9:10的还没来，说明9:10的可能还在路上

• 等到看到9:10的试卷，就知道9:05之前的都收齐了（水印触发窗口计算）


为什么需要水印？
因为网络延迟、数据重发等原因，数据到达的顺序可能和发生顺序不一致。比如：

• 数据1：9:00发生，9:05到达

• 数据2：9:01发生，9:01到达（后发先至）


核心概念：

```
事件时间：数据本身携带的时间（发生时间）
处理时间：数据到达Flink系统的时间（到达时间）
水印：衡量事件时间进展的标志
```

代码示例：

```java
// 定义水印策略：允许乱序等待3秒
WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
    .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))  // 乱序等待3秒
    .withTimestampAssigner((element, ts) -> element.getTs() * 1000L);  // 从数据中提取时间戳

// 使用水印
sensorDS.assignTimestampsAndWatermarks(watermarkStrategy);

// 单调水印（数据按时间顺序到达）
WatermarkStrategy.forMonotonousTimestamps()
```

水印传递规则：

• 上游接收多个水印 → 取最小值（取最保守的，保证数据不丢失）

• 往下游发送 → 广播（所有下游都能收到）


---

### 6. 状态 (State) - 核心重点！

大白话：状态就是Flink的"记忆"。比如你要统计每个传感器的水位总和，需要记住每个传感器之前的水位是多少，这个"记忆"就是状态。

举例：

• 统计单词出现次数：需要记住每个单词之前出现过多少次

• 检测异常：需要记住上一条水位值，和当前比较

• 窗口计算：需要记住窗口内有哪些数据


状态分类：

```
按作用范围分：
├── Keyed State（键控状态）
│   └── 只能用在keyBy之后，每个key有自己的状态
│   ├── ValueState: 存一个值（类似单个变量）
│   ├── ListState: 存一个列表（类似ArrayList）
│   ├── MapState: 存键值对（类似HashMap）
│   ├── ReducingState: 存一个值，每次添加自动聚合
│   └── AggregatingState: 存一个值，自定义聚合逻辑
│
└── Operator State（算子状态）
    └── 所有数据共享一个状态（不常用）
    ├── ListState: 存列表
    └── BroadcastState: 广播状态，所有并行实例都能读到
```

代码示例：

```java
sensorDS.keyBy(r -> r.getId())
    .process(new KeyedProcessFunction<...>() {
        
        // 1. 定义状态
        ValueState<Integer> lastVcState;
        
        @Override
        public void open(Configuration parameters) {
            // 2. 初始化状态
            lastVcState = getRuntimeContext().getState(
                new ValueStateDescriptor<>("lastVcState", Types.INT)
            );
        }
        
        @Override
        public void processElement(WaterSensor value, Context ctx, Collector<String> out) {
            // 3. 使用状态
            Integer lastVc = lastVcState.value();  // 读取
            lastVcState.update(value.getVc());     // 更新
            lastVcState.clear();                   // 清除
        }
    });
```

状态后端（State Backend）：

| 后端类型            | 存储位置           | 优点           | 缺点                   |
| ------------------- | ------------------ | -------------- | ---------------------- |
| HashMapStateBackend | TaskManager堆内存  | 读写快         | 内存有限，存不了太多   |
| RocksDBStateBackend | 本地 RocksDB数据库 | 可以存海量数据 | 读写稍慢（需要序列化） |

```java
// 使用HashMap后端
env.setStateBackend(new HashMapStateBackend());

// 使用RocksDB后端
env.setStateBackend(new EmbeddedRocksDBStateBackend());
```

状态TTL（生存时间）：

```java
// 状态只能存活5秒，过期自动清除
StateTtlConfiguration ttlConfig = StateTtlConfiguration.newBuilder(Duration.ofSeconds(5))
    .setUpdateType(StateTtlConfiguration.UpdateType.OnReadAndWrite)
    .setStateVisibility(StateTtlConfiguration.StateVisibility.NeverReturnExpired)
    .build();

ValueStateDescriptor<Integer> descriptor = new ValueStateDescriptor<>("state", Types.INT);
descriptor.enableTimeToLive(ttlConfig);
```

---

### 7. 检查点 (Checkpoint) - 核心重点！

大白话：检查点就是给Flink照"快照"，保存当前的处理进度。这样如果Flink程序崩溃，可以从快照恢复，不用从头开始。

举例：

• 写论文时每隔10分钟自动保存一次（检查点）

• 电脑突然蓝屏，从上次保存的地方恢复，不用重写（从检查点恢复）


检查点模式：

| 模式                         | 说明                   | 比喻                           |
| ---------------------------- | ---------------------- | ------------------------------ |
| 精准一次 (EXACTLY_ONCE)  | 数据既不丢失也不重复   | 银行存款，转账要么成功要么失败 |
| 至少一次 (AT_LEAST_ONCE) | 数据不丢失，但可能重复 | 发短信，可能收到多条           |

核心配置：

```java
// 开启检查点，每5秒做一次，精准一次模式
env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);

// 检查点存储位置
checkpointConfig.setCheckpointStorage("hdfs://hadoop102:8020/chk");

// checkpoint超时时间
checkpointConfig.setCheckpointTimeout(60000);

// 最多同时存在几个checkpoint
checkpointConfig.setMaxConcurrentCheckpoints(1);

// 最小间隔时间
checkpointConfig.setMinPauseBetweenCheckpoints(1000);

// 取消作业时是否保留检查点
checkpointConfig.setExternalizedCheckpointCleanup(
    CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

// 允许连续失败的次数
checkpointConfig.setTolerableCheckpointFailureNumber(10);
```

Barrier对齐：

```
对齐版本（精准一次）：
Task收到 所有上游 的barrier → 才做checkpoint
barrier后面的数据 → 阻塞等待
优点：数据不重复
缺点：一条流慢了会阻塞整个job

非对齐版本（1.11+）：
Task收到 第一个 barrie r→ 立即开始做checkpoint
barrier后面的数据 → 继续计算，同时写入checkpoint
优点：不会因为一条流慢而阻塞
缺点：需要额外空间存储正在传输的数据
```

```java
// 开启非对齐检查点
checkpointConfig.enableUnalignedCheckpoints();
checkpointConfig.setAlignedCheckpointTimeout(Duration.ofSeconds(1));
```

---

### 8. Sink (输出)

大白话：Sink就是Flink的"出货口"，处理完的数据要放到指定地方。

举例：工厂生产的产品需要打包、发货到各个销售渠道。

常用Sink：

| Sink类型 | 说明         | 代码位置          |
| -------- | ------------ | ----------------- |
| Kafka    | 写到消息队列 | `SinkKafka.java`  |
| 文件     | 写到文件系统 | `SinkFile.java`   |
| MySQL    | 写到数据库   | `SinkMySQL.java`  |
| 自定义   | 自己实现     | `SinkCustom.java` |

Kafka Sink示例：

```java
KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
    .setBootstrapServers("hadoop102:9092")
    .setRecordSerializer(
        KafkaRecordSerializationSchema.<String>builder()
            .setTopic("ws")
            .setValueSerializationSchema(new SimpleStringSchema())
            .build()
    )
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)  // 精准一次
    .setTransactionalIdPrefix("atguigu-")                   // 事务前缀
    .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "600000")  // 事务超时
    .build();

sensorDS.sinkTo(kafkaSink);
```

精准一次写入Kafka的必要条件：

1. 开启Checkpoint
2. 设置事务前缀
3. 设置事务超时时间（Checkpoint间隔 < 事务超时 < 15分钟）

---

### 9. 合流

大白话：合流就是把两条或多条数据流"合并"成一条来处理。

Connect（连接）：

```java
// 特点：只能连2条流，类型可以不同，处理逻辑分开
ConnectedStreams<Integer, String> connect = source1.connect(source2);

connect.map(new CoMapFunction<Integer, String, String>() {
    @Override
    public String map1(Integer value) { return "数字流:" + value; }
    @Override
    public String map2(String value) { return "字母流:" + value; }
});
```

Union（联合）：

```java
// 特点：可以连多条流，类型必须相同，合并在一起
DataStream<Integer> union = stream1.union(stream2, stream3);
```

---

### 10. 定时器 (Timer)

大白话：定时器就像闹钟，到点了喊你干活。

举例：

• 事件时间定时器：等水位数据，等啊等，等到水印到了9点，触发"到点啦，该统计了"

• 处理时间定时器：设置一个5秒后的闹钟，5秒后自动触发


代码示例：

```java
sensorKS.process(new KeyedProcessFunction<...>() {
    @Override
    public void processElement(WaterSensor value, Context ctx, Collector<String> out) {
        TimerService timerService = ctx.timerService();
        
        // 注册事件时间定时器（等水印触发）
        timerService.registerEventTimeTimer(5000L);
        
        // 注册处理时间定时器（等5秒触发）
        timerService.registerProcessingTimeTimer(
            timerService.currentProcessingTime() + 5000L);
    }
    
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) {
        // 定时器触发时执行这里
        System.out.println("定时器触发，时间是：" + timestamp);
    }
});
```

注意：

• 只有KeyedStream才能用定时器

• 事件时间定时器依赖水印触发


---

### 11. 算子链 (Operator Chain)

大白话：算子链就是把多个算子"链"在一起，像流水线一样，一条数据过来一次性经过多个算子处理，减少网络传输开销。

举例：

• 洗菜 → 切菜 → 炒菜，如果每一步都把菜端到下一个工序，效率低

• 改进：在同一个台面上连续完成，效率高（这就是算子链）


链化条件：

1. 一对一的数据传输（forward）
2. 并行度相同

代码示例：

```java
// 全局禁用算子链
env.disableOperatorChaining();

// 某个算子不参与链化
socketDS.flatMap(...).disableChaining()

// 从某个算子开启新链
socketDS.flatMap(...).startNewChain().map(...)
```

---

### 12. 分区策略 (Partition)

大白话：分区就是决定数据"去哪个子任务"处理。就像分快递，不同区域的快递送到不同的配送站。

分区策略：

| 分区策略      | 说明                           | 比喻             |
| ------------- | ------------------------------ | ---------------- |
| Forward   | 一对一，上下游并行度相同       | 直辖市，直接管辖 |
| Shuffle   | 随机分发                       | 摇号，随机分配   |
| Rebalance | 轮询分发，解决数据倾斜         | 轮流发牌         |
| Rescale   | 局部轮询，更高效               | 分区轮询         |
| Broadcast | 复制到所有子任务               | 群发通知         |
| Global    | 全部发往第一个子任务           | 总裁专线         |
| KeyBy     | 按key分发，相同key去同一子任务 | 按部门分组       |

代码示例：

```java
socketDS.shuffle().print();       // 随机
socketDS.rebalance().print();     // 轮询
socketDS.rescale().print();       // 局部轮询
socketDS.broadcast().print();     // 广播
socketDS.global().print();        // 全局
sensorDS.keyBy(sensor -> sensor.getId()).print();  // 按key
```

---

### 13. Table API & SQL

大白话：用写SQL的方式来做流处理，不用写Java代码，适合数据分析师。

举例：就像用Excel的公式 vs 用Python处理数据，SQL更简单直观。

代码示例：

```java
// 创建表环境
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

// 创建源表（从Kafka读）
tableEnv.executeSql("CREATE TABLE source ( \
    id INT, \
    ts BIGINT, \
    vc INT\
) WITH ( \
    'connector' = 'kafka', \
    'topic' = 'ws', \
    'properties.bootstrap.servers' = 'hadoop102:9092', \
    'format' = 'json'\
);");

// 创建目标表（打印到控制台）
tableEnv.executeSql("CREATE TABLE sink (\
    id INT, \
    sumVC INT\
) WITH (\
    'connector' = 'print'\
);");

// 执行SQL查询
Table result = tableEnv.sqlQuery("select id, sum(vc) as sumVC from source group by id");

// 输出
result.executeInsert("sink");
```

自定义函数：

```java
// 标量函数：一进一出
public class MyScalarFunction extends ScalarFunction {
    public String eval(String s) { return s.toUpperCase(); }
}

// 聚合函数：多进一出
public class MyAggregateFunction extends AggregateFunction<Long, Long> {
    @Override
    public Long createAccumulator() { return 0L; }
    @Override
    public Long add(Long value, Long accumulator) { return value + accumulator; }
    @Override
    public Long getResult(Long accumulator) { return accumulator; }
}
```

---

### 14. 双流Join

大白话：双流Join就是"红娘"，把两条流中有关系的数据配对。

Window Join（窗口Join）：

• 两条流的数据必须在同一个时间窗口内才能匹配

• 类似SQL的inner join


```java
ds1.join(ds2)
    .where(r1 -> r1.f0)  // ds1的key
    .equalTo(r2 -> r2.f0)  // ds2的key
    .window(TumblingEventTimeWindows.of(Time.seconds(10)))
    .apply((first, second) -> first + "<--->" + second);
```

Interval Join（区间Join）：

• 基于事件时间，在指定时间范围内匹配

• 比如：ds1的数据和ds2前后2秒内的数据匹配


```java
ks1.intervalJoin(ks2)
    .between(Time.seconds(-2), Time.seconds(2))
    .process(new ProcessJoinFunction<...>() {
        @Override
        public void processElement(Tuple2<String, Integer> left, 
                                   Tuple3<String, Integer, Integer> right, 
                                   Context ctx, Collector<String> out) {
            out.collect(left + "<------>" + right);
        }
    });
```

---

### 15. Process函数

大白话：Process函数是"万能函数"，可以做到前面所有算子能做到的事，还能做到它们做不到的事（比如操作状态、注册定时器、侧输出）。

ProcessWindowFunction：

```java
sensorKS.window(TumblingEventTimeWindows.of(Time.seconds(10)))
    .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
        @Override
        public void process(String key, Context context, 
                           Iterable<WaterSensor> elements, Collector<String> out) {
            // 窗口内所有数据
            long count = elements.spliterator().estimateSize();
            
            // 获取窗口信息
            long start = context.window().getStart();
            long end = context.window().getEnd();
            
            out.collect("key=" + key + ", 有" + count + "条数据");
        }
    });
```

KeyedProcessFunction：

```java
sensorKS.process(new KeyedProcessFunction<String, WaterSensor, String>() {
    @Override
    public void processElement(WaterSensor value, Context ctx, Collector<String> out) {
        // 处理每条数据
    }
});
```

---

### 16. 侧输出流 (Side Output)

大白话：侧输出流就是"分流的秘密通道"，主流走主路，分流走小道。

举例：

• 正常水位数据走主流

• 异常水位数据（超过阈值）走侧输出，专门告警


代码示例：

```java
// 定义侧输出流标签
OutputTag<String> outputTag = new OutputTag<String>("side-output") {};

sensorDS.process(new ProcessWindowFunction<...>() {
    @Override
    public void process(...) {
        for (WaterSensor sensor : elements) {
            if (sensor.getVc() > 100) {
                // 超过100的，走侧输出
                ctx.output(outputTag, "告警：" + sensor);
            } else {
                // 正常的，走主流
                out.collect("正常：" + sensor);
            }
        }
    }
});

// 获取侧输出流
DataStream<String> sideOutput = process.getSideOutput(outputTag);
```

---

## 📁 项目目录结构

```
FlinkTutorial-1.17/
├── pom.xml                         # Maven配置文件
├── input/
│   └── word.txt                    # 测试用词频文件
├── src/main/java/com/atguigu/
│   ├── env/
│   │   └── EnvDemo.java            # 执行环境示例
│   ├── source/
│   │   ├── CollectionDemo.java     # 集合数据源
│   │   ├── KafkaSourceDemo.java    # Kafka数据源
│   │   ├── FileSourceDemo.java     # 文件数据源
│   │   └── DataGeneratorDemo.java  # 数据生成器
│   ├── transfrom/
│   │   ├── MapDemo.java            # Map转换
│   │   ├── FlatmapDemo.java        # FlatMap转换
│   │   ├── FilterDemo.java         # Filter过滤
│   │   └── RichFunctionDemo.java   # 富函数
│   ├── aggreagte/
│   │   ├── KeybyDemo.java          # KeyBy分组
│   │   ├── SimpleAggregateDemo.java # 简单聚合
│   │   └── ReduceDemo.java         # Reduce聚合
│   ├── window/
│   │   ├── WindowApiDemo.java      # 窗口API
│   │   ├── TimeWindowDemo.java     # 时间窗口
│   │   ├── CountWindowDemo.java    # 计数窗口
│   │   ├── WindowReduceDemo.java   # 窗口Reduce
│   │   ├── WindowAggregateDemo.java # 窗口Aggregate
│   │   └── WindowProcessDemo.java  # 窗口Process
│   ├── watermark/
│   │   ├── WatermarkOutOfOrdernessDemo.java # 乱序水印
│   │   ├── WatermarkMonoDemo.java  # 单调水印
│   │   ├── WatermarkLateDemo.java  # 迟到数据处理
│   │   ├── WatermarkIdlenessDemo.java # 水印空闲
│   │   ├── WatermarkCustomDemo.java # 自定义水印
│   │   ├── WindowJoinDemo.java     # 窗口Join
│   │   └── IntervalJoinDemo.java   # 区间Join
│   ├── state/
│   │   ├── KeyedValueStateDemo.java    # Keyed ValueState
│   │   ├── KeyedListStateDemo.java     # Keyed ListState
│   │   ├── KeyedMapStateDemo.java      # Keyed MapState
│   │   ├── KeyedReducingStateDemo.java # Keyed ReducingState
│   │   ├── KeyedAggregatingStateDemo.java # Keyed AggregatingState
│   │   ├── OperatorListStateDemo.java  # Operator ListState
│   │   ├── OperatorBroadcastStateDemo.java # BroadcastState
│   │   ├── StateBackendDemo.java       # 状态后端
│   │   └── StateTTLDemo.java           # 状态TTL
│   ├── checkpoint/
│   │   ├── CheckpointConfigDemo.java   # 检查点配置
│   │   ├── KafkaEOSDemo.java           # Kafka精确一次
│   │   ├── KafkaEOSDemo2.java          # Kafka精确一次2
│   │   └── SavepointDemo.java          # 保存点
│   ├── sink/
│   │   ├── SinkKafka.java              # Kafka Sink
│   │   ├── SinkFile.java               # 文件 Sink
│   │   ├── SinkMySQL.java              # MySQL Sink
│   │   ├── SinkCustom.java             # 自定义 Sink
│   │   └── SinkKafkaWithKey.java       # 带Key的Kafka Sink
│   ├── combine/
│   │   ├── ConnectDemo.java            # Connect合流
│   │   ├── ConnectKeybyDemo.java       # Connect + KeyBy
│   │   └── UnionDemo.java              # Union合流
│   ├── process/
│   │   ├── KeyedProcessTimerDemo.java  # 定时器
│   │   ├── KeyedProcessFunctionTopNDemo.java # TopN
│   │   ├── ProcessAllWindowTopNDemo.java   # 全窗口TopN
│   │   └── SideOutputDemo.java         # 侧输出流
│   ├── partition/
│   │   ├── PartitionDemo.java          # 分区策略
│   │   ├── PartitionCustomDemo.java   # 自定义分区
│   │   └── MyPartitioner.java         # 分区器实现
│   ├── split/
│   │   ├── SplitByFilterDemo.java      # Split+Filter
│   │   └── SideOutputDemo.java         # SideOutput
│   ├── sql/
│   │   ├── SqlDemo.java                # SQL示例
│   │   ├── TableStreamDemo.java        # Table API示例
│   │   ├── MyScalarFunctionDemo.java  # 自定义标量函数
│   │   ├── MyAggregateFunctionDemo.java # 自定义聚合函数
│   │   ├── MyTableFunctionDemo.java   # 自定义表函数
│   │   └── MyTableAggregateFunctionDemo.java # 自定义聚合表函数
│   ├── wc/
│   │   ├── WordCountStreamDemo.java    # 单词计数（流）
│   │   ├── WordCountStreamUnboundedDemo.java # 无界流单词计数
│   │   ├── WordCountBatchDemo.java     # 单词计数（批）
│   │   ├── OperatorChainDemo.java      # 算子链
│   │   └── SlotSharingGroupDemo.java   # 槽位共享组
│   └── bean/
│       └── WaterSensor.java            # 水位传感器实体类
```

---

## 🔧 技术栈

• Flink: 1.17.0

• Kafka: 消息队列

• MySQL: 关系型数据库

• RocksDB: 状态后端存储

• Hadoop: HDFS存储检查点


---

## 📖 学习路径建议

1. 入门阶段：执行环境 → 数据源 → 转换算子 → KeyBy → Sink
2. 核心阶段：窗口 → 水印 → 状态 → 检查点
3. 进阶阶段：定时器 → 双流Join → Table API → 性能优化

---

## ✅ 快速运行

```bash
# 启动Flink集群（如果用本地模式不需要）
# 运行WordCount示例
cd FlinkTutorial-1.17
mvn clean package
# 在IDE中直接运行 Java 类
```

---