Flink map、filter、flatMap 算子详解：原理、性能优化与避坑指南

1. 三条最基础的数据转换算子，为什么值得单独拎出来讲

刚接触 Flink 的人，十有八九是从map、filter、flatMap这三个算子入门的。原因很简单：它们足够直观，写起来跟 Java 8 的 Stream 几乎一模一样，学习曲线几乎为零。但恰恰是因为太熟悉，很多人会跳过对它们的深入理解，直接去啃窗口、状态、Checkpoint 这些“高级话题”，结果到了生产环境才发现，最基础的算子反而最容易出问题。

我自己在带团队做实时数据处理的时候，见过太多类似的场景：一个看起来人畜无害的map算子，在 QPS 上来之后成了整个作业的瓶颈；一个随手写的flatMap，因为返回了 null 导致下游莫名其妙报空指针；还有人把filter和map混着用，逻辑上没错，但性能差了一大截。这些问题不是靠调大并行度就能解决的，根子在于对算子本身的理解不够透彻。

这篇文章就是想把这三个算子彻底讲清楚。不是那种 API 文档式的罗列，而是从实际使用的角度出发，讲清楚它们各自适合什么场景、内部是怎么执行的、有哪些容易踩的坑、以及怎么组合使用才能既保证逻辑正确又保证性能。不管你是刚写完第一个 Flink 作业的新手，还是已经跑过几个生产作业的老手，相信都能从中找到一些之前忽略的细节。

需要说明的是，本文的示例代码以 Java 为主，因为 Flink 的 Java API 是使用最广泛的，但核心概念对 Scala 和 Python API 同样适用。另外，文中涉及的一些性能数据和调优建议，是基于我在实际项目中积累的经验，不同硬件环境和数据特征下可能会有差异，大家参考思路即可，具体参数还需要结合自己的场景去验证。

2. 三个算子的核心定位与设计思路拆解

2.1 为什么 Flink 要把这三个算子分开设计

很多人会问：既然flatMap可以返回 0 到 N 个元素，那它是不是可以完全替代map和filter？从功能上讲，确实可以。一个元素经过flatMap，返回 0 个就是过滤，返回 1 个就是映射，返回多个就是展开。那 Flink 为什么还要保留三个独立的算子？

这个问题的答案，要从两个层面去理解。

第一个层面是语义清晰度。map的语义是“一对一转换”，输入一个元素，输出一个元素，数量不变。filter的语义是“条件筛选”，输入一个元素，要么保留要么丢弃。flatMap的语义是“一对多展开”，输入一个元素，输出零个或多个元素。这三种语义在代码可读性上差异巨大。你看到一个map，立刻知道数据条数不会变；看到一个filter，立刻知道这是在做过筛；看到一个flatMap，就知道这里可能会改变数据条数。这种“一眼看出意图”的能力，在维护复杂作业的时候非常宝贵。

第二个层面是运行时优化空间。Flink 的运行时会对不同的算子做不同的优化。比如map算子，因为输入输出是一对一的，Flink 可以在算子链（Operator Chaining）阶段把它和上下游算子合并到同一个线程里执行，减少序列化和网络传输开销。而flatMap因为输出数量不确定，在某些场景下优化策略会有所不同。虽然这些优化对使用者来说是透明的，但理解背后的逻辑，有助于你写出更容易被优化的代码。

提示：在实际开发中，优先使用语义最贴切的算子。不要为了“少写几行代码”而用flatMap去实现filter的功能，这会让代码的可读性和可维护性都变差。

2.2 算子链机制对这三个算子的影响

说到 Flink 算子，就绕不开算子链（Operator Chaining）这个概念。简单来说，Flink 会把满足条件的多个算子合并成一个算子链，放在同一个 Task 线程里执行。这样做的好处是避免了算子之间的数据序列化和网络传输，能显著提升吞吐量。

map、filter、flatMap这三个算子，默认情况下都是可以参与算子链的。也就是说，如果你写了一个source -> map -> filter -> flatMap -> sink的流程，在满足条件的情况下，这一整条链可能会被合并成一个 Task 来执行。数据在算子之间传递时，不需要经过序列化和反序列化，直接在内存中传递对象引用，效率非常高。

