Flink状态管理与容错
请介绍Flink的状态管理机制，包括状态类型（Keyed State/Operator State）、状态后端（State Backend）以及如何保证Exactly-Once。
回答
小字辈
Flink状态类型：
1. Keyed State（键控状态）：
- 每个Key绑定一份状态，在KeyBy之后使用
- 状态类型：
ValueState<T>：单值状态ListState<T>：列表状态MapState<UK, UV>：Map状态ReducingState<T>：聚合状态AggregatingState<IN, OUT>：聚合状态
- TTL支持：
StateTtlConfig可配置状态的过期时间
2. Operator State（算子状态）：
- 绑定到算子的每个并行实例
- 状态类型：
ListState<T>：列表状态BroadcastState<K, V>：广播状态（所有实例共享）UnionListState<T>：可合并的列表状态
状态后端（State Backend）：
| 状态后端 | 存储位置 | 适用场景 |
|---|---|---|
| HashMapStateBackend（原MemoryStateBackend） | JVM堆内存 | 小状态（<1GB），开发测试 |
| EmbeddedRocksDBStateBackend | RocksDB（本地磁盘+内存Cache） | 大状态（TB级别），需要增量Checkpoint |
RocksDBStateBackend特点：
- 使用LSM-Tree结构的RocksDB存储状态
- 状态超过内存时自动Spill到磁盘
- 支持增量Checkpoint（只写变更部分）
- 适合超大规模状态作业
Exactly-Once实现机制（两阶段提交 Sink）：
- Checkpoint Barrier 在算子间传递
- 所有Operator完成快照后，JobManager发起Checkpoint完成通知
- 实现了
TwoPhaseCommitSinkFunction的Sink：- 预提交（Pre-Commit）： Checkpoint时预提交事务
- 提交（Commit）： Checkpoint完成后正式提交
- 常用支持Exactly-Once的Sink：Kafka（事务）、HBase、JDBC
配置容错：
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setStateBackend(new HashMapStateBackend());
env.getCheckpointConfig().setCheckpointStorage("hdfs:///checkpoints");
env.enableCheckpointing(10000);
env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);