Flink 核心算子详解：map / flatMap / filter / process-编程实验室

文章目录

一、算子能力对照表
二、测试数据
三、map：一进一出
四、filter：只负责“要不要”
五、flatMap：一进多出
六、process：最底层、最强大的算子
七、如何选择算子？
八、总结

在学习 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. 接口定义

publicinterfaceMapFunction<T,R>{Rmap(Tvalue)throwsException;}

特点：

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

2. Demo：字符串转大写

StreamExecutionEnvironmentenv=StreamExecutionEnvironment.getExecutionEnvironment();DataStream<String>source=env.fromElements("hello flink","hello world");source.map(value->value.toUpperCase()).print();try{env.execute("Simple Map Example");}catch(Exceptione){thrownewRuntimeException(e);}

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

3. 运行结果

8> HELLO WORLD 7> HELLO FLINK

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

1. 接口定义

publicinterfaceFilterFunction<T>{booleanfilter(Tvalue)throwsException;}

注意：

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

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

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

3. 运行结果

6> hello flink

五、flatMap：一进多出

1. 接口定义

publicinterfaceFlatMapFunction<T,O>{voidflatMap(Tvalue,Collector<O>out)throwsException;}

2. 为什么要Collector？

因为：

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

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

3. Demo：拆分单词

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

4. 运行结果

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

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

map / filter / flatMap 能做的，process 全都能做
并且还能：
获取时间
使用状态
注册定时器

1. 接口结构

publicabstractclassProcessFunction<I,O>{publicabstractvoidprocessElement(Ivalue,Contextctx,Collector<O>out)throwsException;}

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

source.keyBy(value->value).process(newProcessFunction<String,Tuple2<String,Integer>>(){privateintcount=0;@OverridepublicvoidprocessElement(Stringvalue,Contextctx,Collector<Tuple2<String,Integer>>out){count++;out.collect(Tuple2.of(value,count));}}).print();