4. Flink DataStream API 基础：Source、Transformation、Sink

Flink DataStream API 是构建实时流处理应用的核心编程接口。本章将系统讲解 DataStream 程序的三大核心组件：Source（数据源）、Transformation（转换算子）和 Sink（数据输出端）。掌握这三部分，你就具备了编写基本 Flink 流处理作业的能力。

4.1 DataStream 程序结构概览

一个完整的 Flink DataStream 程序遵循“获取执行环境 → 添加 Source → 应用 Transformation → 配置 Sink → 触发执行”的固定流程。下面的 SVG 图展示了这一架构。

4.2 Source：数据从哪里来？

Source 是 DataStream 的入口，负责从外部系统读取数据并转换为 DataStream<T>。Flink 内置了多种 Source 实现，覆盖最常见的生产场景。

4.2.1 常用 Source 类型对比

4.2.2 Kafka Source 实战

Kafka 是实时流处理中最常用的 Source。以下代码演示如何从 Kafka 读取 JSON 格式的用户行为日志：

// 1. 设置 Kafka 连接参数
Properties kafkaProps = new Properties();
kafkaProps.setProperty("bootstrap.servers", "localhost:9092");
kafkaProps.setProperty("group.id", "flink-consumer-group");

// 2. 创建 Kafka Consumer（Flink 1.12+ 推荐新 API）
DataStream<String> kafkaStream = env
    .addSource(new FlinkKafkaConsumer<>(
        "user_behavior",           // topic 名称
        new SimpleStringSchema(),  // 反序列化器
        kafkaProps
    ));

// 3. 设置起始消费位置（可选）
kafkaStream = ((FlinkKafkaConsumer<String>) kafkaStream.getSource())
    .setStartFromLatest();        // 从最新偏移量开始

4.2.3 Socket 与文件 Source

Socket Source 适合快速验证流处理逻辑，文件 Source 则常用于历史数据回放：

// Socket Source：监听本地 9999 端口
DataStream<String> socketStream = env.socketTextStream("localhost", 9999);

// 文件 Source：读取 HDFS 上的日志文件（逐行读取）
DataStream<String> fileStream = env.readTextFile("hdfs:///data/logs/2024-01-01.log");

// 集合 Source：本地测试数据
DataStream<String> testStream = env.fromCollection(
    Arrays.asList("event_1", "event_2", "event_3")
);

4.3 Transformation：核心算子详解

Transformation 是 DataStream API 的精髓，每个算子都对数据流执行一次转换，形成新的 DataStream。本章重点讲解三个最基础的算子：map、flatMap 和 filter。

4.3.1 基础算子对比

4.3.2 map：一对一转换

map 是最常用的转换算子，对每条数据执行一次函数，输出一条新数据。以下示例将原始字符串解析为 UserEvent POJO：

// 定义 POJO
public static class UserEvent {
    public String userId;
    public String eventType;
    public long timestamp;
}

// map 转换：String → UserEvent
DataStream<UserEvent> eventStream = rawStream.map(new MapFunction<String, UserEvent>() {
    @Override
    public UserEvent map(String value) throws Exception {
        // 假设输入格式：user_123,click,1700000000
        String[] fields = value.split(",");
        UserEvent event = new UserEvent();
        event.userId = fields[0];
        event.eventType = fields[1];
        event.timestamp = Long.parseLong(fields[2]);
        return event;
    }
});

4.3.3 flatMap：一进多出

flatMap 允许一条输入产生零条、一条或多条输出，非常适合分词和展开操作：

// flatMap 示例：将句子拆分为单词
DataStream<String> wordStream = sentenceStream.flatMap(
    new FlatMapFunction<String, String>() {
        @Override
        public void flatMap(String sentence, Collector<String> out) {
            // 按空格分词，每个单词单独输出
            for (String word : sentence.split(" ")) {
                if (!word.isEmpty()) {
                    out.collect(word.toLowerCase());
                }
            }
        }
    }
);

4.3.4 filter：条件筛选

filter 根据布尔条件保留或丢弃数据，是数据清洗的核心工具：

// filter 示例：只保留 click 事件
DataStream<UserEvent> clickStream = eventStream.filter(
    new FilterFunction<UserEvent>() {
        @Override
        public boolean filter(UserEvent event) throws Exception {
            return "click".equals(event.eventType);
        }
    }
);

// Lambda 写法（Java 8+）
DataStream<UserEvent> clickStream = eventStream
    .filter(event -> "click".equals(event.eventType));

