在学习 Flink 的过程中，map、flatMap、filter、process 是最常用、也是最容易让人迷糊的几个算子。

很多初学者都会有这些疑问：

• 为什么 flatMap 里一定要写 Collector？
• 为什么 map 不能返回多个元素？
• process 到底强在哪里？什么时候该用？

本文将 从接口设计出发，结合 可运行 Demo + 实际运行结果，带你真正理解 Flink 算子的设计思想。

---

一、算子能力对照表

算子	输入 → 输出	是否可丢数据	是否可多输出	是否可用时间/状态
map	1 → 1	❌	❌	❌
filter	1 → 0/1	✅	❌	❌
flatMap	1 → 0/N	✅	✅	❌
process	1 → 0/N	✅	✅	✅

一句话总结：

越简单的算子，约束越多，Flink 能优化得越好；
越底层的算子，能力越强，但责任全在你。

---

二、测试数据

hello flink
hello world

---

三、map：一进一出

1. 接口定义

public interface MapFunction<T, R> {
    R map(T value) throws Exception;
}

特点：

• 一个输入
• 必须返回一个输出
• 不能多、不能少

---

2. Demo：字符串转大写

  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<String> source = env.fromElements(
                "hello flink",
                "hello world"
        );

        source.map(value -> value.toUpperCase()).print();
        try {
            env.execute("Simple Map Example");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

DataStream 程序是惰性执行的，必须调用 execute() 才会触发作业执行

3. 运行结果

8> HELLO WORLD
7> HELLO FLINK

---

四、filter：只负责“要不要”

1. 接口定义

public interface FilterFunction<T> {
    boolean filter(T value) throws Exception;
}

注意：

• 不能修改数据
• 只能决定保留 or 丢弃

---

2. Demo：只保留包含 flink 的行

source
    .filter(line -> line.contains("flink"))
    .print();

3. 运行结果

6> hello flink

---

五、flatMap：一进多出

1. 接口定义

public interface FlatMapFunction<T, O> {
    void flatMap(T value, Collector<O> out) throws Exception;
}

2. 为什么要 Collector？

因为：

flatMap 允许一条输入，输出 0 条、1 条或多条数据

返回值已经不够用，所以 Flink 把“输出控制权”交给你。

---

3. Demo：拆分单词

        source.flatMap((String line, Collector<String> out) -> {
                    for (String word : line.split(" ")) {
                        out.collect(word);
                    }
                })
                .returns(Types.STRING)   // ⭐ 关键：补全类型信息
                .print();

4. 运行结果

4> hello
4> world
3> hello
3> flink

---

六、process：最底层、最强大的算子

map / filter / flatMap 能做的，process 全都能做
并且还能：

• 获取时间
• 使用状态
• 注册定时器

---

1. 接口结构

public abstract class ProcessFunction<I, O> {
    public abstract void processElement(
        I value,
        Context ctx,
        Collector<O> out
    ) throws Exception;
}

---

2. Demo：手写 WordCount（不使用 keyBy.sum）

      source
                .keyBy(value -> value)
                .process(new ProcessFunction<String, Tuple2<String, Integer>>() {

                    private int count = 0;

                    @Override
                    public void processElement(
                            String value,
                            Context ctx,
                            Collector<Tuple2<String, Integer>> out) {

                        count++;
                        out.collect(Tuple2.of(value, count));
                    }
                })
                .print();

3. 运行结果

3> (hello flink,1)
6> (hello world,1)

⚠️ 注意：这里只是演示 process 能力
实际生产应使用 Keyed State 而不是普通成员变量

---

七、如何选择算子？

官方推荐原则：

能用 map，就别用 flatMap
能用 flatMap，就别用 process

原因是：

• 简单算子 → Flink 能做更多优化
• process → 灵活但难维护、难调优

---

八、总结

• map：最简单，1 → 1
• filter：只做判断
• flatMap：拆分、多输出
• process：终极武器，慎用

理解算子 ≠ 记 API
理解算子 = 理解接口设计 + 数据流模型