但这里有一个容易被忽略的点：算子链的合并是有条件的。比如当两个算子之间的并行度不同，或者中间涉及keyBy、rebalance这类会触发数据重分布的操作时，算子链就会被断开。一旦断开，数据就需要经过序列化、网络传输、反序列化的过程，性能开销会明显增加。

我在实际项目中遇到过这样一个案例：一个作业的逻辑是kafka source -> map -> filter -> keyBy -> flatMap -> sink。因为keyBy的存在，map和filter被合并到了 source 所在的算子链，而flatMap则单独成了一个算子。后来我们发现flatMap里的逻辑其实可以提前到keyBy之前执行，调整之后，flatMap也被合并进了前面的算子链，整体吞吐量提升了将近 30%。这个例子说明，理解算子链的机制，能帮你写出性能更好的作业。

2.3 无状态特性带来的并行化优势

map、filter、flatMap这三个算子都是无状态算子。所谓无状态，就是说每个元素的处理都是独立的，不依赖于之前处理过的元素，也不需要维护任何跨元素的状态信息。

这个特性带来的最大好处就是并行化非常容易。因为元素之间没有依赖关系，Flink 可以放心地把数据分散到多个并行实例上去处理，每个实例只需要关注自己收到的元素即可。你只需要在算子后面调用.setParallelism(n)，就能轻松把处理能力扩展 n 倍。

但无状态也意味着它们的能力边界。如果你需要做聚合、去重、窗口计算这类需要“记住之前数据”的操作，就必须使用有状态算子，比如reduce、aggregate、process等。在实际作业中，通常是先用map、filter、flatMap做数据清洗和转换，然后再交给有状态算子做聚合计算。这种“先无状态清洗，后有状态计算”的模式，是 Flink 作业中最常见的架构。

3. map 算子：一对一转换的细节与实操要点

3.1 map 的基本用法与函数式写法

map算子的作用非常直接：对数据流中的每一个元素，应用一个转换函数，输出一个且仅一个结果元素。输入和输出的数量严格相等，不会多也不会少。

在 Java API 中，最基本的用法是实现MapFunction接口：

DataStream<String> input = env.fromElements("hello", "flink", "world"); DataStream<Integer> lengths = input.map(new MapFunction<String, Integer>() { @Override public Integer map(String value) throws Exception { return value.length(); } });

如果你用的是 Java 8 及以上版本，可以直接用 Lambda 表达式，代码会简洁很多：

DataStream<Integer> lengths = input.map(String::length);

Scala 和 Python 的写法更加简洁。Scala 中可以直接传一个匿名函数：

val lengths = input.map(_.length)

Python 中也是类似：

lengths = input.map(lambda s: len(s))

三种语言的写法虽然不同，但核心逻辑完全一致：接收一个元素，返回一个元素。选择哪种语言主要看团队的技术栈，功能上没有任何差异。

3.2 类型信息与 TypeInformation 的坑

这里要重点讲一个 Java API 中非常容易踩的坑：类型信息（TypeInformation）的推断。

Flink 在运行时需要知道数据的类型信息，用于序列化、比较、以及生成高效的执行计划。对于基本类型和常见的 POJO，Flink 可以通过反射自动推断出类型信息。但如果你在map中使用了泛型，或者返回了一个 Flink 无法自动推断的类型，就会出问题。

最典型的情况是这样的：

DataStream<Tuple2<String, Integer>> result = input.map(new MapFunction<String, Tuple2<String, Integer>>() { @Override public Tuple2<String, Integer> map(String value) throws Exception { return Tuple2.of(value, value.length()); } });

这段代码看起来没问题，但在某些情况下，Flink 会因为无法推断Tuple2中泛型的具体类型，而退化成使用 Kryo 序列化。Kryo 序列化虽然通用，但性能比 Flink 自带的序列化器差很多，而且可能在某些场景下出现兼容性问题。

