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

3、Flink编程模型：DataStream API概览、Source算子、Transformation算子、Sink算子、Flink程序骨架

好，咱们进入第三章。这一章是整个课程的基石——Flink的编程模型。说白了，就是告诉你Flink程序到底长什么样，怎么把数据读进来、怎么处理、再写出去。

我刚开始学Flink的时候，最困惑的就是：这玩意儿到底怎么跑起来的？后来做多了才发现，Flink的程序结构其实非常固定，就像搭积木一样。你只要掌握了这几块积木，剩下的就是组合问题了。

3.1 DataStream API 概览

DataStream API，是Flink处理无界数据流的核心API。它把数据流抽象成一个叫 DataStream<T> 的对象，所有操作都围绕这个对象展开。

我个人习惯把DataStream API分成三个层次：

Source：数据从哪来？Kafka、文件、Socket，或者自定义。
Transformation：数据怎么变？map、filter、keyBy、window……
Sink：数据去哪？写到Kafka、MySQL、HDFS，或者控制台。

你想想看，这不就是典型的「输入-处理-输出」模型吗？只不过Flink把它做成了流式版本，数据是一条一条进来的，处理也是一条一条做的。

核心要点：DataStream API 是 Flink 流处理的「第一公民」。你写的每一个算子，最终都会被翻译成 Task，在集群上并行执行。

下面这张图，是我每次给团队新人讲Flink时必画的。它把整个编程模型串起来了：

3.2 Source 算子：数据从哪来？

Source 是数据流的起点。Flink 内置了非常丰富的 Source 实现，但最常用的就那几种。

3.2.1 内置 Source

Source 类型	使用场景	示例代码
`fromElements`	测试、Demo	`env.fromElements(1, 2, 3)`
`fromCollection`	测试、小数据量	`env.fromCollection(list)`
`socketTextStream`	调试、快速验证	`env.socketTextStream("localhost", 9999)`
`readTextFile`	批处理、离线	`env.readTextFile("path/to/file")`
Kafka Source	生产环境主流	`new FlinkKafkaConsumer<>(...)`

我的建议：开发阶段用 socketTextStream 调试最方便。但上了生产，99% 的场景都是用 Kafka Source。我曾经见过有人把 Socket Source 直接部署到线上，结果连接一断整个作业就挂了……嗯，血的教训。

3.2.2 自定义 Source

当内置 Source 满足不了需求时，你可以实现 SourceFunction 接口。比如从某个 RPC 接口拉数据：

public class MySource implements SourceFunction<String> {
    private volatile boolean running = true;

    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        while (running) {
            // 模拟从外部系统拉数据
            String data = pullFromRpc();
            ctx.collect(data);
            Thread.sleep(1000);
        }
    }

    @Override
    public void cancel() {
        running = false;
    }
}

注意：自定义 Source 里一定要实现 cancel() 方法，否则作业取消时资源无法释放。我踩过这个坑，线上作业取消后 Source 线程还在跑，导致 ZooKeeper 会话一直不释放。

3.3 Transformation 算子：数据怎么变？

Transformation 是 Flink 最核心的部分。数据进来之后，经过一系列算子转换，变成你想要的格式。

3.3.1 单数据流算子

这些算子只操作一条数据流，不涉及合并或拆分：

map：一条进，一条出。最常用的算子。
flatMap：一条进，零到多条出。适合拆分逻辑。
filter：过滤掉不符合条件的数据。
keyBy：按 key 分组，后续才能做聚合或窗口。

举个例子，从用户行为日志里提取出「购买」事件：

DataStream<String> source = env.socketTextStream("localhost", 9999);

DataStream<UserAction> buyActions = source
    .map(line -> {
        String[] fields = line.split(",");
        return new UserAction(fields[0], fields[1], Long.parseLong(fields[2]));
    })
    .filter(action -> "buy".equals(action.getActionType()))
    .keyBy(UserAction::getUserId);

3.3.2 多数据流算子

有时候需要把多条流合并，或者把一条流拆开：

算子	作用	典型场景
`union`	合并多条同类型流	合并多个 Kafka Topic 的数据
`connect` + `CoMap`	连接两条不同类型流	将订单流和支付流关联
`split` / `select`	按条件拆分一条流	将日志分为 error、warn、info

经验之谈：union 要求所有流的数据类型必须一致。而 connect 可以连接不同类型，但需要额外写 CoMapFunction 来处理。我个人更推荐 connect，因为它更灵活。

3.4 Sink 算子：数据写去哪？

Sink 是数据流的终点。数据经过 Transformation 处理后，最终要写到某个存储系统里。

3.4.1 常用 Sink

print()：调试用，直接打印到控制台。
writeAsText()：写到文件，适合小数据量。
Kafka Sink：生产环境最常用，通过 FlinkKafkaProducer 实现。
JDBC Sink：写到关系型数据库，注意要开启批量写入。
自定义 Sink：实现 SinkFunction 接口。

3.4.2 写 Sink 的注意事项

写 Sink 时，最容易出问题的是「一致性」和「性能」。我分享两个经验：

开启 Checkpointing：只有开启了 Checkpoint，Sink 才能实现 exactly-once 语义。
批量写入：不要一条一条地写数据库，用 buffer 攒一批再写。我见过有人没做批量，结果每秒几万条数据直接把 MySQL 打挂了。

// 一个带缓冲的 JDBC Sink 示例
public class BufferedJdbcSink extends RichSinkFunction<UserAction> {
    private List<UserAction> buffer = new ArrayList<>();
    private static final int BATCH_SIZE = 1000;

    @Override
    public void invoke(UserAction value, Context ctx) {
        buffer.add(value);
        if (buffer.size() >= BATCH_SIZE) {
            flush();
        }
    }

    private void flush() {
        // 批量写入数据库
        // ...
        buffer.clear();
    }
}

3.5 Flink 程序骨架：四步走

好了，Source、Transformation、Sink 都讲完了。现在把它们拼起来，就是一个完整的 Flink 程序。我总结了一个「四步走」骨架：

public class FlinkJobSkeleton {
    public static void main(String[] args) throws Exception {
        // 第一步：获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 第二步：添加 Source
        DataStream<String> source = env.socketTextStream("localhost", 9999);

        // 第三步：定义 Transformation
        DataStream<UserAction> processed = source
            .map(new MyMapFunction())
            .filter(new MyFilterFunction())
            .keyBy(UserAction::getUserId);

        // 第四步：指定 Sink 并触发执行
        processed.print();
        env.execute("Flink Job Skeleton");
    }
}

记住：env.execute() 是必须的。没有这行代码，Flink 不会启动任何任务。我刚开始学的时候经常忘记写，然后盯着控制台发呆……

这个骨架适用于 90% 的 Flink 作业。你只需要替换 Source、Transformation 和 Sink 的具体实现，就能应对各种场景。

3.6 本章小结

这一章我们走通了 Flink 编程模型的完整链路：

Source 负责读数据，内置的够用就用，不够就自定义。
Transformation 负责处理数据，单流用 map/filter，多流用 union/connect。
Sink 负责写数据，注意批量写入和 Checkpoint。
程序骨架就是「环境 → Source → Transformation → Sink → 执行」这四步。