📚 课程目录第4章 Flink 编程模型

4. Flink 编程模型：DataStream API 核心算子与 SQL 入门

好，我们进入第四章。这一章是 Flink 的实操核心——DataStream API 和 Table API。说白了，你写 Flink 作业，大部分时间就是在跟这些东西打交道。我个人习惯把这一章看作是「从代码到逻辑」的桥梁，你理解了这些算子，才算真正入了实时流处理的门。

4.1 DataStream API 核心算子：从数据流到业务逻辑

DataStream API 是 Flink 最底层的编程接口。它给你提供了最细粒度的控制能力。你想想看，一个实时数据流进来，你要做过滤、转换、分组、聚合，这些操作在 Flink 里都对应着具体的算子。

4.1.1 map：一对一转换

map 是最简单的算子。输入一条数据，输出一条数据。我在项目中经常用它来做数据清洗，比如把 JSON 字符串解析成 Java 对象。

DataStream<String> rawStream = env.socketTextStream("localhost", 9999);
DataStream<EventLog> parsedStream = rawStream.map(new MapFunction<String, EventLog>() {
    @Override
    public EventLog map(String value) throws Exception {
        // 解析 JSON，返回对象
        return JsonUtil.parse(value, EventLog.class);
    }
});

嗯，这里要注意：map 函数里不要做耗时操作，比如数据库查询。因为它是单条处理的，会阻塞整个算子。

4.1.2 filter：数据过滤

filter 用来剔除不符合条件的数据。我曾经在一个日志清洗作业里，用 filter 过滤掉了所有状态码为 200 的请求，只保留异常日志，这样下游的告警系统压力小了很多。

DataStream<EventLog> errorStream = parsedStream.filter(new FilterFunction<EventLog>() {
    @Override
    public boolean filter(EventLog value) throws Exception {
        return value.getStatus() >= 400;
    }
});

4.1.3 keyBy：数据分组

keyBy 是 Flink 里最关键的算子之一。它把数据按照某个 key 进行分区，相同 key 的数据会进入同一个 subtask。说白了，这是后续做窗口聚合的前提。

避坑指南：我曾经犯过一个错误——用了一个自定义对象作为 keyBy 的 key，结果没有正确重写 hashCode() 和 equals() 方法。导致相同业务含义的数据被分到了不同的分区，窗口聚合结果完全不对。记住：keyBy 的 key 必须是一个可序列化的、有正确 hashCode 实现的对象。

DataStream<EventLog> keyedStream = parsedStream
    .keyBy(EventLog::getUserId);

4.1.4 window：窗口计算

窗口是流处理的核心概念。无界流怎么聚合？靠窗口。Flink 支持多种窗口类型：

窗口类型	说明	适用场景
Tumbling Window	滚动窗口，固定大小，不重叠	每分钟统计 PV
Sliding Window	滑动窗口，固定大小，可重叠	每 5 秒统计过去 1 分钟的 UV
Session Window	会话窗口，基于不活动间隔	用户行为 session 分析
Global Window	全局窗口，需要自定义触发器	特殊业务逻辑

举个例子，统计每个用户每分钟的点击次数：

DataStream<Tuple2<String, Long>> result = keyedStream
    .window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
    .aggregate(new AggregateFunction<EventLog, Long, Long>() {
        @Override
        public Long createAccumulator() { return 0L; }
        @Override
        public Long add(EventLog value, Long accumulator) { return accumulator + 1; }
        @Override
        public Long getResult(Long accumulator) { return accumulator; }
        @Override
        public Long merge(Long a, Long b) { return a + b; }
    });

个人经验：窗口大小和滑动步长的选择，直接影响延迟和吞吐。我一般建议：如果业务对延迟敏感，用滑动窗口；如果对准确性要求高，用滚动窗口加迟到数据处理。

4.2 Flink SQL 入门：用 SQL 写流处理

