3. Flink DataStream API 基础：从入门到动手

好，咱们正式开始接触 Flink 的核心编程接口——DataStream API。说实话，我刚开始学 Flink 的时候，面对一堆抽象概念也有点懵。但后来我发现，只要抓住三个关键点，一切就清晰了：数据从哪来、数据怎么变、数据去哪了。这一章，我们就先搞定前两个。

3.1 DataStream API 是什么？

DataStream API，说白了就是 Flink 提供的一套 Java/Scala 接口，让你能像操作集合一样处理无界数据流。你想想看，传统的数据处理都是批量的——数据攒够了再算。但实时场景下，数据是源源不断来的，没法等。DataStream API 就是解决这个问题的。

我个人习惯把 DataStream 理解成一个「流动的集合」。它跟 Java 的 List 很像，但有两个本质区别：

• 元素是动态到达的——你不知道下一秒会来什么数据
• 操作是延迟执行的——你定义好转换逻辑，数据到了才触发计算

3.2 Source 算子：数据从哪来？

Source 是 DataStream 的起点。Flink 提供了丰富的内置 Source，我们重点讲三种最常见的。

3.2.1 从集合读取

这是学习阶段最常用的方式。适合写 Demo 和单元测试。

// 从 List 创建 DataStream
DataStream<String> stream = env.fromElements(
    "hello flink",
    "hello world",
    "flink streaming"
);

// 或者从集合创建
List<Integer> data = Arrays.asList(1, 2, 3, 4, 5);
DataStream<Integer> numStream = env.fromCollection(data);

嗯，这里要注意：fromElements 和 fromCollection 在 Flink 1.12 之后被标记为过时了，官方推荐用 fromSource。但作为学习入门，它们依然是最直观的。

3.2.2 从文件读取

生产环境中，文件 Source 常用于批处理或回溯历史数据。

// 读取文本文件，每行一条记录
DataStream<String> lineStream = env.readTextFile("hdfs:///data/input.log");

// 读取 CSV 文件（需要自定义解析）
DataStream<MyRecord> csvStream = env
    .readFile(new CsvInputFormat<MyRecord>(path) { ... }, 
              "hdfs:///data/records.csv");

3.2.3 从 Kafka 读取（重点）

这才是实时流处理的标配。Kafka Source 是 Flink 最常用的 Source 之一。

Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty("group.id", "flink-consumer");

DataStream<String> kafkaStream = env
    .addSource(new FlinkKafkaConsumer<>(
        "input-topic",          // 主题名
        new SimpleStringSchema(), // 反序列化器
        props                   // 消费者配置
    ));

这里有几个关键参数我建议你记住：

参数	作用	我的建议
bootstrap.servers	Kafka 集群地址	至少写两个 broker，防止单点
group.id	消费者组 ID	不同作业用不同 group.id
enable.auto.commit	是否自动提交 offset	建议设为 false，用 Flink 的 Checkpoint 管理

3.3 Transformation 算子：数据怎么变？

数据进来了，接下来就是各种转换。我挑四个最常用的算子来讲，它们覆盖了 80% 的日常场景。

3.3.1 map：一对一转换

map 是最简单的算子——输入一条，输出一条。就像流水线上的工人，每个零件经过他，他做一次加工。

// 将字符串转为大写
DataStream<String> upperStream = stream
    .map(new MapFunction<String, String>() {
        @Override
        public String map(String value) {
            return value.toUpperCase();
        }
    });

// Lambda 写法（Java 8+）
DataStream<String> upperStream = stream
    .map(value -> value.toUpperCase());

我个人强烈推荐用 Lambda 写法，代码简洁很多。但要注意：如果逻辑复杂（超过 3 行），还是单独写一个类吧，方便单元测试。

3.3.2 flatMap：一对多转换

flatMap 跟 map 的区别在于：输入一条，可以输出零条、一条或多条。典型的场景就是分词。

// 将一行文本拆成多个单词
DataStream<String> wordStream = lineStream
    .flatMap(new FlatMapFunction<String, String>() {
        @Override
        public void flatMap(String line, Collector<String> out) {
            String[] words = line.split(" ");
            for (String word : words) {
                out.collect(word);  // 每条单词都发射出去
            }
        }
    });

你看，flatMap 多了一个 Collector 参数。你调用 out.collect() 的次数决定了输出多少条。一次都不调用，那就没输出——这可以用来做过滤。

3.3.3 filter：过滤算子

filter 就是做筛选的。返回 true 保留，false 丢弃。

// 只保留长度大于 5 的单词
DataStream<String> longWords = wordStream
    .filter(value -> value.length() > 5);

// 或者过滤掉空字符串
DataStream<String> nonEmpty = wordStream
    .filter(value -> !value.isEmpty());

嗯，这里有个小细节：filter 不会改变数据类型，它只是决定哪些元素能通过。所以输入输出类型是一样的。

3.3.4 keyBy：分组算子（核心）

keyBy 是 Flink 里最重要的算子之一。它把数据按照某个 key 分组，后续的聚合操作（比如 sum、reduce、window）都是基于分组后的子流进行的。

// 按照单词分组
DataStream<Tuple2<String, Integer>> wordCounts = wordStream
    .map(word -> Tuple2.of(word, 1))
    .keyBy(value -> value.f0)   // 按单词分组
    .sum(1);                     // 对第二个字段求和

你想想看，没有 keyBy，sum 就是对全局求和。有了 keyBy，每个 key 都有自己的状态，各自独立累加。这就是 WordCount 的核心逻辑。

3.4 完整示例：一个简单的 WordCount

把上面学的串起来，写一个完整的 Flink 程序：

public class WordCount {
    public static void main(String[] args) throws Exception {
        // 1. 创建执行环境
        StreamExecutionEnvironment env = 
            StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 2. Source：从 socket 读取（方便测试）
        DataStream<String> text = env.socketTextStream("localhost", 9999);
        
        // 3. Transformation
        DataStream<Tuple2<String, Integer>> counts = text
            .flatMap((String line, Collector<String> out) -> {
                for (String word : line.split(" ")) {
                    out.collect(word);
                }
            })
            .returns(Types.STRING)  // Lambda 需要指定返回类型
            .map(word -> Tuple2.of(word, 1))
            .returns(Types.TUPLE(Types.STRING, Types.INT))
            .keyBy(value -> value.f0)
            .sum(1);
        
        // 4. Sink：打印到控制台
        counts.print();
        
        // 5. 启动执行
        env.execute("Socket WordCount");
    }
}