Flink DataStream API 入门：从 Source 到 Transformation

好，我们正式开始 DataStream API 的学习。说实话，这是 Flink 开发中最常用、最核心的 API。你想想看，不管你是做实时 ETL、实时数仓，还是做风控、推荐，最终落到代码层面，都是在跟 DataStream 打交道。

我个人习惯把 DataStream API 分成三块来看：数据从哪来（Source）、数据怎么处理（Transformation）、数据往哪去（Sink）。这一章我们先搞定前两块。

DataStream API 概述

DataStream API 是 Flink 处理无界数据流（也就是实时数据）的标准编程接口。它提供了一套丰富的算子，让你可以像搭积木一样构建实时数据处理管道。

核心概念其实就几个：

• Stream（数据流）：一个永不停止的数据序列。可以是来自 Kafka 的日志，也可以是 Socket 的文本行。
• Transformation（转换）：对数据流施加的操作，比如过滤、映射、分组。每个 Transformation 会生成一个新的 DataStream。
• Pipeline（管道）：由 Source、多个 Transformation、Sink 串联而成的完整作业。

我记得刚接触 Flink 时，总觉得它跟 Spark Streaming 很像。但用久了你会发现，Flink 的 DataStream API 更贴近「流」的本质——每条数据到达后立即处理，而不是等一批凑齐了再处理。这个差异在延迟敏感的场景下非常明显。

Source 算子：数据从哪来

Source 是数据管道的起点。Flink 内置了多种 Source，覆盖了最常见的生产场景。我挑三个最常用的来讲：文件、Kafka、Socket。

1. 从文件读取数据

文件 Source 适合做离线数据的重放、测试，或者读取一些静态的维度数据。Flink 支持读取文本文件、CSV、Parquet 等格式。

// 读取文本文件，每行作为一个 String 元素
DataStream<String> textStream = env.readTextFile("hdfs:///data/input/access.log");

// 读取 CSV 文件，需要指定 POJO 类型
DataStream<EventLog> csvStream = env.readCsvFile("hdfs:///data/input/events.csv")
    .pojoType(EventLog.class, "userId", "eventType", "timestamp");

这里有个坑，我踩过。Flink 的文件 Source 默认是有界流，读完后作业就结束了。如果你想让作业一直跑，监听新文件，需要用 FileProcessingMode.PROCESS_CONTINUOUSLY。但说实话，生产环境我很少用文件 Source 做实时流，它更适合做批处理或测试。

2. 从 Kafka 读取数据

Kafka Source 是生产环境最常用的 Source，没有之一。Flink 和 Kafka 的集成非常成熟，支持 exactly-once 语义、动态分区发现、offset 提交等高级特性。

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

DataStream<String> kafkaStream = env.addSource(
    new FlinkKafkaConsumer<>("input-topic", 
        new SimpleStringSchema(), 
        props)
);

我个人习惯在 Kafka Source 上设置几个关键参数：

• setStartFromLatest()：从最新的 offset 开始消费，适合对历史数据不敏感的场景。
• setStartFromEarliest()：从头开始消费，适合首次上线或需要回溯的场景。
• enableAutoCommit()：我一般关掉自动提交，用 Flink 的 checkpoint 来管理 offset，更可靠。

3. 从 Socket 读取数据

Socket Source 主要用于测试和原型验证。它从指定的 host:port 读取文本行，每行作为一个元素。

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

嗯，这里要注意。Socket Source 在生产环境几乎不用，因为它没有持久化、没有容错。但用来快速验证 Transformation 逻辑非常方便。我经常在本地调试时用 nc 命令配合 Socket Source：

# 终端 1：启动 nc 服务
nc -lk 9999

# 终端 2：启动 Flink 作业，从 Socket 读取数据

这样你敲一行，Flink 处理一行，调试效率很高。

Transformation 算子：数据怎么变