解决办法是显式指定类型信息，有两种方式。第一种是在map之后调用returns方法：

DataStream<Tuple2<String, Integer>> result = input .map(new MapFunction<String, Tuple2<String, Integer>>() { @Override public Tuple2<String, Integer> map(String value) throws Exception { return Tuple2.of(value, value.length()); } }) .returns(Types.TUPLE(Types.STRING, Types.INT));

第二种是让MapFunction实现ResultTypeQueryable接口，在getProducedType方法中返回类型信息。这种方式适合需要复用的函数类。

注意：如果你在开发时开启了env.getConfig().disableGenericTypes()，那么当 Flink 无法推断类型信息时，作业会直接抛出异常而不是退化成 Kryo。这个配置在开发阶段非常有用，可以帮你提前发现类型信息的问题，避免上线后才暴露。

3.3 map 中的异常处理策略

map函数中抛出的异常，会直接影响作业的运行。默认情况下，如果map函数抛出异常，Flink 会根据重启策略来决定是否重启作业。如果没有配置重启策略，作业会直接失败。

在实际生产中，数据往往不是完全干净的，总会有一些脏数据或者边界情况导致map函数抛异常。这时候有几种处理策略：

第一种是在函数内部捕获异常，返回一个兜底值。比如解析 JSON 失败时，返回一个默认对象或者空对象。这种策略适合脏数据不影响整体业务逻辑的场景。

第二种是在函数内部捕获异常，记录日志后重新抛出。这种策略适合脏数据需要被感知和处理的场景，通过日志可以追踪到问题数据的来源。

第三种是使用侧输出流（Side Output），把处理失败的数据分流到另一个流中，主流程继续正常运行。这种策略适合需要对脏数据进行单独分析和处理的场景。

OutputTag<String> failedTag = new OutputTag<String>("failed"){}; SingleOutputStreamOperator<ParsedData> parsed = input .process(new ProcessFunction<String, ParsedData>() { @Override public void processElement(String value, Context ctx, Collector<ParsedData> out) { try { out.collect(parse(value)); } catch (Exception e) { ctx.output(failedTag, value); } } }); DataStream<String> failedData = parsed.getSideOutput(failedTag);

虽然这里用的是process而不是map，但思路是一样的。侧输出流是处理脏数据最优雅的方式之一，推荐在生产环境中使用。

3.4 map 的性能考量与并行度设置

map算子本身非常轻量，每个元素的处理都是独立的，没有状态维护开销。它的性能瓶颈通常不在算子本身，而在于map函数内部的逻辑。

我见过很多性能问题，根源都是map函数里做了太重的事情。比如在map里调用外部 HTTP 接口、查询数据库、读写文件等。这些操作都是同步阻塞的，会严重拖慢整个算子的处理速度。

如果确实需要在map中做外部调用，有几个优化方向。一是使用异步 IO，Flink 提供了AsyncDataStream来支持异步调用，可以显著提升吞吐量。二是批量处理，把多个元素攒成一批再调用外部接口，减少调用次数。三是本地缓存，对于变化不频繁的数据，可以在算子内部维护一个缓存，减少外部查询。

关于并行度的设置，map算子的并行度默认继承上游算子的并行度。你可以通过.setParallelism()单独设置。设置并行度时需要考虑数据源的分区数、下游算子的处理能力、以及集群的资源情况。并行度不是越大越好，过大的并行度会增加线程切换和资源调度的开销，反而可能降低性能。

4. filter 算子：数据筛选的逻辑与性能优化

4.1 filter 的基本用法与布尔逻辑

filter算子的作用是对数据流中的元素进行筛选，保留满足条件的元素，丢弃不满足条件的元素。它的核心是一个返回布尔值的判断函数。

Java API 中的基本用法：

DataStream<String> filtered = input.filter(new FilterFunction<String>() { @Override public boolean filter(String value) throws Exception { return value.startsWith("flink"); } });

Lambda 写法：

DataStream<String> filtered = input.filter(s -> s.startsWith("flink"));