说实话，DataStream API 虽然灵活，但写起来还是有点繁琐。尤其是团队里数据分析师也想参与实时计算时，让他们写 Java 代码不太现实。这时候，Flink SQL 就派上用场了。

Flink SQL 把流当作一张「动态表」。你写 SQL 查询这张表，Flink 自动把它翻译成 DataStream 作业。我刚开始接触时也觉得神奇——原来 SQL 还能跑在流上。

4.2.1 创建表环境

第一步，创建 TableEnvironment。它负责管理表的元数据和执行 SQL。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

4.2.2 注册表

你可以从 DataStream 注册表，也可以直接从 Kafka 等外部系统注册。

// 从 DataStream 注册
DataStream<EventLog> eventStream = ...;
tableEnv.createTemporaryView("event_log", eventStream, 
    $("user_id"), $("event_type"), $("event_time"), $("status"));

// 或者直接从 Kafka 注册
tableEnv.executeSql("CREATE TABLE kafka_source (\n" +
    "  user_id STRING,\n" +
    "  event_type STRING,\n" +
    "  event_time TIMESTAMP(3),\n" +
    "  status INT\n" +
    ") WITH (\n" +
    "  'connector' = 'kafka',\n" +
    "  'topic' = 'user_events',\n" +
    "  'properties.bootstrap.servers' = 'localhost:9092',\n" +
    "  'format' = 'json'\n" +
    ")");

4.2.3 执行 SQL 查询

注册好表之后，就可以写 SQL 了。比如统计每分钟每个用户的点击次数：

Table resultTable = tableEnv.sqlQuery(
    "SELECT \n" +
    "  user_id,\n" +
    "  TUMBLE_START(event_time, INTERVAL '1' MINUTE) AS window_start,\n" +
    "  COUNT(*) AS cnt\n" +
    "FROM event_log\n" +
    "GROUP BY \n" +
    "  user_id,\n" +
    "  TUMBLE(event_time, INTERVAL '1' MINUTE)"
);

你看，这段 SQL 和传统 SQL 几乎一样，只是多了个 TUMBLE 窗口函数。Flink 会自动把它翻译成 DataStream 的 window 操作。

4.3 Table & SQL API 概览：统一批流

Table API 是介于 DataStream API 和 SQL 之间的一种编程接口。它用 Java/Scala 的链式调用表达查询逻辑，比 SQL 更灵活，比 DataStream API 更简洁。

我个人觉得，Table API 最大的价值在于「统一批流」。同一套逻辑，既可以跑在流上，也可以跑在批上。你想想看，这给开发和测试带来了多大的便利。

4.3.1 Table API 示例

Table resultTable = tableEnv.from("event_log")
    .groupBy($("user_id"), $("event_time").window(Tumble.over(lit(1).minutes()).on($("event_time")).as("w")))
    .select($("user_id"), $("w").start().as("window_start"), $("event_type").count().as("cnt"));

这段代码和上面的 SQL 是等价的。选择哪种方式，取决于你的团队习惯。我个人偏好：简单查询用 SQL，复杂逻辑用 Table API，需要细粒度控制时用 DataStream API。

4.3.2 三种 API 的对比

API 类型	表达能力	易用性	性能	适用人群
DataStream API	最强	较低	最高	资深开发
Table API	中等	中等	高	开发/数据工程师
SQL	有限	最高	高	分析师/开发

核心观点：三种 API 可以混合使用。比如用 SQL 做大部分 ETL，用 DataStream API 处理特殊逻辑，最后再用 SQL 输出结果。Flink 的架构设计允许你在同一个作业里自由切换。

4.4 本章知识体系图

下面这张图展示了 DataStream API、Table API 和 SQL 之间的关系，以及核心算子的数据流转路径。

4.5 小结

这一章我们覆盖了 Flink 编程模型的三个层次：

DataStream API：map、filter、keyBy、window 四个核心算子，是流处理的基础。
Flink SQL：用 SQL 写流处理，降低了实时计算的门槛。
Table API：介于两者之间，兼顾灵活性和易用性。