Transformation 是 DataStream API 的核心。数据从 Source 进来后，经过一系列 Transformation，最终变成你想要的格式或结果。我挑四个最基础的算子来讲：map、flatMap、filter、keyBy。

1. map：一对一转换

map 是最简单的算子。输入一个元素，输出一个元素。常用于数据清洗、格式转换、字段提取。

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

// 将每行文本转为大写
DataStream<String> upperStream = rawStream.map(new MapFunction<String, String>() {
    @Override
    public String map(String value) throws Exception {
        return value.toUpperCase();
    }
});

用 Lambda 更简洁：

DataStream<String> upperStream = rawStream.map(value -> value.toUpperCase());

map 的输入输出类型可以不同。比如从 String 转成 Tuple2：

DataStream<Tuple2<String, Integer>> pairStream = rawStream.map(
    value -> Tuple2.of(value, value.length())
);

2. flatMap：一对多转换

flatMap 跟 map 的区别在于：map 必须输出一个元素，而 flatMap 可以输出零个、一个或多个元素。它非常适合做「拆分」操作。

// 将每行文本按空格拆分成单词
DataStream<String> wordStream = rawStream.flatMap(new FlatMapFunction<String, String>() {
    @Override
    public void flatMap(String value, Collector<String> out) throws Exception {
        String[] words = value.split("\\s+");
        for (String word : words) {
            out.collect(word);
        }
    }
});

你看，flatMap 多了一个 Collector 参数。你可以多次调用 out.collect() 输出多个元素，也可以不调用——这样输入元素就被过滤掉了。

3. filter：过滤

filter 算子根据条件保留或丢弃元素。条件返回 true 则保留，false 则丢弃。

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

filter 可以跟其他算子组合使用。比如先 filter 掉脏数据，再 map 做转换：

DataStream<EventLog> validStream = rawStream
    .filter(line -> line.startsWith("EVENT:"))
    .map(line -> parseEvent(line));

这样做的好处是，尽早过滤掉无效数据，减少后续算子的处理压力。

4. keyBy：分组

keyBy 是 DataStream API 中最重要的算子之一。它根据 key 对数据流进行逻辑分组，后续的聚合操作（如 sum、reduce）都是在分组的基础上进行的。

// 按单词分组，统计每个单词出现的次数
DataStream<Tuple2<String, Integer>> wordCountStream = wordStream
    .map(word -> Tuple2.of(word, 1))
    .returns(Types.TUPLE(Types.STRING, Types.INT))
    .keyBy(value -> value.f0)  // 按单词分组
    .sum(1);                     // 对第二个字段求和

keyBy 背后做了两件事：

• 计算 key 的 hash 值，决定数据发往哪个分区。
• 维护一个 keyed state，每个 key 都有自己的状态。

这里有个容易混淆的点：keyBy 之后的数据流类型变成了 KeyedStream，不再是普通的 DataStream。只有 KeyedStream 才能调用 sum、reduce、aggregate 等聚合算子。

知识体系总览

下面这张图把本章的核心知识点串起来了。你可以看到数据从 Source 流入，经过 Transformation 处理，最终流向 Sink 的完整路径。

算子对比速查表

为了方便你快速查阅，我把四个 Transformation 算子的核心差异整理成了表格：

算子	输入 → 输出	典型场景	注意事项
map	1 个元素 → 1 个元素	格式转换、字段提取、数据清洗	必须返回一个非 null 元素
flatMap	1 个元素 → 0~N 个元素	拆分字符串、展开嵌套结构	Collector 非线程安全
filter	1 个元素 → 0 或 1 个元素	数据过滤、脏数据剔除	尽早过滤可减轻下游压力
keyBy	DataStream → KeyedStream	分组聚合、状态管理	key 分布影响并行度

好了，这一章的内容就到这里。Source 和 Transformation 是 DataStream API 的基石，你先把这几个算子练熟，后续的复杂场景都是它们的组合和延伸。