3. Flink DataStream API 入门：Source、Transformation、Sink

好，咱们进入第三章。前面聊了那么多流批一体的概念和架构，今天得动真格的了——上手写代码。

Flink DataStream API 是实时特征计算的基石。说白了，你后面所有的特征逻辑，都得靠它来表达。我个人习惯把 DataStream 程序拆成三块来看：数据从哪来（Source）、数据怎么变（Transformation）、数据去哪了（Sink）。这三块搞明白了，Flink 编程的基本功就算打牢了。

3.1 Source：数据从哪来

Source 是 Flink 作业的入口。你想想看，没有数据进来，后面再牛的计算逻辑也是白搭。

Flink 内置了非常丰富的 Source 实现。我挑几个最常用的给你说说：

Source 类型	适用场景	并行度	常用方式
基于文件	批处理、测试数据	可配置	readTextFile(path)
基于 Socket	调试、演示	1	socketTextStream(host, port)
基于集合	单元测试、本地调试	1	fromCollection(list)
Kafka	生产环境主流	与分区数相关	FlinkKafkaConsumer
自定义	特殊数据源	自定义	实现 SourceFunction

我在项目中遇到过最典型的场景：线上实时特征计算，Source 几乎清一色是 Kafka。为什么呢？因为 Kafka 能扛高吞吐，还能回溯消费，配合 Flink 的 Checkpoint 机制，能做到 exactly-once 语义。

来看一段最简单的 Source 代码：

// 从集合创建 DataStream
DataStream<String> textStream = env.fromCollection(
    Arrays.asList("hello flink", "hello world", "flink streaming")
);

// 从 Socket 读取（用于调试）
DataStream<String> socketStream = env.socketTextStream("localhost", 9999);

// 从文件读取
DataStream<String> fileStream = env.readTextFile("path/to/input.txt");

3.2 Transformation：数据怎么变

Transformation 是 DataStream API 的灵魂。所有的业务逻辑、特征计算，都在这一层完成。

Flink 提供了几十种 Transformation 算子。但说实话，日常开发中 80% 的场景，只需要掌握下面这几个：

• map：一对一转换，输入一条输出一条。最常用的算子。
• flatMap：一对多转换，输入一条可能输出零到多条。适合做分词、展开。
• filter：过滤，保留满足条件的数据。
• keyBy：按 key 分组，后续的聚合操作必须基于 keyBy。
• sum / reduce / aggregate：聚合操作，做累加、归约、自定义聚合。
• window：窗口操作，流处理中做时间窗口聚合的核心。

咱们用 WordCount 的 Transformation 部分来演示：

// 1. 按空格切分单词
DataStream<String> words = textStream
    .flatMap(new FlatMapFunction<String, String>() {
        @Override
        public void flatMap(String value, Collector<String> out) {
            for (String word : value.split(" ")) {
                out.collect(word);
            }
        }
    });

// 2. 映射成 (word, 1) 二元组
DataStream<Tuple2<String, Integer>> wordAndOne = words
    .map(new MapFunction<String, Tuple2<String, Integer>>() {
        @Override
        public Tuple2<String, Integer> map(String word) {
            return Tuple2.of(word, 1);
        }
    });

// 3. 按单词分组，然后求和
DataStream<Tuple2<String, Integer>> wordCounts = wordAndOne
    .keyBy(value -> value.f0)
    .sum(1);

你想想看，这段代码的逻辑其实很简单：切分 → 映射 → 分组 → 聚合。但背后 Flink 做了大量工作——数据分区、网络传输、容错恢复，这些对用户都是透明的。

3.3 Sink：数据去哪了

Sink 是作业的终点。计算完的结果，总得有个去处吧？

Flink 同样提供了丰富的 Sink 实现：

Sink 类型	输出目标	常用方式
print / printToErr	控制台	调试用，每个元素打印到标准输出
writeAsText	文件系统	批量输出到文件
Kafka Producer	Kafka 主题	生产环境主流，FlinkKafkaProducer
JDBC Sink	关系型数据库	写入 MySQL、PostgreSQL 等
Elasticsearch Sink	ES 索引	适合日志、指标类数据
自定义 Sink	任意目标	实现 SinkFunction

在实时特征计算的场景里，我一般会把特征结果写到两个地方：一份写到 Kafka，供下游服务实时消费；另一份写到 Redis 或 HBase，做特征存储，供模型在线推理时查询。

// 最简单的 Sink：打印到控制台
wordCounts.print();

// 写入文件
wordCounts.writeAsText("path/to/output.txt");

// 写入 Kafka（生产环境常用）
wordCounts.addSink(new FlinkKafkaProducer<>(
    "localhost:9092",
    "feature-output",
    new SimpleStringSchema()
));

3.4 实战：WordCount 完整代码

好了，理论说完了，咱们来写一个完整的 WordCount。这是 Flink 的 "Hello World"，但别小看它——麻雀虽小五脏俱全。

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class WordCount {
    public static void main(String[] args) throws Exception {
        // 1. 创建执行环境
        StreamExecutionEnvironment env = 
            StreamExecutionEnvironment.getExecutionEnvironment();

        // 2. Source：从 Socket 读取数据
        DataStream<String> textStream = 
            env.socketTextStream("localhost", 9999);

        // 3. Transformation：计算逻辑
        DataStream<Tuple2<String, Integer>> wordCounts = textStream
            .flatMap(new FlatMapFunction<String, String>() {
                @Override
                public void flatMap(String value, Collector<String> out) {
                    for (String word : value.split("\\s+")) {
                        if (word.length() > 0) {
                            out.collect(word.toLowerCase());
                        }
                    }
                }
            })
            .map(new MapFunction<String, Tuple2<String, Integer>>() {
                @Override
                public Tuple2<String, Integer> map(String word) {
                    return Tuple2.of(word, 1);
                }
            })
            .keyBy(value -> value.f0)
            .sum(1);

        // 4. Sink：打印结果
        wordCounts.print();

        // 5. 启动作业
        env.execute("Socket WordCount");
    }
}

这段代码怎么跑？我建议你分两步走：

1. 先在本机启动一个 netcat 服务：nc -lk 9999
2. 然后运行上面的 WordCount 程序，在 netcat 窗口输入单词，观察控制台输出

3.5 避坑指南

最后，分享几个我这些年踩过的坑：

• 并行度设置不当：Source 的并行度不是越大越好。比如 Socket Source 只能单并行度，设大了反而报错。Kafka Source 的并行度最好和分区数一致。
• 忘记调用 execute：嗯，这个听起来很蠢，但我真见过有人写了半天代码，结果忘了 env.execute()，作业根本没启动。
• 状态大小失控：WordCount 看起来简单，但如果数据量极大，key 的基数又高，状态会膨胀得很快。生产环境一定要配置 State TTL。
• 背压问题：Sink 写入慢，Source 读得快，中间就会产生背压。我曾经遇到过 ES Sink 写入性能跟不上，导致整个作业延迟飙升。后来加了缓冲和批量写入才解决。