1. 为什么这三个算子值得单独拎出来讲刚接触 Flink 的人十有八九是从map、filter、flatMap这三个算子入门的。它们看起来简单到不需要解释——map就是一进一出filter就是过滤flatMap就是一进多出。但我在实际带人做流处理任务的过程中发现恰恰是这三个最基础的算子藏着最多想当然的坑。举个真实的场景。之前有个做实时数据清洗的任务需求是把上游过来的原始日志做字段提取、无效记录剔除、再把嵌套结构打平。一个同事用map硬扛结果遇到脏数据直接抛异常整个作业重启另一个同事用flatMap加filter组合同样的数据量下吞吐反而更高容错也更稳。同样的功能写法不同线上表现天差地别。这篇文章就是想把这三个算子彻底讲透。不管你是刚学 Flink 的新手还是已经写过几个作业但总觉得能用但不够优雅的老手都能从里面找到能直接抄作业的东西。我会从它们各自的工作机制讲起然后重点说清楚什么时候该用哪个、组合使用时怎么设计、性能上有什么讲究最后给几段可以直接复现的代码。核心关键词就三个Flink 算子、map、flatMap外加一个绕不开的filter。需要先明确一点这三个算子都属于 Flink 的转换算子Transformation作用在DataStream上返回的也是DataStream。它们都是无状态的除非你手动引入状态也就是说每条数据独立处理互不影响。这个特性决定了它们特别适合做数据清洗、格式转换、字段映射这类逐条处理的活儿。理解了这一点后面的选型和优化才有根基。2. 三个算子的工作机制拆解2.1 map最纯粹的一进一出map的语义非常干净输入一条数据输出一条数据数量严格 1:1。你给它一个MapFunction它就把T类型的元素变成R类型的元素。DataStreamString lines env.fromElements(1,apple, 2,banana); DataStreamString names lines.map(new MapFunctionString, String() { Override public String map(String value) throws Exception { return value.split(,)[1]; } });从执行层面看map在算子链里是最轻量的。它不需要维护任何跨记录的状态每条数据进来调用一次map()方法结果直接往下游发。正因为如此map的吞吐能力非常强在算子链优化Operator Chaining开启的情况下它通常会和上下游算子合并成一个 Task减少线程间切换和序列化开销。但这里有个容易被忽略的点map里抛异常会怎样默认情况下一条数据在map中抛出异常会导致整个算子失败进而触发作业重启取决于你的重启策略。所以如果你在map里做split、类型转换这类可能失败的操作一定要想清楚容错方案——要么提前用filter把脏数据挡掉要么在map内部 try-catch 返回一个兜底值。我见过太多作业因为一条格式不对的日志就反复重启排查半天才发现是map里没做防御。2.2 filter只做减法不做变形filter的职责单一到极致判断每条数据是否保留返回true留下返回false丢弃。它不改变数据的类型输入T输出还是T。DataStreamString valid lines.filter(new FilterFunctionString() { Override public boolean filter(String value) throws Exception { return value ! null value.contains(,); } });很多人觉得filter没什么可讲的但实际用起来有两个细节值得注意。第一filter的返回值必须是布尔值不能返回 null。如果你写return value.length() 0 ? true : null;这种运行时会直接 NPE。这个错误在编译期不报跑起来才炸很隐蔽。第二filter和map的顺序会影响性能。假设你要先过滤掉无效数据再对有效数据做复杂转换那一定要先 filter 后 map。因为filter把数据量降下来之后map里那些耗时的解析、计算操作就少执行很多次。反过来先 map 后 filter等于对一堆马上要丢掉的数据白做了一遍转换。这个顺序问题在数据量大、脏数据比例高的时候性能差距能到几倍。2.3 flatMap一进多出的万能选手flatMap是三个里最灵活的。它的输入是一条数据输出是零条、一条或多条数据。注意这个零条——意味着flatMap其实可以同时干filter和map的活儿。DataStreamString words lines.flatMap(new FlatMapFunctionString, String() { Override public void flatMap(String value, CollectorString out) throws Exception { for (String w : value.split(,)) { if (w.length() 0) { out.collect(w); } } } });flatMap的核心是那个Collector。你通过调用out.collect()来决定往下游发多少条数据。不调用就是丢弃相当于 filter调用一次就是 map调用多次就是一对多展开。这种设计让flatMap特别适合处理一条记录里包含多个子元素的场景比如一行日志里有多个字段、一个订单里有多个商品、一段文本要分词。不过灵活是有代价的。flatMap因为要处理变长的输出内部实现比map稍微重一点而且如果collect调用次数过多会给下游造成压力。我一般的原则是能用 map 或 filter 明确表达的就不要用 flatMap 兜底。因为明确的算子语义更清晰也更容易被 Flink 的优化器识别和优化。只有当一进多出或可能零输出是真实需求时才用flatMap。2.4 三者的本质对比把三个算子放在一起看差异其实就集中在两个维度上输出数量和输出类型。算子输入→输出数量输出类型典型用途map1 → 1可变T→R字段提取、格式转换、类型映射filter1 → 0 或 1不变T→T数据清洗、条件筛选flatMap1 → 0/1/N可变T→R拆分、展开、分词、多路输出理解这张表选型基本就不会错。但选型只是第一步真正决定作业质量的是怎么把它们组合起来以及在组合时避开哪些坑。这才是下一部分要重点讲的。3. 选型与组合什么时候用哪个3.1 一个判断口诀先看数量再看类型我在实际项目里总结了一个简单的判断顺序基本能覆盖 90% 的场景。第一步问自己这条数据我可能不要吗如果可能丢弃那filter或flatMap是候选。如果每条都必须保留那map或flatMap是候选。第二步问自己输出的类型和输入一样吗如果一样且只是筛选用filter。如果类型变了且严格一对一用map。第三步只有当数量不固定或类型和数量都不固定时才轮到flatMap。这个顺序的好处是逼你先想清楚需求而不是一上来就抓flatMap这个万能工具。我见过不少代码明明就是简单的字段提取非要用flatMap写结果可读性差、性能也没优势。3.2 组合使用的经典模式实际任务里这三个算子几乎不会单独出现而是串成一条处理链。下面几种组合模式是我用得最多的。模式一先 filter 后 map做数据清洗。这是最常见的组合。上游数据往往有脏记录先用filter把明显不合法的挡掉再用map做字段解析和转换。顺序不能反原因前面说过——先降数据量再算省资源。DataStreamOrder orders raw .filter(s - s ! null s.split(,).length 4) .map(s - { String[] p s.split(,); return new Order(p[0], p[1], Double.parseDouble(p[2]), Long.parseLong(p[3])); });模式二flatMap 拆分 filter 二次筛选。当一条记录要展开成多条且展开后还要再筛一遍时这个组合很自然。比如日志按分隔符拆成多个事件再过滤掉其中不需要的事件类型。模式三map 做预处理 flatMap 做展开。有时候原始数据格式很乱先用map规整成统一结构再用flatMap展开。这样flatMap里的逻辑会干净很多不用一边解析一边展开。3.3 别为了少写算子牺牲语义有个常见的误区为了减少算子数量把filter和map的逻辑全塞进一个flatMap里。代码是短了但问题不少。一是可读性下降。一个flatMap里又判断又转换又展开别人接手时得逐行读才能搞懂在干嘛。二是调试困难。出问题时你分不清是过滤逻辑错了还是转换逻辑错了。三是优化受限。Flink 对独立的filter和map有更明确的优化空间混在一起反而不好处理。我的建议是除非性能压测证明合并算子有显著收益否则优先保证语义清晰。算子数量多一两个对性能的影响远没有想象中大但可维护性的差距是实打实的。4. 性能与并行度的那些讲究4.1 算子链为什么你的 map 和 filter 会合并成一个 TaskFlink 有个很重要的优化叫算子链Operator Chaining。默认情况下如果两个算子之间是forward或one-to-one的数据分发关系也就是不需要 shuffleFlink 会把它们合并到同一个 Task 里执行。这意味着你写的filter().map().flatMap()这条链很可能在运行时就是一个线程里顺序执行的三段逻辑中间没有网络传输也没有序列化反序列化。这是 Flink 性能好的关键原因之一。但这个优化有个前提算子之间的并行度必须一致。如果你给map单独设了setParallelism(4)而上下游是 2那它们就没法链在一起中间会插入网络 shuffle性能会明显下降。所以除非有明确的负载不均问题我一般不建议对map、filter、flatMap单独设并行度让它们跟着作业的全局并行度走就好。4.2 并行度设置的经验值那全局并行度怎么定这跟数据量、单条处理耗时、集群资源都有关。一个粗略的经验并行度 ≈ 峰值 QPS × 单条处理耗时秒 / 目标单并行度吞吐。举个例子假设峰值每秒来 10 万条数据每条在map里处理耗时 0.1 毫秒那单并行度理论能处理 1 万条/秒10 万条就需要至少 10 个并行度。实际还要留余量设 12 到 16 比较稳。但要注意map、filter这类无状态算子单并行度的处理能力其实很强瓶颈往往不在它们身上而在下游的聚合、join 或 sink。所以调并行度时先看整个链路的瓶颈在哪别盲目给map加并行度。4.3 序列化开销别在 map 里造大对象map和flatMap的输出如果要在算子之间传输比如后面接了keyBy就需要序列化。这时候如果你在map里 new 了一个巨大的对象序列化的开销会很可观。我踩过的一个坑在map里把整条原始日志几 KB塞进一个 POJO 往下传后面其实只用到了其中两个字段。结果序列化成了瓶颈。后来改成在map里就只提取需要的字段输出一个小对象吞吐直接上了一个台阶。原则很简单map 的输出尽量精简只带下游真正需要的字段。别图省事把整个原始对象往下传。4.4 异步与阻塞map 里别做同步 IOmap、filter、flatMap都是同步执行的。如果你在map里调用一个外部接口比如查数据库、调 HTTP那这个算子的吞吐会被 IO 延迟死死拖住。一条数据等 50 毫秒单并行度每秒就只能处理 20 条再多的并行度也救不回来。这种场景应该用 Flink 的异步 IOAsync I/O或者把外部查询结果提前广播Broadcast进来做本地关联。在map里做同步 IO 是新手最容易犯的性能错误之一一定要警惕。5. 踩坑实录那些年我在这三个算子上栽的跟头5.1 坑一filter 返回 null 导致的 NPE前面提过这里展开说。FilterFunction的filter方法签名返回boolean但如果你写的是包装类型Boolean并且返回了nullFlink 在自动拆箱时会抛 NPE。这个错误在本地小数据量测试时可能不出现因为没走到那条分支一上生产就炸。排查过程很典型作业日志里只有一句NullPointerException堆栈指向 Flink 内部看不出是哪条数据。后来在filter里加了日志才发现某条数据的某个字段为 null导致判断逻辑返回了 null。修复很简单保证filter永远返回基本类型boolean所有分支都有明确返回值。但教训是任何返回布尔值的判断逻辑都要考虑 null 的情况。5.2 坑二flatMap 里 collect 次数过多拖垮下游有个任务要把一条包含上千个元素的记录展开。一开始用flatMap一次性collect出上千条结果下游算子瞬间被压垮反压一路传到源头。问题在于flatMap的collect是同步的你 collect 多少条下游就得立刻处理多少条。如果一次展开的量太大会造成瞬时流量尖峰。解决办法有两个一是分批 collect比如每 100 条做一次小批量的输出控制二是在下游加缓冲或限流。更根本的思路是评估这个一对多是否真的合理能不能在更上游就把数据拆开。5.3 坑三map 里做重计算导致反压反压Backpressure是流处理里最常见的现象。有一次作业一直报反压排查发现是map里做了一段正则匹配而正则写得不够高效单条处理耗时波动很大。这里的关键认知是map 里的逻辑耗时必须稳定且可控。如果某条数据的处理耗时是其他数据的几十倍就会造成处理速率的抖动进而引发反压。解决办法是优化那段逻辑比如预编译正则、用更简单的字符串操作替代或者把重计算拆到独立的算子并单独调并行度。5.4 坑四误以为 flatMap 能替代所有场景前面反复强调过但还是值得单独列出来。flatMap灵活但灵活不等于合适。我见过有人用flatMap实现filter的功能代码里一堆if判断最后collect或者不collect。这种写法不仅可读性差而且因为flatMap内部要维护Collector的状态性能上也不如纯filter。选型的第一原则永远是语义匹配而不是哪个算子更万能。6. 可直接复现的完整示例6.1 环境与依赖下面这段代码基于 Flink 1.17Java 11。Maven 依赖只需要核心的两个dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version1.17.0/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-clients/artifactId version1.17.0/version /dependency6.2 一个完整的清洗任务需求模拟一批订单日志格式为订单ID,商品名,金额,时间戳。要做三件事——过滤掉字段数不对的脏数据、把金额从字符串转成 double、把商品名按逗号拆成多个标签。import org.apache.flink.api.common.functions.FilterFunction; import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.util.Collector; public class MapFilterFlatMapDemo { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(2); DataStreamString raw env.fromElements( 1001,apple|banana,19.9,1700000000, 1002,orange,29.9,1700000001, bad-data, 1003,grape|pear|peach,9.9,1700000002 ); // 第一步filter 挡掉脏数据 DataStreamString valid raw.filter(new FilterFunctionString() { Override public boolean filter(String value) { if (value null) return false; String[] parts value.split(,); return parts.length 4; } }); // 第二步map 做字段转换 DataStreamOrder orders valid.map(new MapFunctionString, Order() { Override public Order map(String value) { String[] p value.split(,); return new Order(p[0], p[1], Double.parseDouble(p[2]), Long.parseLong(p[3])); } }); // 第三步flatMap 把商品标签展开 DataStreamString tags orders.flatMap(new FlatMapFunctionOrder, String() { Override public void flatMap(Order order, CollectorString out) { for (String tag : order.getItems().split(\\|)) { if (!tag.isEmpty()) { out.collect(order.getOrderId() - tag); } } } }); tags.print(); env.execute(map-filter-flatmap-demo); } public static class Order { private String orderId; private String items; private double amount; private long ts; public Order() {} public Order(String orderId, String items, double amount, long ts) { this.orderId orderId; this.items items; this.amount amount; this.ts ts; } public String getOrderId() { return orderId; } public String getItems() { return items; } public double getAmount() { return amount; } public long getTs() { return ts; } } }跑起来你会看到bad-data被filter挡掉1001和1003的商品标签被flatMap展开成多条输出。这条链就是最典型的清洗—转换—展开三段式。6.3 用 Lambda 简化写法Java 8 之后这三个算子都支持 Lambda代码能短一大截。上面那段可以改写成DataStreamString valid raw.filter(s - s ! null s.split(,).length 4); DataStreamOrder orders valid.map(s - { String[] p s.split(,); return new Order(p[0], p[1], Double.parseDouble(p[2]), Long.parseLong(p[3])); }); DataStreamString tags orders.flatMap((Order o, CollectorString out) - { for (String tag : o.getItems().split(\\|)) { if (!tag.isEmpty()) out.collect(o.getOrderId() - tag); } });Lambda 写起来爽但有个坑泛型擦除。Flink 需要从函数签名里推断输出类型Lambda 有时候推断不出来会报InvalidTypesException。这时候要么显式加returns()要么老老实实写匿名内部类。我一般建议在类型复杂的地方用匿名类简单的地方用 Lambda别为了短而短。6.4 验证与观察跑完之后重点看两件事。一是输出条数是否符合预期——filter挡掉了几条、flatMap展开成了几条心里要有数。二是算子链是否生效——在 Flink Web UI 的作业图里如果filter、map、flatMap显示在同一个 Task 框里说明链化成功中间没有 shuffle。如果发现它们被拆成了不同的 Task检查一下是不是中间有keyBy、rebalance这类会触发 shuffle 的操作或者并行度设置不一致。7. 几个容易被忽略的进阶细节7.1 RichFunction需要生命周期时用它普通的MapFunction、FilterFunction是无状态的但如果你需要在算子启动时初始化一些资源比如加载字典、建立连接或者结束时释放资源就要用Rich 版本RichMapFunction、RichFilterFunction、RichFlatMapFunction。它们多了open()和close()两个方法分别在算子初始化时和结束时调用。典型用法是在open()里加载一份配置或字典然后在map()里直接用避免每条数据都去读一次。public class DictMap extends RichMapFunctionString, String { private transient MapString, String dict; Override public void open(Configuration parameters) { dict loadDict(); // 只加载一次 } Override public String map(String value) { return dict.getOrDefault(value, value); } }注意dict要声明为transient因为它不需要被序列化到各个并行子任务——每个子任务在open()时各自加载一份。7.2 累加器在 map 里统计指标Flink 提供了累加器Accumulator可以在map、flatMap里做计数、求和这类统计然后在作业结束后读取。这在排查数据质量问题时特别有用比如统计有多少条数据被filter挡掉了。private LongCounter dirtyCounter getRuntimeContext() .getLongCounter(dirty-records, new LongCounter());在filter里判断为脏数据时dirtyCounter.add(1)作业结束后就能拿到总数。比翻日志高效得多。7.3 类型信息别让 Flink 猜Flink 需要知道每个算子的输入输出类型才能做序列化和优化。大多数时候它能自动推断但遇到泛型嵌套、Lambda、或者自定义类型时推断可能失败。这时候用returns()显式声明DataStreamTuple2String, Integer result stream .map(...) .returns(Types.TUPLE(Types.STRING, Types.INT));显式声明类型还有个好处性能。Flink 对已知类型会用更高效的序列化器比如对 POJO 用字段级序列化比通用的 Kryo 快不少。所以自定义类型尽量写成标准 POJO有 public 无参构造、字段有 getter/setter别用 Map 或 Object 这种模糊类型。8. 我个人的使用心得写了这么多 Flink 作业关于这三个算子我最想分享的其实是一条心法把语义清晰放在性能优化之前。新手容易走两个极端。一个是过度设计明明一个map能搞定的事非要拆成好几个算子或者反过来全塞进flatMap。另一个是过早优化还没跑起来就纠结并行度、序列化结果代码写得又复杂又难维护。我的做法是先用最直白的算子组合把逻辑写对跑通观察指标再针对真正的瓶颈优化。绝大多数情况下filtermapflatMap的直白组合性能已经足够好瓶颈往往在别的地方——数据倾斜、外部 IO、状态大小。另外一个小技巧在开发阶段给每个算子后面加一个临时的print()或计数累加器能快速定位数据在哪一步被过滤掉或展开错了。等逻辑稳定了再删掉。这个习惯帮我省了无数次翻日志的时间。最后说个选型的直觉判断如果你在写flatMap的时候发现collect只在一个分支里调用一次那大概率这个flatMap应该拆成filtermap。这个信号很准遇到时不妨停下来重构一下。 
长河君 · 2026/10/19 16:12:50 · 阅读 1.2k Flink 算子实战：map、filter、flatMap 的选型、组合与性能优化
1. 为什么这三个算子值得单独拎出来讲刚接触 Flink 的人十有八九是从map、filter、flatMap这三个算子入门的。它们看起来简单到不需要解释——map就是一进一出filter就是过滤flatMap就是一进多出。但我在实际带人做流处理任务的过程中发现恰恰是这三个最基础的算子藏着最多想当然的坑。举个真实的场景。之前有个做实时数据清洗的任务需求是把上游过来的原始日志做字段提取、无效记录剔除、再把嵌套结构打平。一个同事用map硬扛结果遇到脏数据直接抛异常整个作业重启另一个同事用flatMap加filter组合同样的数据量下吞吐反而更高容错也更稳。同样的功能写法不同线上表现天差地别。这篇文章就是想把这三个算子彻底讲透。不管你是刚学 Flink 的新手还是已经写过几个作业但总觉得能用但不够优雅的老手都能从里面找到能直接抄作业的东西。我会从它们各自的工作机制讲起然后重点说清楚什么时候该用哪个、组合使用时怎么设计、性能上有什么讲究最后给几段可以直接复现的代码。核心关键词就三个Flink 算子、map、flatMap外加一个绕不开的filter。需要先明确一点这三个算子都属于 Flink 的转换算子Transformation作用在DataStream上返回的也是DataStream。它们都是无状态的除非你手动引入状态也就是说每条数据独立处理互不影响。这个特性决定了它们特别适合做数据清洗、格式转换、字段映射这类逐条处理的活儿。理解了这一点后面的选型和优化才有根基。2. 三个算子的工作机制拆解2.1 map最纯粹的一进一出map的语义非常干净输入一条数据输出一条数据数量严格 1:1。你给它一个MapFunction它就把T类型的元素变成R类型的元素。DataStreamString lines env.fromElements(1,apple, 2,banana); DataStreamString names lines.map(new MapFunctionString, String() { Override public String map(String value) throws Exception { return value.split(,)[1]; } });从执行层面看map在算子链里是最轻量的。它不需要维护任何跨记录的状态每条数据进来调用一次map()方法结果直接往下游发。正因为如此map的吞吐能力非常强在算子链优化Operator Chaining开启的情况下它通常会和上下游算子合并成一个 Task减少线程间切换和序列化开销。但这里有个容易被忽略的点map里抛异常会怎样默认情况下一条数据在map中抛出异常会导致整个算子失败进而触发作业重启取决于你的重启策略。所以如果你在map里做split、类型转换这类可能失败的操作一定要想清楚容错方案——要么提前用filter把脏数据挡掉要么在map内部 try-catch 返回一个兜底值。我见过太多作业因为一条格式不对的日志就反复重启排查半天才发现是map里没做防御。2.2 filter只做减法不做变形filter的职责单一到极致判断每条数据是否保留返回true留下返回false丢弃。它不改变数据的类型输入T输出还是T。DataStreamString valid lines.filter(new FilterFunctionString() { Override public boolean filter(String value) throws Exception { return value ! null value.contains(,); } });很多人觉得filter没什么可讲的但实际用起来有两个细节值得注意。第一filter的返回值必须是布尔值不能返回 null。如果你写return value.length() 0 ? true : null;这种运行时会直接 NPE。这个错误在编译期不报跑起来才炸很隐蔽。第二filter和map的顺序会影响性能。假设你要先过滤掉无效数据再对有效数据做复杂转换那一定要先 filter 后 map。因为filter把数据量降下来之后map里那些耗时的解析、计算操作就少执行很多次。反过来先 map 后 filter等于对一堆马上要丢掉的数据白做了一遍转换。这个顺序问题在数据量大、脏数据比例高的时候性能差距能到几倍。2.3 flatMap一进多出的万能选手flatMap是三个里最灵活的。它的输入是一条数据输出是零条、一条或多条数据。注意这个零条——意味着flatMap其实可以同时干filter和map的活儿。DataStreamString words lines.flatMap(new FlatMapFunctionString, String() { Override public void flatMap(String value, CollectorString out) throws Exception { for (String w : value.split(,)) { if (w.length() 0) { out.collect(w); } } } });flatMap的核心是那个Collector。你通过调用out.collect()来决定往下游发多少条数据。不调用就是丢弃相当于 filter调用一次就是 map调用多次就是一对多展开。这种设计让flatMap特别适合处理一条记录里包含多个子元素的场景比如一行日志里有多个字段、一个订单里有多个商品、一段文本要分词。不过灵活是有代价的。flatMap因为要处理变长的输出内部实现比map稍微重一点而且如果collect调用次数过多会给下游造成压力。我一般的原则是能用 map 或 filter 明确表达的就不要用 flatMap 兜底。因为明确的算子语义更清晰也更容易被 Flink 的优化器识别和优化。只有当一进多出或可能零输出是真实需求时才用flatMap。2.4 三者的本质对比把三个算子放在一起看差异其实就集中在两个维度上输出数量和输出类型。算子输入→输出数量输出类型典型用途map1 → 1可变T→R字段提取、格式转换、类型映射filter1 → 0 或 1不变T→T数据清洗、条件筛选flatMap1 → 0/1/N可变T→R拆分、展开、分词、多路输出理解这张表选型基本就不会错。但选型只是第一步真正决定作业质量的是怎么把它们组合起来以及在组合时避开哪些坑。这才是下一部分要重点讲的。3. 选型与组合什么时候用哪个3.1 一个判断口诀先看数量再看类型我在实际项目里总结了一个简单的判断顺序基本能覆盖 90% 的场景。第一步问自己这条数据我可能不要吗如果可能丢弃那filter或flatMap是候选。如果每条都必须保留那map或flatMap是候选。第二步问自己输出的类型和输入一样吗如果一样且只是筛选用filter。如果类型变了且严格一对一用map。第三步只有当数量不固定或类型和数量都不固定时才轮到flatMap。这个顺序的好处是逼你先想清楚需求而不是一上来就抓flatMap这个万能工具。我见过不少代码明明就是简单的字段提取非要用flatMap写结果可读性差、性能也没优势。3.2 组合使用的经典模式实际任务里这三个算子几乎不会单独出现而是串成一条处理链。下面几种组合模式是我用得最多的。模式一先 filter 后 map做数据清洗。这是最常见的组合。上游数据往往有脏记录先用filter把明显不合法的挡掉再用map做字段解析和转换。顺序不能反原因前面说过——先降数据量再算省资源。DataStreamOrder orders raw .filter(s - s ! null s.split(,).length 4) .map(s - { String[] p s.split(,); return new Order(p[0], p[1], Double.parseDouble(p[2]), Long.parseLong(p[3])); });模式二flatMap 拆分 filter 二次筛选。当一条记录要展开成多条且展开后还要再筛一遍时这个组合很自然。比如日志按分隔符拆成多个事件再过滤掉其中不需要的事件类型。模式三map 做预处理 flatMap 做展开。有时候原始数据格式很乱先用map规整成统一结构再用flatMap展开。这样flatMap里的逻辑会干净很多不用一边解析一边展开。3.3 别为了少写算子牺牲语义有个常见的误区为了减少算子数量把filter和map的逻辑全塞进一个flatMap里。代码是短了但问题不少。一是可读性下降。一个flatMap里又判断又转换又展开别人接手时得逐行读才能搞懂在干嘛。二是调试困难。出问题时你分不清是过滤逻辑错了还是转换逻辑错了。三是优化受限。Flink 对独立的filter和map有更明确的优化空间混在一起反而不好处理。我的建议是除非性能压测证明合并算子有显著收益否则优先保证语义清晰。算子数量多一两个对性能的影响远没有想象中大但可维护性的差距是实打实的。4. 性能与并行度的那些讲究4.1 算子链为什么你的 map 和 filter 会合并成一个 TaskFlink 有个很重要的优化叫算子链Operator Chaining。默认情况下如果两个算子之间是forward或one-to-one的数据分发关系也就是不需要 shuffleFlink 会把它们合并到同一个 Task 里执行。这意味着你写的filter().map().flatMap()这条链很可能在运行时就是一个线程里顺序执行的三段逻辑中间没有网络传输也没有序列化反序列化。这是 Flink 性能好的关键原因之一。但这个优化有个前提算子之间的并行度必须一致。如果你给map单独设了setParallelism(4)而上下游是 2那它们就没法链在一起中间会插入网络 shuffle性能会明显下降。所以除非有明确的负载不均问题我一般不建议对map、filter、flatMap单独设并行度让它们跟着作业的全局并行度走就好。4.2 并行度设置的经验值那全局并行度怎么定这跟数据量、单条处理耗时、集群资源都有关。一个粗略的经验并行度 ≈ 峰值 QPS × 单条处理耗时秒 / 目标单并行度吞吐。举个例子假设峰值每秒来 10 万条数据每条在map里处理耗时 0.1 毫秒那单并行度理论能处理 1 万条/秒10 万条就需要至少 10 个并行度。实际还要留余量设 12 到 16 比较稳。但要注意map、filter这类无状态算子单并行度的处理能力其实很强瓶颈往往不在它们身上而在下游的聚合、join 或 sink。所以调并行度时先看整个链路的瓶颈在哪别盲目给map加并行度。4.3 序列化开销别在 map 里造大对象map和flatMap的输出如果要在算子之间传输比如后面接了keyBy就需要序列化。这时候如果你在map里 new 了一个巨大的对象序列化的开销会很可观。我踩过的一个坑在map里把整条原始日志几 KB塞进一个 POJO 往下传后面其实只用到了其中两个字段。结果序列化成了瓶颈。后来改成在map里就只提取需要的字段输出一个小对象吞吐直接上了一个台阶。原则很简单map 的输出尽量精简只带下游真正需要的字段。别图省事把整个原始对象往下传。4.4 异步与阻塞map 里别做同步 IOmap、filter、flatMap都是同步执行的。如果你在map里调用一个外部接口比如查数据库、调 HTTP那这个算子的吞吐会被 IO 延迟死死拖住。一条数据等 50 毫秒单并行度每秒就只能处理 20 条再多的并行度也救不回来。这种场景应该用 Flink 的异步 IOAsync I/O或者把外部查询结果提前广播Broadcast进来做本地关联。在map里做同步 IO 是新手最容易犯的性能错误之一一定要警惕。5. 踩坑实录那些年我在这三个算子上栽的跟头5.1 坑一filter 返回 null 导致的 NPE前面提过这里展开说。FilterFunction的filter方法签名返回boolean但如果你写的是包装类型Boolean并且返回了nullFlink 在自动拆箱时会抛 NPE。这个错误在本地小数据量测试时可能不出现因为没走到那条分支一上生产就炸。排查过程很典型作业日志里只有一句NullPointerException堆栈指向 Flink 内部看不出是哪条数据。后来在filter里加了日志才发现某条数据的某个字段为 null导致判断逻辑返回了 null。修复很简单保证filter永远返回基本类型boolean所有分支都有明确返回值。但教训是任何返回布尔值的判断逻辑都要考虑 null 的情况。5.2 坑二flatMap 里 collect 次数过多拖垮下游有个任务要把一条包含上千个元素的记录展开。一开始用flatMap一次性collect出上千条结果下游算子瞬间被压垮反压一路传到源头。问题在于flatMap的collect是同步的你 collect 多少条下游就得立刻处理多少条。如果一次展开的量太大会造成瞬时流量尖峰。解决办法有两个一是分批 collect比如每 100 条做一次小批量的输出控制二是在下游加缓冲或限流。更根本的思路是评估这个一对多是否真的合理能不能在更上游就把数据拆开。5.3 坑三map 里做重计算导致反压反压Backpressure是流处理里最常见的现象。有一次作业一直报反压排查发现是map里做了一段正则匹配而正则写得不够高效单条处理耗时波动很大。这里的关键认知是map 里的逻辑耗时必须稳定且可控。如果某条数据的处理耗时是其他数据的几十倍就会造成处理速率的抖动进而引发反压。解决办法是优化那段逻辑比如预编译正则、用更简单的字符串操作替代或者把重计算拆到独立的算子并单独调并行度。5.4 坑四误以为 flatMap 能替代所有场景前面反复强调过但还是值得单独列出来。flatMap灵活但灵活不等于合适。我见过有人用flatMap实现filter的功能代码里一堆if判断最后collect或者不collect。这种写法不仅可读性差而且因为flatMap内部要维护Collector的状态性能上也不如纯filter。选型的第一原则永远是语义匹配而不是哪个算子更万能。6. 可直接复现的完整示例6.1 环境与依赖下面这段代码基于 Flink 1.17Java 11。Maven 依赖只需要核心的两个dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version1.17.0/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-clients/artifactId version1.17.0/version /dependency6.2 一个完整的清洗任务需求模拟一批订单日志格式为订单ID,商品名,金额,时间戳。要做三件事——过滤掉字段数不对的脏数据、把金额从字符串转成 double、把商品名按逗号拆成多个标签。import org.apache.flink.api.common.functions.FilterFunction; import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.util.Collector; public class MapFilterFlatMapDemo { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(2); DataStreamString raw env.fromElements( 1001,apple|banana,19.9,1700000000, 1002,orange,29.9,1700000001, bad-data, 1003,grape|pear|peach,9.9,1700000002 ); // 第一步filter 挡掉脏数据 DataStreamString valid raw.filter(new FilterFunctionString() { Override public boolean filter(String value) { if (value null) return false; String[] parts value.split(,); return parts.length 4; } }); // 第二步map 做字段转换 DataStreamOrder orders valid.map(new MapFunctionString, Order() { Override public Order map(String value) { String[] p value.split(,); return new Order(p[0], p[1], Double.parseDouble(p[2]), Long.parseLong(p[3])); } }); // 第三步flatMap 把商品标签展开 DataStreamString tags orders.flatMap(new FlatMapFunctionOrder, String() { Override public void flatMap(Order order, CollectorString out) { for (String tag : order.getItems().split(\\|)) { if (!tag.isEmpty()) { out.collect(order.getOrderId() - tag); } } } }); tags.print(); env.execute(map-filter-flatmap-demo); } public static class Order { private String orderId; private String items; private double amount; private long ts; public Order() {} public Order(String orderId, String items, double amount, long ts) { this.orderId orderId; this.items items; this.amount amount; this.ts ts; } public String getOrderId() { return orderId; } public String getItems() { return items; } public double getAmount() { return amount; } public long getTs() { return ts; } } }跑起来你会看到bad-data被filter挡掉1001和1003的商品标签被flatMap展开成多条输出。这条链就是最典型的清洗—转换—展开三段式。6.3 用 Lambda 简化写法Java 8 之后这三个算子都支持 Lambda代码能短一大截。上面那段可以改写成DataStreamString valid raw.filter(s - s ! null s.split(,).length 4); DataStreamOrder orders valid.map(s - { String[] p s.split(,); return new Order(p[0], p[1], Double.parseDouble(p[2]), Long.parseLong(p[3])); }); DataStreamString tags orders.flatMap((Order o, CollectorString out) - { for (String tag : o.getItems().split(\\|)) { if (!tag.isEmpty()) out.collect(o.getOrderId() - tag); } });Lambda 写起来爽但有个坑泛型擦除。Flink 需要从函数签名里推断输出类型Lambda 有时候推断不出来会报InvalidTypesException。这时候要么显式加returns()要么老老实实写匿名内部类。我一般建议在类型复杂的地方用匿名类简单的地方用 Lambda别为了短而短。6.4 验证与观察跑完之后重点看两件事。一是输出条数是否符合预期——filter挡掉了几条、flatMap展开成了几条心里要有数。二是算子链是否生效——在 Flink Web UI 的作业图里如果filter、map、flatMap显示在同一个 Task 框里说明链化成功中间没有 shuffle。如果发现它们被拆成了不同的 Task检查一下是不是中间有keyBy、rebalance这类会触发 shuffle 的操作或者并行度设置不一致。7. 几个容易被忽略的进阶细节7.1 RichFunction需要生命周期时用它普通的MapFunction、FilterFunction是无状态的但如果你需要在算子启动时初始化一些资源比如加载字典、建立连接或者结束时释放资源就要用Rich 版本RichMapFunction、RichFilterFunction、RichFlatMapFunction。它们多了open()和close()两个方法分别在算子初始化时和结束时调用。典型用法是在open()里加载一份配置或字典然后在map()里直接用避免每条数据都去读一次。public class DictMap extends RichMapFunctionString, String { private transient MapString, String dict; Override public void open(Configuration parameters) { dict loadDict(); // 只加载一次 } Override public String map(String value) { return dict.getOrDefault(value, value); } }注意dict要声明为transient因为它不需要被序列化到各个并行子任务——每个子任务在open()时各自加载一份。7.2 累加器在 map 里统计指标Flink 提供了累加器Accumulator可以在map、flatMap里做计数、求和这类统计然后在作业结束后读取。这在排查数据质量问题时特别有用比如统计有多少条数据被filter挡掉了。private LongCounter dirtyCounter getRuntimeContext() .getLongCounter(dirty-records, new LongCounter());在filter里判断为脏数据时dirtyCounter.add(1)作业结束后就能拿到总数。比翻日志高效得多。7.3 类型信息别让 Flink 猜Flink 需要知道每个算子的输入输出类型才能做序列化和优化。大多数时候它能自动推断但遇到泛型嵌套、Lambda、或者自定义类型时推断可能失败。这时候用returns()显式声明DataStreamTuple2String, Integer result stream .map(...) .returns(Types.TUPLE(Types.STRING, Types.INT));显式声明类型还有个好处性能。Flink 对已知类型会用更高效的序列化器比如对 POJO 用字段级序列化比通用的 Kryo 快不少。所以自定义类型尽量写成标准 POJO有 public 无参构造、字段有 getter/setter别用 Map 或 Object 这种模糊类型。8. 我个人的使用心得写了这么多 Flink 作业关于这三个算子我最想分享的其实是一条心法把语义清晰放在性能优化之前。新手容易走两个极端。一个是过度设计明明一个map能搞定的事非要拆成好几个算子或者反过来全塞进flatMap。另一个是过早优化还没跑起来就纠结并行度、序列化结果代码写得又复杂又难维护。我的做法是先用最直白的算子组合把逻辑写对跑通观察指标再针对真正的瓶颈优化。绝大多数情况下filtermapflatMap的直白组合性能已经足够好瓶颈往往在别的地方——数据倾斜、外部 IO、状态大小。另外一个小技巧在开发阶段给每个算子后面加一个临时的print()或计数累加器能快速定位数据在哪一步被过滤掉或展开错了。等逻辑稳定了再删掉。这个习惯帮我省了无数次翻日志的时间。最后说个选型的直觉判断如果你在写flatMap的时候发现collect只在一个分支里调用一次那大概率这个flatMap应该拆成filtermap。这个信号很准遇到时不妨停下来重构一下。