filter的逻辑可以很复杂，包含多个条件的组合。比如同时判断多个字段：

DataStream<Order> validOrders = orders.filter(order -> order.getAmount() > 0 && order.getStatus() != null && !order.getStatus().equals("CANCELLED") );

这里有一个小技巧：把过滤条件按计算成本从低到高排列。比如order.getAmount() > 0只是一个数值比较，成本极低；而order.getStatus().equals("CANCELLED")涉及字符串比较，成本相对高一些。把成本低的判断放在前面，可以利用短路求值的特性，减少不必要的计算。

4.2 filter 与 map 的组合使用模式

在实际作业中，filter和map经常组合使用。一个典型的模式是：先filter掉不需要的数据，再map做转换。这样做的好处是减少后续算子的处理量。

但组合的顺序是有讲究的。假设你要从一批订单中筛选出金额大于 100 的订单，并计算它们的税额。有两种写法：

写法一：先 map 再 filter

orders.map(order -> new OrderWithTax(order, order.getAmount() * 0.1)) .filter(o -> o.getAmount() > 100);

写法二：先 filter 再 map

orders.filter(order -> order.getAmount() > 100) .map(order -> new OrderWithTax(order, order.getAmount() * 0.1));

两种写法结果一样，但性能不同。写法二更优，因为filter先减少了数据量，后续map需要处理的元素更少。如果map的计算成本很高，这个差异会非常明显。

提示：在算子链中，filter和map通常会被合并到同一个 Task 中执行，数据在它们之间传递不需要序列化。但即使如此，先filter后map仍然能减少map函数的调用次数，从而降低 CPU 开销。

4.3 filter 的并行度与数据倾斜问题

filter算子也是无状态的，可以自由设置并行度。但这里有一个容易被忽略的问题：数据倾斜。

假设你的数据流中，某个 key 的数据量特别大，而你的filter条件恰好对这个 key 的数据特别“友好”（大部分都保留），对其他 key 的数据特别“苛刻”（大部分都丢弃）。那么经过filter之后，不同并行实例处理的数据量会差异巨大，导致部分实例负载过高，成为整个作业的瓶颈。

解决数据倾斜的思路有几种。一是调整 filter 条件，尽量让筛选后的数据分布均匀。二是在 filter 之前做一次 rebalance，把数据均匀打散到各个并行实例。三是使用两阶段过滤，先做一次粗筛，再做一次精筛，中间通过 rebalance 重新分布数据。

DataStream<Event> result = input .rebalance() .filter(event -> event.isValid()) .filter(event -> event.getPriority() > 5);

rebalance会以轮询的方式把数据均匀分配到下游算子，能有效缓解倾斜问题。但rebalance本身会引入网络传输开销，所以只在确实存在倾斜时才使用。

4.4 filter 中返回 null 的隐患

这是一个非常隐蔽的坑：filter函数返回的布尔值如果是 null，会怎样？

在 Java 中，FilterFunction.filter方法的返回类型是基本类型boolean，不是包装类型Boolean。所以从编译层面就杜绝了返回 null 的可能。但如果你用的是 Scala 或者 Python，情况就不一样了。

在 Scala 中，如果filter函数的返回类型被推断为Boolean（包装类型），那么返回 null 时会在运行时抛出空指针异常。在 Python 中，filter函数返回 None 会被当作 False 处理，虽然不会报错，但逻辑上可能不符合预期。

所以无论用哪种语言，都要确保filter函数的返回值是明确的 true 或 false，不要出现 null 或 None 的情况。

5. flatMap 算子：一对多展开的核心机制

5.1 flatMap 的基本用法与 Collector 机制

flatMap是三个算子中最灵活的一个。它接收一个元素，可以输出零个、一个或多个元素。这种“一对多”的能力，让它能处理很多map和filter无法直接处理的场景。

Java API 中的基本用法：

DataStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() { @Override public void flatMap(String line, Collector<String> out) throws Exception { for (String word : line.split(" ")) { out.collect(word); } } });

