📚 课程目录第4章 Flink DataStream API

4. Flink DataStream API：Source与Sink算子、Transformation算子详解、KeyBy与Window操作

这一章，我们正式进入 Flink DataStream API 的核心地带。说实话，很多初学者一开始就被各种算子搞晕了。Source、Sink、Transformation、KeyBy、Window……名字一个比一个长。但你别怕，这些东西说白了就是「数据从哪来、怎么变、去哪存」。我带你捋一遍，保证你心里有数。

  本章核心脉络： 数据接入（Source）→ 数据转换（Transformation）→ 数据分组（KeyBy）→ 窗口计算（Window）→ 数据输出（Sink）。一条线串起整个流处理管道。

4.1 Source 算子：数据从哪来

Source 是管道的起点。我见过不少团队，数据源还没搞清楚就开始写业务逻辑，结果跑起来才发现数据格式不对。Source 选错了，后面全白搭。

Flink 内置了丰富的 Source。常用的有这几类：

基于集合的 Source：fromCollection、fromElements，适合本地测试。
基于文件的 Source：readTextFile，支持 HDFS、本地文件。
基于 Socket 的 Source：socketTextStream，调试时用得多。
基于消息队列的 Source：KafkaSource、PulsarSource，生产环境标配。
自定义 Source：实现 SourceFunction 或 RichSourceFunction。

我的经验： 生产环境 90% 的场景用 Kafka Source。我个人习惯在开发阶段先用 fromElements 模拟数据，等逻辑调通了再切到 Kafka。这样能省不少调试时间。

看个例子，从 Kafka 读取数据：

DataStream<String> stream = env.fromSource(
    KafkaSource.<String>builder()
        .setBootstrapServers("localhost:9092")
        .setTopics("input-topic")
        .setGroupId("my-group")
        .setStartingOffsets(OffsetsInitializer.latest())
        .build(),
    WatermarkStrategy.noWatermarks(),
    "Kafka Source"
);

4.2 Transformation 算子：数据怎么变

Transformation 是 DataStream API 的灵魂。说白了，就是把一条数据流变成另一条数据流。我刚开始学的时候，总觉得算子太多记不住。后来发现，核心就那几个，其他的都是变种。

4.2.1 基础转换算子

算子	作用	输入 → 输出	典型场景
`map`	一对一转换	1条 → 1条	字段提取、格式转换
`flatMap`	一对多转换	1条 → 0~N条	分词、拆分事件
`filter`	过滤	1条 → 0或1条	数据清洗、异常剔除
`keyBy`	按 key 分组	DataStream → KeyedStream	分组聚合

举个例子，从日志中提取错误信息：

DataStream<String> errorStream = logStream
    .filter(line -> line.contains("ERROR"))
    .map(line -> {
        String[] parts = line.split("\\|");
        return parts[1]; // 提取错误描述
    });

注意： 我曾经在项目中遇到过一个坑——flatMap 里忘记调用 out.collect()，结果数据全丢了。检查了半天才发现。记住，flatMap 必须显式调用 out.collect() 才能输出数据。

4.2.2 多流转换算子

实际业务中，经常需要合并或拆分多条流。常用的有：

union：合并多条同类型流。
connect：连接两条不同类型流，配合 CoMapFunction 使用。
split / select：拆分一条流为多条（已标记为废弃，推荐用 Side Output）。

我个人更推荐用 Side Output 替代 split，更灵活，类型安全也更好。

4.3 KeyBy 操作：分组是窗口的前提

KeyBy 是 Flink 里一个非常重要的概念。它把数据按照某个 key 进行分区，相同 key 的数据会进入同一个 subtask。你想想看，如果没有 KeyBy，Window 就不知道要对谁做聚合。

KeyBy 的底层原理是哈希分区。它会计算 key 的 hash 值，然后对并行度取模，决定数据去哪个分区。

DataStream<Order> orderStream = ...;
KeyedStream<Order, String> keyedStream = orderStream
    .keyBy(order -> order.getUserId());

  关键点： KeyBy 之后的数据类型变成了 KeyedStream。只有 KeyedStream 才能调用 Window 算子。这是 Flink 的设计约束，也是很多新手容易忽略的地方。

4.4 Window 操作：时间窗口的威力

Window 是流处理中「批」思想的体现。流数据是无限的，但业务往往需要统计「过去 5 分钟」或「最近 100 条」的数据。Window 就是干这个的。

4.4.1 窗口类型

窗口类型	说明	适用场景
Tumbling Window	滚动窗口，不重叠	每分钟统计一次 PV
Sliding Window	滑动窗口，有重叠	每 5 秒统计过去 1 分钟的平均值
Session Window	会话窗口，按不活动间隔划分	用户行为会话分析
Global Window	全局窗口，需配合 Trigger	自定义窗口逻辑

看一个滚动窗口的例子，统计每 5 秒的用户点击量：

DataStream<ClickEvent> clicks = ...;
clicks
    .keyBy(ClickEvent::getUserId)
    .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
    .aggregate(new AggregateFunction<ClickEvent, Long, Long>() {
        @Override
        public Long createAccumulator() { return 0L; }
        @Override
        public Long add(ClickEvent 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; }
    });

避坑指南： 我曾经在生产环境遇到过窗口数据延迟的问题。原因是默认的 ProcessingTime 不关心事件实际发生时间。如果你的数据有乱序，一定要用 EventTime + Watermark。否则统计结果会偏差很大。

4.4.2 时间语义与 Watermark

Flink 支持三种时间语义：

ProcessingTime：处理时间，简单但不准。
EventTime：事件时间，准确但需要 Watermark。
IngestionTime：摄入时间，介于两者之间。

Watermark 是 Flink 解决乱序问题的核心机制。它表示「在这个时间戳之前的数据都已经到达了」。举个例子，如果 Watermark 是 10:00:05，那么 10:00:00 这个窗口就可以触发了。

DataStream<Event> stream = env
    .fromSource(source, WatermarkStrategy
        .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
        .withTimestampAssigner((event, timestamp) -> event.getTimestamp()),
        "Event Source");

4.5 Sink 算子：数据去哪存

Sink 是管道的终点。数据经过 Source、Transformation、KeyBy、Window 之后，最终要落到某个存储里。

常用的 Sink 包括：

Kafka Sink：写回 Kafka，用于下游消费。
JDBC Sink：写入 MySQL、PostgreSQL 等关系型数据库。
File Sink：写入 HDFS、S3 等文件系统。
Elasticsearch Sink：写入 ES 用于搜索和分析。
自定义 Sink：实现 SinkFunction。

看一个写入 Kafka 的例子：

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

注意： Sink 的 DeliveryGuarantee 要谨慎选择。EXACTLY_ONCE 需要两阶段提交，性能开销大。如果业务允许少量重复，用 AT_LEAST_ONCE 就够了。我见过有人为了「精确一次」把吞吐量降了一半，其实没必要。

4.6 小结

这一章我们走完了 Flink DataStream API 的完整链路。Source 负责接入，Transformation 负责转换，KeyBy 负责分组，Window 负责窗口计算，Sink 负责输出。每个环节都有它的设计哲学和最佳实践。