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

3. Flink 编程模型：DataStream API 基础、Source、Transformation、Sink

好，咱们进入第三章。这一章是 Flink 开发的核心——DataStream API。说白了，你写 Flink 作业，大部分时间就是在跟这三个东西打交道：Source（数据从哪来）、Transformation（数据怎么变）、Sink（数据往哪去）。

我记得刚带团队做实时数仓那会儿，有个同事把 Source 和 Sink 的并行度配反了，结果数据全堵在算子链上，背压飙到 99%。排查了半天，最后发现是 API 理解不透彻。所以这一章，咱们把这三个环节彻底掰开揉碎。

3.1 DataStream API 的核心抽象

Flink 的 DataStream API，本质上就是一个有状态、无界、并行的数据流处理引擎。你想想看，你定义好 Source，然后连一串 Transformation，最后接到 Sink，Flink 就帮你把整个 DAG（有向无环图）跑起来。

核心就三个概念：

DataStream：代表一个并行流，数据在里面流动。
算子（Operator）：对流的操作，比如 map、filter、keyBy。
并行度（Parallelism）：每个算子可以有几个线程同时跑。

一句话总结：DataStream API 让你用声明式的方式描述“数据怎么流”，Flink 负责“怎么高效地流”。

3.2 Source：数据从哪来

Source 是数据流的起点。Flink 内置了很多 Source，但实际项目中，我用的最多的就三种：Kafka Source、文件 Source、自定义 Source。

3.2.1 内置 Source 快速上手

先看个最简单的——从 socket 读数据。这常用于测试：

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

但生产环境千万别这么干。我曾经见过有人把 socket 当生产 Source 用，结果网络抖动一次，整个作业重启，数据全丢了。

真正生产级的是 Kafka Source。Flink 1.12 之后推荐用新的 KafkaSource（基于 Source API V2）：

KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers("localhost:9092")
    .setTopics("input-topic")
    .setGroupId("my-group")
    .setStartingOffsets(OffsetsInitializer.earliest())
    .setValueOnlyDeserializer(SimpleStringSchema.class)
    .build();

DataStream<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

我的习惯：Kafka Source 的并行度最好和 Kafka 分区数保持一致。多了浪费资源，少了会加剧分区倾斜。

3.2.2 自定义 Source

当内置 Source 满足不了你时，就得自己写。比如对接 MQTT、Pulsar 或者某个老旧的消息中间件。

实现一个自定义 Source 有两种方式：

实现 SourceFunction 接口（旧 API，已标记为 @PublicEvolving，不推荐新项目用）
实现 Source 接口（新 API，Flink 1.12+ 推荐）

我拿新 API 举个例子。假设你要从某个 HTTP 接口轮询数据：

public class HttpSource implements Source<String, HttpSourceSplit, HttpSourceEnumState> {
    @Override
    public void run(SourceReaderContext ctx) throws Exception {
        // 轮询逻辑
        while (isRunning) {
            String data = httpClient.get("http://api.example.com/data");
            ctx.sendSourceEvent(data);
            Thread.sleep(1000);
        }
    }
}

避坑指南：我曾经写过一个自定义 Source，忘了在 checkpoint 时保存偏移量。结果作业重启后，数据从 0 开始重读，下游 Kafka 直接炸了。记住：自定义 Source 一定要实现 CheckpointedFunction 接口，否则 Exactly-Once 就是空话。

3.3 Transformation：数据怎么变

Transformation 是 DataStream API 的灵魂。Flink 提供了几十种算子，但常用的就十来个。我按使用频率排个序：

算子	用途	并行度影响
map	一对一转换	不变
flatMap	一对多转换	不变
filter	过滤数据	不变
keyBy	按 key 分区	变为 hash 分区
reduce	增量聚合	按 key 分组
window	开窗计算	按 key 分组
connect	连接两个流	取最大值

3.3.1 基础算子实战

先看 map 和 flatMap。map 是一对一，flatMap 是一对多。比如你从 Kafka 读到 JSON 字符串，想解析成 Java 对象：

DataStream<UserEvent> events = stream
    .map(json -> JsonUtil.parse(json, UserEvent.class));

flatMap 常用于拆分。比如一条日志里包含多条事件：

DataStream<Event> events = stream
    .flatMap((String line, Collector<Event> out) -> {
        for (String part : line.split("\\|")) {
            out.collect(parseEvent(part));
        }
    });

3.3.2 keyBy 与分区

keyBy 是 Flink 里最重要的算子之一。它把数据按 key 哈希到不同分区，后续的 window、reduce 才能按组计算。

DataStream<UserEvent> keyedStream = events
    .keyBy(UserEvent::getUserId);

这里有个坑：keyBy 之后的数据分布是哈希均匀的，但如果你 key 的基数很小（比如只有 0 和 1），就会产生数据倾斜。我在做用户行为分析时就遇到过，某个 key 的数据量是其他 key 的 100 倍，导致那个子任务一直跑，其他子任务闲着。

