📚 课程目录第3章 Flink DataStream API入门

3. Flink DataStream API 入门：Source、Transformation、Sink 基础操作与 WordCount 实战

本章是 Flink 实时特征开发的起点。你将掌握 DataStream API 的三大核心组件：Source（数据源）、Transformation（转换算子）和 Sink（数据输出），并通过经典的 WordCount 案例串联整个流程。理解这些基础操作，是后续构建复杂实时特征 pipeline 的基石。

3.1 DataStream API 核心三要素

Flink DataStream API 将实时数据流抽象为 DataStream[T]，所有操作围绕以下三个环节展开：

环节	职责	常见实现	特征场景
Source	接入外部数据，将数据转换为 DataStream	Kafka、Socket、文件、自定义	实时日志、交易流、传感器数据
Transformation	对 DataStream 进行转换、过滤、聚合等操作	map、flatMap、filter、keyBy、window	特征提取、数据清洗、指标计算
Sink	将结果数据输出到外部系统	Kafka、Redis、MySQL、文件、自定义	特征存储、实时报表、下游触发

开发提示：在实时特征服务中，Source 通常对接 Kafka（事件流），Sink 通常对接 Redis 或 HBase（特征存储）。Transformation 层是特征逻辑的核心战场。

3.2 Source：数据接入

Flink 提供了丰富的内置 Source，同时也支持通过 SourceFunction 或 RichSourceFunction 实现自定义数据源。以下是开发中最常用的几种方式：

Socket Source：用于测试和原型验证，从 TCP Socket 读取文本流。
File Source：读取文件（支持 HDFS、本地文件），可配置读取模式（一次性或持续监控）。
Kafka Source：生产环境首选，通过 FlinkKafkaConsumer 接入 Kafka Topic。
自定义 Source：实现 SourceFunction 接口，适用于特殊协议或模拟数据生成。

// 示例：Socket Source（测试用）
DataStream<String> textStream = env.socketTextStream("localhost", 9999);

// 示例：Kafka Source（生产用）
Properties kafkaProps = new Properties();
kafkaProps.setProperty("bootstrap.servers", "localhost:9092");
kafkaProps.setProperty("group.id", "flink-feature-group");

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

3.3 Transformation：核心算子

Transformation 是特征计算的引擎。以下算子构成了实时特征开发的基本工具箱：

算子	输入 → 输出	典型用途	是否触发 shuffle
`map`	1 → 1	字段提取、类型转换、特征值映射	否
`flatMap`	1 → 0..n	分词、日志拆分、事件展开	否
`filter`	1 → 0/1	过滤无效事件、白名单筛选	否
`keyBy`	DataStream → KeyedStream	按用户ID、设备ID分组	是（网络传输）
`sum / reduce`	KeyedStream → DataStream	累加计数、滚动聚合	否（在分区内）
`window`	KeyedStream → WindowedStream	滑动/滚动窗口特征（如近5分钟均值）	否（在分区内）

  关键理解：keyBy 是 Flink 中唯一触发网络数据重分区的算子。在特征服务中，keyBy 通常按“实体ID”（如 user_id、device_id）进行，确保同一个实体的所有事件被路由到同一个并行子任务，从而保证状态一致性和窗口计算的正确性。

3.4 Sink：结果输出

Sink 负责将计算后的特征值写入外部存储。Flink 提供了 addSink 方法，支持多种输出方式：

print()：调试用，将结果输出到标准输出。
Kafka Sink：将特征结果写回 Kafka，供下游服务消费。
Redis Sink：实时特征服务中最常用的 Sink，将特征值写入 Redis 供在线推理使用。
JDBC Sink：写入关系型数据库，适用于离线分析或特征回溯。

// 示例：print Sink（调试）
wordCountStream.print("WordCount Result");

// 示例：Kafka Sink（生产）
kafkaStream.addSink(new FlinkKafkaProducer<>(
    "feature-output-topic",
    new SimpleStringSchema(),
    kafkaProps
));

3.5 WordCount 实战：完整 Pipeline

WordCount 是分布式计算的“Hello World”。在 Flink 中，它完整展示了 Source → Transformation → Sink 的串联过程。下面是一个基于 Socket 的实时 WordCount 实现：

public class SocketWordCount {
    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>> wordCountStream = textStream
            .flatMap(new Tokenizer())          // 1 → n：拆分单词
            .keyBy(value -> value.f0)          // 按单词分组
            .sum(1);                           // 累加计数

        // 4. Sink：打印结果
        wordCountStream.print("WordCount");

        // 5. 启动任务
        env.execute("Socket WordCount Job");
    }

    // 自定义 flatMap 函数：将每行文本拆分为单词
    public static class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
            String[] words = value.toLowerCase().split("\\W+");
            for (String word : words) {
                if (word.length() > 0) {
                    out.collect(new Tuple2<>(word, 1));
                }
            }
        }
    }
}

运行前提：在执行上述代码前，需要先启动一个 Socket 数据源。在终端执行 nc -lk 9999（Linux/Mac）或使用其他 TCP 工具，然后输入文本，Flink 作业会实时输出单词计数结果。

3.6 WordCount 数据流图

下图展示了 WordCount 作业在 Flink 运行时中的数据流动与算子转换关系：

3.7 并行度与任务链

Flink 默认会将连续的算子合并为 任务链（Operator Chain），以减少线程切换和网络开销。在 WordCount 示例中：

flatMap 和 keyBy 之间由于存在网络 shuffle，不能合并为同一个任务链。
keyBy 之后的 sum 和 print 可以合并（如果并行度一致且未禁用链）。
通过 env.setParallelism(N) 设置全局并行度，或通过算子后的 .setParallelism(N) 单独指定。

算子	默认并行度	是否可链	说明
Source	env 全局并行度	是（与 map/flatMap）	非并行 Source（如 Socket）只能为 1
flatMap	继承上游	是（与 Source）	无状态算子，可任意链
keyBy	继承上游	否（触发 shuffle）	网络重分区，必须断开链
sum	继承 keyBy	是（与后续 Sink）	有状态算子，但可链
Sink	继承上游	是（与上游算子）	print 可链，Kafka Sink 通常独立

性能建议：在特征服务中，尽量将无状态转换（map、flatMap、filter）与 Source 链在一起，减少线程数。keyBy 后的聚合操作建议单独设置并行度，与下游 Sink 的写入能力匹配，避免背压。

3.8 本章小结

通过本章学习，你已掌握 Flink DataStream API 的三大核心操作：

Source：从 Socket、Kafka 等系统接入实时数据流。
Transformation：使用 map、flatMap、filter、keyBy、sum 等算子构建特征计算逻辑。
Sink：将结果输出到控制台、Kafka 或外部存储。