3. Flink核心概念：DataStream API、Table API、State、Checkpoint、Savepoint

好，咱们进入第三章。这一章我打算把Flink的几个核心概念掰开揉碎讲清楚。说实话，很多初学者一上来就被这些术语搞懵了——DataStream API和Table API到底啥关系？State和Checkpoint是不是一回事？Savepoint又是个啥？

我当年刚接触Flink时也踩过这些坑。有一次线上作业挂了，恢复后发现数据对不上，折腾了一整天才明白是Checkpoint配置和State后端没搞对。嗯，咱们今天就把这些概念彻底理清。

核心要点：DataStream API是底层能力，Table API是上层抽象；State是计算的心脏，Checkpoint是容错的基石，Savepoint是运维的瑞士军刀。

Flink核心概念关系图 Flink Runtime Table API / SQL DataStream API ProcessFunction State（状态存储） Checkpoint（快照） Savepoint（手动快照） 说明： • DataStream API 提供细粒度控制，适合复杂逻辑 • Table API 声明式编程，开发效率高，适合批流统一 • State 是计算中间结果，Checkpoint 是自动定期备份 • Savepoint 是手动触发的快照，用于版本升级/迁移

3.1 DataStream API：流计算的"汇编语言"

DataStream API是Flink最底层的编程接口。说白了，它让你能直接操作数据流中的每一条记录。我个人的习惯是：如果业务逻辑复杂、需要精细控制状态和时机，就用DataStream API。

来看一个最简单的例子——从Kafka读取数据并做过滤：

// 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 从Kafka读取数据流
DataStream<String> sourceStream = env.addSource(
    new FlinkKafkaConsumer<>("topic", new SimpleStringSchema(), properties)
);

// 转换操作：过滤出包含"error"的日志
DataStream<String> errorStream = sourceStream
    .filter(line -> line.contains("error"))
    .map(line -> line.toUpperCase());

// 输出到控制台
errorStream.print();

这段代码看着简单，但背后Flink帮你做了很多事：数据分区、并行执行、容错恢复。DataStream API的核心操作可以分为三类：

  • Source（数据源）：从哪里读数据？Kafka、文件、Socket、自定义Source
  • Transformation（转换）：怎么处理数据？map、filter、flatMap、keyBy、window
  • Sink（输出）：结果写到哪里？Kafka、Elasticsearch、文件、数据库

我的经验：DataStream API虽然灵活，但代码量会比较大。如果业务逻辑能用SQL表达，我建议优先考虑Table API。别为了炫技而用底层API，开发效率也是成本。

3.2 Table API：让批流统一成为现实

Table API是Flink提供的关系型API。你写SQL或者Table API的链式调用，Flink自动帮你优化执行计划。我记得第一次用Table API做流计算时，最大的感受就是——终于不用手写窗口逻辑了！

同样的过滤逻辑，用Table API写是这样的：

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

// 将DataStream注册为表
tableEnv.createTemporaryView("log_table", sourceStream, $("line"));

// 用SQL查询
Table resultTable = tableEnv.sqlQuery(
    "SELECT UPPER(line) FROM log_table WHERE line LIKE '%error%'"
);

// 将结果转回DataStream并输出
tableEnv.toDataStream(resultTable, Row.class).print();

你发现没？代码量少了一半。而且Table API有个杀手锏——同样的SQL，在批模式和流模式下都能跑。这就是流批一体的核心价值。

我给大家整理了一个对比表格，方便理解两者的差异：

对比维度 DataStream API Table API / SQL
编程范式 命令式（怎么做的细节都要写） 声明式（告诉我要什么结果）
开发效率 低，代码量大 高，SQL一行顶十行
灵活性 极高，能控制每条记录 中等，受限于SQL表达能力
性能优化 手动优化 优化器自动优化
批流统一 需要写两套逻辑 一套SQL跑批和流
适用场景 复杂ETL、自定义窗口、状态密集型计算 报表统计、简单过滤、批处理作业

注意：Table API虽然方便，但有些场景它搞不定。比如你需要精确控制每条数据的处理时机，或者要访问外部存储做维表关联，这时候还得回到DataStream API。我建议两者结合使用——Table API做主体逻辑，DataStream API做"脏活累活"。

3.3 State：流计算的"记忆"

State是Flink最核心的概念之一。为什么？因为流计算是无界数据，每条数据来了处理完就走了。但很多场景需要"记住"之前的数据——比如统计过去5分钟的点击量、计算用户累计消费金额。