注意flatMap方法的签名和map不同。它没有返回值，而是通过Collector对象来输出结果。你可以调用out.collect()任意多次，每次调用都会向下游发送一个元素。如果一次都不调用，就相当于过滤掉了这个元素。

Lambda 写法在 Java 中稍微麻烦一些，因为Collector是泛型接口，需要显式声明类型：

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

这里必须调用.returns(Types.STRING)，否则 Flink 无法推断 Lambda 表达式的返回类型，会退化成 Kryo 序列化。

5.2 flatMap 实现 filter 和 map 的组合效果

前面提到过，flatMap可以模拟filter和map的行为。比如要实现“筛选出长度大于 3 的单词，并转换为大写”，用filter+map需要两步：

DataStream<String> result = words .filter(w -> w.length() > 3) .map(String::toUpperCase);

用flatMap一步就能完成：

DataStream<String> result = words.flatMap((String w, Collector<String> out) -> { if (w.length() > 3) { out.collect(w.toUpperCase()); } });

虽然flatMap写法更紧凑，但我不推荐在简单场景下这么做。原因还是前面说的：语义清晰度。filter+map的组合，任何人一眼就能看出意图；而flatMap里的条件判断和转换逻辑混在一起，可读性会差一些。只有在逻辑确实需要“一对多”的时候，才应该使用flatMap。

5.3 flatMap 在数据展开与拆分中的应用

flatMap最典型的应用场景就是数据展开。比如：

• 一行文本拆分成多个单词
• 一个订单拆分成多个商品条目
• 一条日志拆分成多个字段
• 一个 JSON 数组拆分成多个独立对象

以订单拆分为例：

DataStream<OrderItem> items = orders.flatMap((Order order, Collector<OrderItem> out) -> { for (Product p : order.getProducts()) { out.collect(new OrderItem(order.getOrderId(), p.getProductId(), p.getQuantity())); } });

这种展开操作在 ETL 场景中非常常见。原始数据往往是大颗粒度的，需要拆分成小颗粒度才能做后续的聚合分析。

还有一个常见的场景是条件展开。比如根据订单类型，决定输出哪些字段：

DataStream<Event> events = orders.flatMap((Order order, Collector<Event> out) -> { out.collect(new Event("order_created", order.getOrderId())); if (order.isPaid()) { out.collect(new Event("order_paid", order.getOrderId())); } if (order.isShipped()) { out.collect(new Event("order_shipped", order.getOrderId())); } });

这种“一个输入产生多个不同类型输出”的场景，用flatMap处理起来非常自然。

5.4 flatMap 的性能陷阱与优化建议

flatMap因为输出数量不确定，在某些情况下会带来性能问题。

第一个陷阱是输出数量过大。如果一个元素的展开结果有成千上万个，那么flatMap会瞬间产生大量数据，可能导致下游算子来不及处理，或者内存溢出。这种情况下，需要考虑分批输出，或者调整下游算子的并行度。

第二个陷阱是在 flatMap 中做重计算。因为flatMap的输出逻辑比较灵活，有些人会把复杂的计算逻辑都塞进去，导致单个函数的执行时间过长。建议把计算逻辑拆分成独立的函数，保持flatMap本身的简洁。

第三个陷阱是Collector 的线程安全问题。Collector对象不是线程安全的，只能在flatMap方法内部使用，不能传递给其他线程。如果你在flatMap中启动了额外的线程来异步输出，会导致数据错乱甚至作业崩溃。

注意：flatMap中调用out.collect()的次数没有限制，但每次调用都会产生一次向下游的数据传递。如果输出数量非常大，建议先攒成一批再输出，或者考虑使用ProcessFunction配合定时器来做更精细的控制。

6. 三个算子的选型对比与组合实战

6.1 什么场景该用哪个算子

经过前面的分析，三个算子的适用场景已经比较清晰了。这里用一个表格做个总结：

选型的基本原则是：用语义最贴切的算子。需要转换就用map，需要筛选就用filter，需要展开就用flatMap。不要为了少写代码而混用，代码的可读性和可维护性比省几行代码重要得多。