4.3.5 算子链式调用

实际开发中，三个算子经常组合使用，形成清晰的 ETL 管道：

DataStream<String> result = env
    .socketTextStream("localhost", 9999)          // Source
    .filter(line -> !line.startsWith("#"))        // 过滤注释行
    .flatMap(new Tokenizer())                     // 分词
    .map(word -> Tuple2.of(word, 1))              // 映射为 (word, 1)
    .keyBy(value -> value.f0)                     // 按单词分组
    .sum(1);                                      // 求和

4.4 Sink：结果输出到哪里？

Sink 是 DataStream 的终点，负责将处理结果写入外部系统。Flink 提供了丰富的内置 Sink，同时也支持自定义 Sink。

4.4.1 常用 Sink 类型对比

4.4.2 print Sink：最简调试输出

// 直接打印到标准输出（每个并行度一个前缀）
DataStream<String> stream = ...;
stream.print("my-job");  // 输出格式：[my-job:1] hello

// 打印到标准错误
stream.printToErr();

4.4.3 文件 Sink：持久化到存储

// 简单文件输出（仅适合小数据量测试）
stream.writeAsText("hdfs:///output/result.txt");

// 生产推荐：StreamingFileSink（支持 Exactly-Once）
StreamingFileSink<String> fileSink = StreamingFileSink
    .forRowFormat(
        new Path("hdfs:///output/"),
        new SimpleStringEncoder<>("UTF-8")
    )
    .withBucketAssigner(new DateTimeBucketAssigner<>("yyyy-MM-dd"))
    .build();

stream.addSink(fileSink);

4.4.4 Kafka Sink：结果回写消息队列

// 将处理结果写回 Kafka
Properties producerProps = new Properties();
producerProps.setProperty("bootstrap.servers", "localhost:9092");

DataStream<String> resultStream = ...;

resultStream.addSink(new FlinkKafkaProducer<>(
    "output_topic",              // 目标 Topic
    new SimpleStringSchema(),    // 序列化器
    producerProps
));

// 开启 Exactly-Once 语义（需配合 Checkpoint）
((FlinkKafkaProducer<String>) resultStream.getSink())
    .setSemantic(Semantic.EXACTLY_ONCE);

4.5 完整示例：实时 ETL 管道

以下代码整合了本章所有知识点，构建一个完整的实时 ETL 管道：从 Kafka 读取日志 → 过滤无效数据 → 拆分字段 → 写入文件系统：

public class RealTimeETLJob {
    public static void main(String[] args) throws Exception {
        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);  // 5 秒一次 Checkpoint

        // 2. Kafka Source
        Properties kafkaProps = new Properties();
        kafkaProps.setProperty("bootstrap.servers", "localhost:9092");
        kafkaProps.setProperty("group.id", "etl-consumer");

        DataStream<String> rawStream = env
            .addSource(new FlinkKafkaConsumer<>(
                "raw_logs",
                new SimpleStringSchema(),
                kafkaProps
            )).setStartFromLatest();

        // 3. Transformation 链
        DataStream<String> cleanStream = rawStream
            .filter(line -> !line.trim().isEmpty())          // 过滤空行
            .flatMap(new FlatMapFunction<String, String>() { // 拆分多事件
                @Override
                public void flatMap(String value, Collector<String> out) {
                    for (String event : value.split("\\|")) {
                        out.collect(event.trim());
                    }
                }
            })
            .map(new MapFunction<String, String>() {         // 统一格式
                @Override
                public String map(String value) {
                    return "[ETL] " + value;
                }
            });

        // 4. 文件 Sink
        StreamingFileSink<String> sink = StreamingFileSink
            .forRowFormat(
                new Path("hdfs:///data/clean_logs/"),
                new SimpleStringEncoder<>("UTF-8")
            )
            .withBucketAssigner(new DateTimeBucketAssigner<>("yyyy-MM-dd/HH"))
            .build();

        cleanStream.addSink(sink);

        // 5. 触发执行
        env.execute("RealTime ETL Pipeline");
    }
}

4.6 本章小结

• Source 是数据流的起点，Kafka Source 是生产环境的首选，支持 Exactly-Once 语义。
• Transformation 是业务逻辑的核心：map 做一对一转换，flatMap 做一对多展开，filter 做条件筛选。
• Sink 是数据流的终点，print 用于调试，StreamingFileSink 和 FlinkKafkaProducer 是生产级输出方案。
• 三个组件通过链式调用组合成完整的 DAG，由 env.execute() 触发执行。