State就是Flink用来"记住"这些东西的机制。它分为几种类型：

  • ValueState<T>：存单个值，比如用户当前积分
  • ListState<T>：存列表，比如用户最近10条操作记录
  • MapState<UK, UV>：存键值对，比如用户各品类的购买次数
  • ReducingState<T>：存聚合结果，自动用ReduceFunction合并
  • AggregatingState<IN, OUT>：更灵活的聚合，可以改变输出类型

来看一个实际例子——统计每个用户的点击次数：

public static class ClickCountFunc extends RichFlatMapFunction<ClickEvent, Tuple2<String, Long>> {
    private ValueState<Long> countState;

    @Override
    public void open(Configuration parameters) {
        ValueStateDescriptor<Long> descriptor = 
            new ValueStateDescriptor<>("clickCount", Types.LONG);
        countState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void flatMap(ClickEvent event, Collector<Tuple2<String, Long>> out) throws Exception {
        Long currentCount = countState.value();
        if (currentCount == null) {
            currentCount = 0L;
        }
        currentCount += 1;
        countState.update(currentCount);
        out.collect(new Tuple2<>(event.userId, currentCount));
    }
}

这段代码里，countState就是每个用户的状态。Flink自动帮你管理状态的创建、更新和清理。你想想看，如果没有State，你要自己维护一个全局Map，还得考虑并发和容错，多麻烦。

避坑指南：State不是无限大的！我曾经有个项目，状态数据量太大，直接把堆内存撑爆了。后来改用RocksDB状态后端才解决。记住：ValueState适合小数据量，MapState适合中等规模，超大状态一定要用RocksDB。

3.4 Checkpoint：自动容错的"保险丝"

Checkpoint是Flink实现容错的机制。它定期把作业的状态拍个快照，保存到持久化存储（比如HDFS）。如果作业挂了，Flink从最近的Checkpoint恢复，保证数据不丢、不重。

Checkpoint的工作原理可以概括为三个步骤：

  1. Barrier对齐：Source算子插入Checkpoint Barrier，像水流中的标记线
  2. 状态快照：每个算子收到Barrier后，把当前状态写入状态后端
  3. 确认完成：所有算子完成快照后，JobManager标记Checkpoint成功

配置Checkpoint的代码很简单：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 开启Checkpoint，间隔5秒
env.enableCheckpointing(5000);

// 设置模式：EXACTLY_ONCE（精确一次）或 AT_LEAST_ONCE（至少一次）
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// Checkpoint超时时间
env.getCheckpointConfig().setCheckpointTimeout(60000);

// 同时进行的Checkpoint数量
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

// 两次Checkpoint的最小间隔
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);

我的建议：生产环境一定要开启Checkpoint！我见过太多线上事故，就是因为没开Checkpoint，作业挂了只能从头消费数据。另外，Checkpoint间隔别设太短，否则频繁快照会影响性能。一般5-10秒是比较合理的值。

3.5 Savepoint：运维的"时光机"

Savepoint和Checkpoint本质上是一样的——都是状态快照。但有几个关键区别：

对比维度 Checkpoint Savepoint
触发方式 自动，周期性触发 手动，用户主动触发
生命周期 作业停止后自动清理 手动管理，长期保留
主要用途 故障恢复 版本升级、作业迁移、A/B测试
存储格式 内部优化格式 标准格式，可跨版本兼容
触发命令 代码配置 flink savepoint 命令

Savepoint最常见的场景是作业升级。比如你要修改业务逻辑，或者升级Flink版本，先停掉旧作业，用Savepoint启动新作业：

# 触发Savepoint
flink savepoint <jobId> /tmp/savepoints

# 从Savepoint启动新作业
flink run -s /tmp/savepoints/savepoint-xxxxx \
  -c com.example.MyJob \
  my-flink-job.jar

曾经踩过的坑：有一次我升级作业，忘了Savepoint中状态的UID和算子ID的对应关系。结果新作业启动后，状态对不上，数据全乱了。后来我养成了一个习惯——每个算子都显式指定uid("my-operator")，这样Savepoint恢复时就不会出问题。

3.6 四个概念如何协同工作？

最后，咱们把这四个概念串起来。一个典型的Flink作业运行流程是这样的：

  • 你用DataStream API或Table API编写业务逻辑
  • Flink在运行时维护State来存储中间结果
  • Checkpoint定期自动备份State，保证故障恢复
  • 需要升级或迁移时，用Savepoint手动保存状态快照

说白了，API是"怎么算"，State是"算什么"，Checkpoint和Savepoint是"算错了怎么办"。这四个概念构成了Flink流计算的核心骨架。理解了它们，你就掌握了Flink的精髓。

← 上一章 📖 返回目录 下一章 →