6.2 组合使用的典型模式

在实际作业中，三个算子经常组合使用。这里分享几个我常用的组合模式。

模式一：清洗 -> 转换 -> 展开

DataStream<WordCount> result = lines .filter(line -> line != null && !line.trim().isEmpty()) .map(String::trim) .flatMap((String line, Collector<WordCount> out) -> { for (String word : line.split("\\s+")) { out.collect(new WordCount(word, 1)); } });

这是最经典的词频统计的前半段。先过滤空行，再去掉首尾空格，最后拆分成单词。

模式二：条件分流

DataStream<Event> validEvents = events .filter(e -> e.getType() != null) .flatMap((Event e, Collector<Event> out) -> { if (e.getType().equals("click")) { out.collect(enrichClick(e)); } else if (e.getType().equals("view")) { out.collect(enrichView(e)); } });

这种模式适合需要根据条件做不同处理的场景。虽然也可以用filter+map组合实现，但flatMap写起来更紧凑。

模式三：多阶段过滤

DataStream<Order> result = orders .filter(o -> o.getAmount() > 0) .filter(o -> o.getStatus() != null) .filter(o -> !o.getStatus().equals("CANCELLED")) .map(this::enrichOrder);

把多个过滤条件拆成多个filter，每个filter只负责一个判断。这样做的好处是每个filter的逻辑都很简单，便于单独测试和复用。而且 Flink 的算子链会把这些filter合并到一起，性能上不会有额外开销。

6.3 算子链断开对性能的影响

前面提到过算子链的概念，这里再展开说一下。算子链的合并与断开，对性能的影响非常大。

默认情况下，Flink 会尽可能把多个算子合并成一个算子链。合并的条件包括：上下游算子并行度相同、没有数据重分布操作、没有禁用算子链等。map、filter、flatMap之间如果没有keyBy、rebalance等操作，通常都会被合并到一起。

一旦算子链断开，数据就需要经过序列化、网络传输、反序列化的过程。这个过程的开销，根据数据大小和网络状况，可能是算子内部处理的几倍甚至几十倍。

所以，在写作业的时候，要尽量减少不必要的算子链断开。比如：

• 避免在map、filter、flatMap之间插入keyBy、rebalance等操作
• 如果确实需要重分布，考虑是否可以把重分布操作放到更靠后的位置
• 使用disableChaining()要谨慎，除非确实有必要

当然，算子链也不是越长越好。如果一个算子链太长，会导致单个 Task 的负载过重，不利于资源调度和故障恢复。通常建议单个算子链的算子数量不超过 10 个。

6.4 并行度设置与资源分配建议

并行度的设置是一个需要综合考虑的问题。对于map、filter、flatMap这三个无状态算子，并行度的设置主要考虑以下几个因素：

数据源的并行度。如果数据源是 Kafka，那么并行度最好不要超过 Kafka 的分区数，否则会有部分并行实例空闲。如果数据源是文件，并行度可以设置得大一些。

下游算子的处理能力。如果下游是有状态算子，比如keyBy+window，那么上游无状态算子的并行度要和下游匹配，避免上游产生数据的速度超过下游处理的速度。

集群资源。并行度越大，需要的 TaskManager 和 slot 就越多。要根据集群的实际资源来设置，不要盲目调大。

数据倾斜情况。如果数据分布不均匀，可以考虑适当增大并行度，让数据更均匀地分散到各个实例。

我的经验是，对于大多数场景，map、filter、flatMap的并行度设置为 Kafka 分区数的 1 到 2 倍比较合适。如果发现某个算子成为瓶颈，再单独调整它的并行度。

7. 常见问题排查与避坑经验实录

7.1 类型信息推断失败导致序列化异常

这是 Java API 中最常见的问题之一。典型的表现是作业运行时报InvalidTypesException，或者性能明显低于预期（因为退化成了 Kryo 序列化）。

排查思路：

