3. Flink编程模型

好，咱们正式开始动手写Flink代码了。这一章是整个实时ETL的基石——DataStream API。说白了，你后面所有的数据清洗、转换、输出，都得靠这套API来搭桥。

我记得刚接触Flink时，最困惑的就是“这玩意儿到底怎么跑起来的？”后来发现，Flink程序的结构其实非常固定，就像一个四步走的流水线：拿数据 → 换数据 → 送数据 → 点火。咱们今天就把这四步拆开揉碎，再拼回去。

3.1 Flink程序骨架：四步走

先看一个最简的骨架，你以后写的每个Flink作业，都长这样：

// 1. 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 2. 添加数据源（Source）
DataStream<String> sourceStream = env.readTextFile("input/data.txt");

// 3. 数据转换（Transformation）
DataStream<String> resultStream = sourceStream
    .map(line -> line.toUpperCase())
    .filter(line -> line.startsWith("A"));

// 4. 输出结果（Sink）
resultStream.print();

// 5. 启动执行
env.execute("My First Flink Job");

看到了吗？就这五步。我刚开始带团队时，经常有人问“为什么我写了代码没反应？”——十有八九是忘了写env.execute()。你想想看，没有点火，引擎再漂亮也动不了。

3.2 Source（数据源）：数据从哪来？

Source就是数据的入口。Flink支持的数据源类型非常多，我把它分成三大类：

类别	典型实现	适用场景
基于文件	readTextFile、readFile	离线测试、批量回溯
基于Socket	socketTextStream	快速原型、调试
基于消息队列	addSource(new FlinkKafkaConsumer<>(...))	生产环境实时ETL

我个人习惯，在开发阶段先用socketTextStream或readTextFile快速验证逻辑，等确认无误后再切换到Kafka。为什么呢？因为调试Kafka需要启动外部服务，太慢了。

3.3 Transformation（转换算子）：数据的魔法工坊

Transformation是Flink最核心的部分。数据进来后，怎么洗、怎么切、怎么合并，全靠这些算子。我挑几个最常用的给你讲透：

3.3.1 单流基础算子

• map：一对一转换。比如把每行日志的级别从INFO改成WARN。
• filter：过滤。只保留符合条件的元素。
• flatMap：一对多转换。比如把一句话拆成单词。

DataStream<String> words = sentences.flatMap(
    (String sentence, Collector<String> out) -> {
        for (String word : sentence.split(" ")) {
            out.collect(word);
        }
    }
).returns(Types.STRING);

嗯，这里要注意.returns(Types.STRING)。Java的泛型擦除会导致Flink推断不出类型，不加这行可能会报错。我曾经在这个坑里爬了半小时。

3.3.2 分组与聚合算子

实时ETL里最常用的就是keyBy + 聚合。比如统计每个用户的点击次数：

DataStream<Tuple2<String, Integer>> clicks = ...;
DataStream<Tuple2<String, Integer>> result = clicks
    .keyBy(value -> value.f0)   // 按用户名分组
    .sum(1);                     // 对第二个字段求和

为什么用keyBy而不是groupBy？因为Flink是流处理，数据是无穷的，keyBy会把相同key的数据路由到同一个并行子任务，这样才能做增量聚合。

3.3.3 多流合并算子

实际项目中，经常需要把多个流合并处理。比如把订单流和支付流关联起来：

• union：合并多个同类型流。
• connect + coMap：连接两个不同类型流，可以共享状态。

3.4 Sink（数据输出）：数据去哪？

Sink是数据的终点。Flink内置了丰富的Sink实现：

Sink类型	常用方法	说明
控制台	print()	调试用，每个并行度输出一行
文件	writeAsText()	生产环境慎用，容易产生小文件
Kafka	addSink(new FlinkKafkaProducer<>(...))	生产环境首选，支持exactly-once
自定义	addSink(new MySink())	实现SinkFunction接口

我个人建议，生产环境尽量用Kafka或JDBC Sink，并且一定要开启checkpoint。为什么？因为只有checkpoint + 合适的Sink才能保证数据不丢不重。这个后面专门讲。

3.5 知识体系总览

下面这张图，把Flink编程模型的整体脉络画出来了。你对照着看，心里就有谱了：

3.6 实战：写一个完整的实时ETL雏形

光说不练假把式。咱们写一个完整的例子：从Socket读取日志，过滤出ERROR级别的，然后输出到控制台。

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

        // 2. Source：从Socket读取（启动前先 nc -lk 9999）
        DataStream<String> logStream = env.socketTextStream("localhost", 9999);

        // 3. Transformation：过滤出ERROR日志
        DataStream<String> errorStream = logStream
            .filter(line -> line.contains("ERROR"))
            .map(line -> "[严重] " + line);

        // 4. Sink：打印到控制台
        errorStream.print();

        // 5. 启动
        env.execute("Socket Error Filter");
    }
}

你可以在终端先运行nc -lk 9999，然后输入几行日志试试。看到没？只有包含ERROR的行才会被打印出来，而且前面加上了“[严重]”前缀。这就是一个最简的实时ETL流程。