我的建议：如果 key 基数小，可以考虑加盐（salted key），比如 key + "_" + (random % N)，然后在 reduce 里再去掉盐。

3.3.3 窗口与聚合

窗口是实时计算的精髓。Flink 支持三种窗口：

Tumbling Window：固定大小，不重叠。比如每 5 分钟统计一次。
Sliding Window：固定大小，可重叠。比如每 1 分钟统计过去 5 分钟的数据。
Session Window：按不活动间隔划分。比如用户连续操作，超过 30 秒没动作就结束窗口。

来个实际例子：统计每个用户每 5 分钟的点击次数。

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

注意：窗口的触发时间取决于 watermark。如果你用 ProcessingTime，那窗口在机器时间到达时触发。如果你用 EventTime，那必须等 watermark 推进到窗口结束时间。我见过有人用 EventTime 但没设 watermark，结果窗口永远不触发。

3.4 Sink：数据往哪去

Sink 是数据流的终点。Flink 内置了 Kafka Sink、JDBC Sink、File Sink 等。但生产环境里，我几乎只用两种：Kafka Sink 和 自定义 Sink。

3.4.1 Kafka Sink

Kafka Sink 是最常用的，因为它天然支持 Exactly-Once 语义（配合两阶段提交）。

KafkaSink<String> sink = KafkaSink.<String>builder()
    .setBootstrapServers("localhost:9092")
    .setRecordSerializer(KafkaRecordSerializationSchema.builder()
        .setTopic("output-topic")
        .setValueSerializationSchema(new SimpleStringSchema())
        .build()
    )
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
    .build();

stream.sinkTo(sink);

关键点：setDeliveryGuarantee 有三个选项：NONE（至少一次）、AT_LEAST_ONCE（至少一次）、EXACTLY_ONCE（精确一次）。EXACTLY_ONCE 需要配合 checkpoint 和事务，性能会差一些，但数据一致性最高。

3.4.2 自定义 Sink

当你要写入 Redis、Elasticsearch 或者某个自研存储时，就得自己写 Sink。实现方式有两种：

实现 SinkFunction 接口（旧 API）
实现 Sink 接口（新 API，推荐）

我拿写入 Redis 举个例子。用新 API 实现一个简单的 Redis Sink：

public class RedisSink implements Sink<String, Void, Void> {
    private transient Jedis jedis;

    @Override
    public void open() {
        jedis = new Jedis("localhost", 6379);
    }

    @Override
    public void write(String value, Context ctx) {
        jedis.set("key", value);
    }

    @Override
    public void close() {
        if (jedis != null) jedis.close();
    }
}

我的经验：自定义 Sink 一定要处理好连接池和重试。我曾经有个 Sink 没做重试，Redis 集群抖动一次，整个 Flink 作业就挂了。后来加了指数退避重试，稳如老狗。

3.5 完整示例：从 Kafka 到 Kafka 的 ETL

把上面三个环节串起来，就是一个完整的实时 ETL 作业。假设我们要从 Kafka 读 JSON 日志，解析后过滤掉无效数据，再写回 Kafka：

public class KafkaToKafkaETL {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);

        // Source
        KafkaSource<String> source = KafkaSource.<String>builder()
            .setBootstrapServers("localhost:9092")
            .setTopics("input-topic")
            .setGroupId("etl-group")
            .setStartingOffsets(OffsetsInitializer.earliest())
            .setValueOnlyDeserializer(SimpleStringSchema.class)
            .build();

        DataStream<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

        // Transformation
        DataStream<String> result = stream
            .map(json -> JsonUtil.parse(json, LogEvent.class))
            .filter(event -> event.getLevel() != null)
            .map(event -> JsonUtil.toJson(event));

        // Sink
        KafkaSink<String> sink = KafkaSink.<String>builder()
            .setBootstrapServers("localhost:9092")
            .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                .setTopic("output-topic")
                .setValueSerializationSchema(new SimpleStringSchema())
                .build()
            )
            .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
            .build();

        result.sinkTo(sink);

        env.execute("Kafka to Kafka ETL");
    }
}

这个例子虽然简单，但涵盖了 DataStream API 的完整流程。你想想看，Source 读数据，Transformation 处理数据，Sink 写数据——Flink 帮你把这三件事串成一个有状态的 DAG，自动容错、自动恢复。

3.6 本章知识体系

下面这张图把 DataStream API 的核心流程画出来了。Source、Transformation、Sink 三个环节，加上并行度和 checkpoint，构成了 Flink 实时作业的骨架。

嗯，这一章的内容就到这。Source、Transformation、Sink 这三个环节，是 Flink 开发的基石。你只要把这三块搞透了，后面学 ProcessFunction、状态后端、容错机制都会轻松很多。