1. 检查map、flatMap的返回类型是否包含泛型
2. 如果包含泛型，检查是否调用了.returns()显式指定类型
3. 检查自定义的 POJO 是否满足 Flink 的类型要求（有无参构造函数、字段是 public 或有 getter/setter）

解决办法就是显式指定类型信息。对于map，可以用.returns(Types.TUPLE(...))或.returns(MyPojo.class)。对于flatMap，Lambda 写法必须加.returns()，匿名内部类写法如果类型明确可以不加。

7.2 flatMap 返回 null 导致的空指针问题

flatMap中如果直接return null而不是通过Collector输出，会导致编译错误（因为方法返回类型是 void）。但如果在Collector中 collect 了 null，下游算子可能会报空指针。

// 错误写法 out.collect(null); // 正确写法：不输出任何元素 if (result != null) { out.collect(result); }

排查这类问题时，可以在flatMap之后加一个filter(Objects::nonNull)来兜底，但这只是治标不治本，根本的解决办法还是在flatMap内部做好空值判断。

7.3 算子链断开导致的性能下降

如果发现作业的吞吐量突然下降，或者某个算子的处理延迟明显增加，可以检查一下算子链是否被意外断开。

在 Flink 的 Web UI 中，可以看到每个算子的 Operator Chain 情况。如果发现map、filter、flatMap被分到了不同的 Task 中，就说明算子链断开了。

常见的断开原因包括：

• 中间插入了keyBy、rebalance、shuffle等操作
• 上下游算子的并行度不同
• 显式调用了disableChaining()或startNewChain()
• 算子之间设置了不同的 slot 共享组

解决办法就是根据实际情况调整算子顺序或并行度设置，尽量让可以合并的算子合并到一起。

7.4 数据倾斜的识别与处理

数据倾斜的表现是：大部分并行实例很快就处理完了，但少数几个实例一直处于忙碌状态，导致整个作业的完成时间被拖长。

识别方法：在 Flink Web UI 中查看各个 SubTask 的 Records Sent/Received 和 Bytes Sent/Received，如果差异超过 2 倍，就说明存在倾斜。

处理思路：

1. 如果是filter导致的倾斜，检查过滤条件是否对某些 key 特别“友好”
2. 如果是flatMap导致的倾斜，检查展开逻辑是否对某些输入产生了特别多的输出
3. 在倾斜算子之前加rebalance或rescale，重新分布数据
4. 如果是 keyBy 之后的倾斜，考虑使用两阶段聚合或加盐处理

7.5 常见问题速查表

7.6 几个容易被忽略的实操心得

最后分享几个我在实际项目中总结的小技巧。

技巧一：用filter替代flatMap中的条件判断。如果flatMap中只是简单地根据条件决定是否输出，而且输出内容就是输入本身，那么用filter会更清晰、更高效。

技巧二：在map之前做filter。这个前面提过，但值得再强调一次。先过滤再转换，能减少转换函数的调用次数，尤其是当转换逻辑比较复杂时，效果更明显。

技巧三：给flatMap的输出加一个上限。如果flatMap的展开结果可能非常大，建议加一个数量上限，避免单个元素产生过多输出导致内存问题。

技巧四：用name()给算子起个有意义的名字。Flink Web UI 中默认显示的是算子类型，比如Map、Filter。如果给算子起了名字，比如.name("parse-json")，排查问题时能快速定位到具体的算子。

技巧五：在开发环境开启disableGenericTypes()。这个配置能让类型信息推断失败的问题在开发阶段就暴露出来，而不是等到上线后才报错。

技巧六：用uid()给算子设置稳定的 ID。从 Savepoint 恢复作业时，Flink 依赖算子的 uid 来匹配状态。如果不设置 uid，修改作业逻辑后可能无法从 Savepoint 恢复。虽然map、filter、flatMap是无状态算子，但设置 uid 仍然是一个好习惯，尤其是当它们和上下游算子组成算子链时。

这些技巧看起来都是小细节，但在实际项目中，正是这些细节决定了作业的稳定性和可维护性。我在带新人的时候，通常会要求他们在第一个作业里就把这些习惯养好，后面能省很多事。