1. 从旁路输出到真正理解SideOutput做Flink开发的同学十有八九都遇到过这种场景主流程算得好好的突然来了几条异常数据比如JSON解析失败、水位线异常、或者业务上需要把某些特殊事件单独拎出来做监控。以前的做法是什么在主流里加if判断打到日志里或者再开一个Map算子把异常数据单独发出去。结果就是主流程代码越写越臃肿异常处理和正常业务逻辑搅在一起后面接手的人看得头皮发麻。这时候Flink的SideOutput旁路输出就能派上用场了。这个特性的核心价值在于在保持主数据流干净整洁的前提下把那些需要特殊处理的数据分流出去。如果说DataStream API里的split和select是早期Flink提供的分流工具那么SideOutput就是彻底替代它的更优解——split在Flink 1.13以后已经被标记为废弃原因很简单它没法在灵活性和类型安全之间取得平衡。这个demo本身其实不难核心也就三个环节定义OutputTag、调用sideOutputLateData或process函数里的ctx.output方法、最后用getSideOutput把分流出来的数据接住。但真正能把SideOutput用得顺手需要理解几个关键点OutputTag为什么必须是匿名的、SideOutput和分支流的区别、以及它在乱序数据处理中扮演的重要角色。所以这篇博客我会从这个demo出发把SideOutput的底层设计逻辑、典型使用场景、包括一些源码级别的细节都拆开讲清楚。适合谁看刚接触Flink、正在用DataStream API做实时计算的开发者以及那些已经在用SideOutput但碰到过为什么我的侧输出流收不到数据这类问题的同学。后面的内容里有我对各种奇怪问题的排查记录可以少走不少弯路。2. SideOutput背后的设计逻辑为什么需要它2.1 从split说起旧方案到底差在哪在正式讲SideOutput之前有必要先聊聊它的前身split算子这样能更好地理解SideOutput的设计取舍。split的用法很简单调用DataStream.split()把流拆成多个逻辑分支然后用select()选择其中一个分支继续处理。但这里有个致命伤——split只是给数据打上了逻辑标签并没有真正物理拆分数据流。这意味着每条记录仍然会流经所有后续算子只是通过标签决定是否被下游选中。换句话说它只是看起来分流了效率和隔离性都很差。更麻烦的是类型安全的问题。split基于字符串标签做筛选标签打错了只能等到运行时才能发现。而SideOutput通过泛型OutputTag在编译期就限制了输出类型类型不匹配直接编译报错根本不会给你留到运行时踩坑的机会。还有一个很多人没注意到的问题split只能用在DataStream上在KeyedStream的处理过程中想做主干和分支分离split是做不到的。SideOutput则灵活得多在ProcessFunction里的任意位置都能调用ctx.output()也就是说你在keyBy之后的状态管理、定时器回调里都可以随时分流。2.2 OutputTag与类型安全机制OutputTag是SideOutput的核心标识符。它的定义方式看起来有点绕习惯Java的朋友一眼就能认出来这是匿名内部类的写法OutputTagString lateTag new OutputTagString(late-data) {};注意末尾那对花括号不能丢。这个花括号的意义在于创建了一个OutputTag的匿名子类而Flink内部依赖TypeExtractor来处理泛型信息时这个匿名子类的存在能让Flink拿到完整的TypeInformation。如果去掉花括号某些场景下会遇到类型信息丢失导致序列化异常。OutputTag本质上带有两个信息一个字符串ID和完整的类型信息。ID用于标识侧输出流类型信息用于保证写入和读取时的类型一致。在Flink内部侧输出流的数据会被打上相应的标签并存放在OutputTag对应的StreamRecord中。当执行getSideOutput时Flink根据标签匹配把对应流的数据取出来。2.3 两条流分离的关键flush与lazy这里说一个理解SideOutput的重要视角SideOutput不是把数据从一个流移动到另一个流而是在算子输出的时候让数据同时带着主输出和侧输出两个去向。ProcessFunction的processElement方法参数里的Context对象带有一个output(X tag, X value)方法就是用来发射到侧输出流的。主输出和侧输出并不是二选一的关系完全可以同时输出。比如一条数据既发到主流做常规聚合又发到侧流做实时监控这是经常见到的用法。但需要说明的是侧输出流并不会自动地被下游算子处理必须在后续某个环节显式调用getSideOutput拿到侧流再接算子否则侧输出流的数据就会被丢弃。3. 核心场景拆解SideOutput最合适的三个位置3.1 处理迟到数据窗口计算的救命稻草这是SideOutput最经典、使用率最高的场景。Flink窗口计算中数据乱序是家常便饭。通常我们设置WatermarkStrategy来控制等多久但无论水位线设得多合理总有数据会迟到。这时候如果直接丢弃结果就有偏差如果全部允许迟到窗口又迟迟不能关闭。用SideOutput就优雅了主流的窗口计算结果照常下发迟到的数据通过sideOutputLateData发送到侧输出流由业务决定怎么处理——单独存储、重新计算、或者监控告警。如下图所示的处理逻辑DataStreamSensorReading lateStream windowedStream .sideOutputLateData(lateTag) .process(new MyProcessFunction());有些同学一开始可能会觉得意外sideOutputLateData是在WindowedStream上调用的不是在DataStream上。这意味着它在窗口逻辑结束后额外开了一条侧流专门承接晚于水印到达的数据。3.2 异常分流与监控链路解耦另一类非常常见的用法是异常数据分流。以一个实时解析Kafka消息的作业为例消息体偶尔会解析失败或者字段缺失、格式异常。如果这些解析失败的消息直接丢弃出了问题很难排查如果全部写入主流正常业务逻辑又会被污染。我的做法是在ProcessFunction里判断解析结果正常数据走主流异常数据通过侧输出流单独接入一个监控链路。侧流可以连接计数算子做阈值告警也可以直接输出到Kafka的dead letter topic方便后续分析问题根因。if (parsedValue null) { ctx.output(dirtyDataTag, originalMessage); return; }这种模式的核心优势是让主流的正常处理逻辑保持简单。监控需求是独立于业务需求的把它们揉在一起会让主流程的复杂度成倍上升。3.3 多级分流在一个流里实现复杂路由有些业务场景需要根据数据内容把流分拆成多个子流。比如订单消息需要区分新订单已取消已退款等多种状态。用SideOutput来做的思路是定义多个OutputTag在ProcessFunction里根据业务规则把数据发到不同的侧输出流主流甚至可以只保留一个默认出口或者什么数据都不发。这里有一个小设计上的建议如果只是简单的维度拆分比如一个枚举字段的不同取值不要用SideOutput硬拆。filter算子更适合这种场景因为它天然就是一个筛选动作。SideOutput更适合那些主流程继续走分支流程作为旁路处理的场景语义上更清晰。4. 实操完整实现一个SideOutput的Demo4.1 环境准备与版本选择下面是完整的demo代码。我先说明一下环境Flink 1.17版本Java 11使用DataStream API的ProcessFunction实现。版本选1.17的原因是它已经稳定支持新API比如统一的状态存储、checkpoint优化等同时保留了经典DataStream写法。Maven依赖dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version1.17.1/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-clients/artifactId version1.17.1/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-kafka/artifactId version1.17.1/version /dependency4.2 主流程窗口聚合 迟到数据侧流为了演示效果我构造了一个传感器温度监控的典型场景每隔一段时间读取一批温度数据计算窗口内的平均温度把迟到的数据单独输出到侧流。import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner; import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.typeinfo.Types; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction; import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows; import org.apache.flink.streaming.api.windowing.time.Time; import org.apache.flink.util.Collector; import org.apache.flink.util.OutputTag; import java.time.Duration; public class SideOutputDemo { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); // 为演示延迟效果降低并行度到1 env.setParallelism(1); // 定义迟到数据的侧输出标签 OutputTagSensorReading lateReadingTag new OutputTagSensorReading(late-readings) {}; // 模拟数据源生成传感器读数time为事件时间 DataStreamSensorReading inputStream env .fromElements( new SensorReading(sensor_1, 1000L, 35.5), new SensorReading(sensor_1, 3000L, 36.0), new SensorReading(sensor_1, 5000L, 36.5), new SensorReading(sensor_2, 2000L, 37.0), // 这条数据事件时间远小于水位线会触发迟到逻辑 new SensorReading(sensor_1, 500L, 34.8) ) .assignTimestampsAndWatermarks( WatermarkStrategy.SensorReadingforBoundedOutOfOrderness(Duration.ofSeconds(2)) .withTimestampAssigner((event, timestamp) - event.timestamp) ); // 按传感器ID分组开10秒滚动窗口 SingleOutputStreamOperatorDouble avgTempStream inputStream .keyBy(r - r.sensorId) .window(TumblingEventTimeWindows.of(Time.seconds(10))) .sideOutputLateData(lateReadingTag) .process(new ProcessWindowFunctionSensorReading, Double, String, TimeWindow() { Override public void process(String key, Context context, IterableSensorReading elements, CollectorDouble out) { double sum 0.0; int count 0; for (SensorReading r : elements) { sum r.temperature; count; } out.collect(sum / count); } }); // 取出侧输出流 DataStreamSensorReading lateStream avgTempStream.getSideOutput(lateReadingTag); // 打印 avgTempStream.print(avg); lateStream.print(late); env.execute(SideOutput Demo); } public static class SensorReading { public String sensorId; public long timestamp; public double temperature; public SensorReading() {} public SensorReading(String sensorId, long timestamp, double temperature) { this.sensorId sensorId; this.timestamp timestamp; this.temperature temperature; } Override public String toString() { return SensorReading( sensorId , timestamp , temperature ); } } }这段代码的流程拆解如下先构造传感器数据并分配时间戳和水位线用forBoundedOutOfOrderness(Duration.ofSeconds(2))声明最多允许2秒的乱序。然后按传感器ID分组开滚动窗口计算窗口平均值把迟到的数据通过sideOutputLateData分流到侧输出流最后在主流的avg和侧流的late上分别打印输出。4.3 调试与执行观察迟到数据流向运行上面这段代码你会看到两条打印流avg 35.875 late SensorReading(sensor_1, 500, 34.8)这里有一条很有意思的细节sensor_1在500毫秒的数据为什么会迟到因为水位线推进到了当前最大事件时间减去2秒的位置。在最后一条数据到达、水位线推到3000毫秒最大事件时间5000L减去允许乱序2秒以后窗口[5000, 10000)并没有被这条500毫秒的数据触发因为数据时间已经远小于窗口边界了。所以它被判定为迟到数据进入侧流。而窗口内的数据1000、3000、5000计算出了平均温度35.875。这个调试过程可以很好地帮你理解水位线机制和迟到的判定依据数据是否迟到取决于它的事件时间与水位线的比较结果而不是数据到达的时刻。4.4 ProcessFunction手动输出的另一种实现除了在窗口流上直接调用sideOutputLateData更通用的写法是在ProcessFunction里手动调用ctx.output()。这种方式更灵活可以用在任意自定义逻辑中不需要依赖窗口的存在。OutputTagString errorTag new OutputTagString(invalid-data) {}; DataStreamString source ...; SingleOutputStreamOperatorString processed source .process(new ProcessFunctionString, String() { Override public void processElement(String value, Context ctx, CollectorString out) { try { // 模拟解析 String parsed parse(value); out.collect(parsed); } catch (Exception e) { ctx.output(errorTag, value); } } }); DataStreamString errorStream processed.getSideOutput(errorTag);注意ProcessFunction的Context和OnTimerContext都有一个output方法后者在定时器回调里也可以发侧流。这在需要基于时间触发的告警场景里特别有用比如窗口关闭后把没有达到阈值的统计信息输出到侧流。5. 容易踩的坑和排查思路5.1 侧输出流没数据先检查这三个原因用SideOutput最让人头疼的问题就是代码写了侧流却始终为空。按我的排查顺序一般是按下面三个方向来找第一个检查点OutputTag的ID是不是重复了。如果同一个算子里定义了多个ID相同的OutputTag侧流解析会互相覆盖。不同用途的侧流必须用完全不同的ID。这个问题在简单demo里不会出现但在代码里大量使用SideOutput时会容易疏忽。第二个检查点getSideOutput用错了对象。有些同学在窗口流上调用sideOutputLateData后直接拿原来的windowedStream去getSideOutput——这是不行的。getSideOutput必须作用在返回的那条SingleOutputStreamOperator上也就是包含主流输出结果的流上而不是原始输入流。第三个检查点水位线没有推进或者推进速度太慢。如果水位线一直停在初始值所有稍晚到达的数据都会进入侧流反过来如果数据量太少导致水位线没有触发窗口关闭主流和侧流可能都暂时没有输出。测试时建议加大数据量或者用assignAscendingTimestamps严格递增时间戳来简化验证。5.2 序列化异常与OutputTag花括号在执行作业时有时会遇到典型的序列化异常InvalidTypesException或者“The generic type information of the OutputTag is missing”这类错误。根本原因是OutputTag定义时缺少了匿名内部类。强调一下这个定义方式// 正确写法带花括号 OutputTagString tag new OutputTagString(my-tag) {}; // 错误写法不带花括号可能丢失类型信息 OutputTagString tag new OutputTag(my-tag);前一种是Flink官方推荐且内部依赖的方式。花括号创建了匿名子类让Flink的TypeExtractor能从类签名里获得泛型类型。Java的泛型擦除会让普通实例在运行时无法知道String类型而匿名子类保留了这个信息。类似的情况还出现在继承RichFunction并重写open方法引用外部变量时瞬态字段缺失也会导致序列化失败。但属于另外一个问题了不在今天讨论范围。5.3 并行度对侧流输出顺序的影响如果并行度大于1侧输出流的数据顺序就不再保证全局有序这和主流的并行处理逻辑一致。在测试阶段如果你需要精确验证输出顺序可以先设置env.setParallelism(1)就像我前面demo里做的那样。生产中如果对顺序有要求需要在业务逻辑里靠事件时间排序而不能依赖Flink的流顺序。另外侧输出流的数据量和主流的吞吐量共享同一个算子出口如果侧流输出数据量巨大会对背压产生一定影响。尤其是异常数据的全量转发场景需要评估侧流对整体作业吞吐的影响。5.4 状态后端与checkpoint对侧流的影响最后补充一个容易被忽视的细节侧输出流的数据会不会进入状态后端实际上侧输出流本身不占用checkpoint状态空间它的数据是直接发射到下游的。但如果你在侧流上再做keyBy和窗口操作那这些算子才占有状态。理解这一点很重要——侧流的成本主要在下游算子的状态大小而不是侧流本身。所以说对侧流做长时间窗口的聚合要谨慎评估状态量大小。特别是从Kafka消费后做过一次SideOutput分流侧流再跑一个长窗口去重或累积很可能会吞噬大量堆外内存。6. 从Demo到生产落地我的一些经验谈写到这里SideOutput的核心概念、用法和坑基本都覆盖了。最后说几个我从多次实战中总结出来的具体建议。关于侧流数量的控制一个并行子任务里你有多少条侧流在实时计算引擎的指标面板里都能看到。建议一个作业里侧流数量控制在个位数太多会让作业拓扑变得很难梳理也增加监控成本。关于迟到数据的处理策略很多人把所有迟到数据交给侧流输出就完事了这其实只是保留证据的层面。更完整的做法是侧流数据落到一个单独的Kafka topic或者存储介质里再配合一套定时任务去做数据补偿。比如计算窗口已经关闭但因为迟到数据需要修正结果这种情况就要做结果修正流。我见过不少团队用Flink CEP做延迟数据补偿实际上用SideOutput把迟到数据缓冲起来每天或者每小时批量回算一次反而更简单可控。另外配合Flink的Web UI诊断侧流问题很有用。作业提交后打开Metrics面板可以看到侧输出流相关的numRecordsOut等指标。如果侧流正常但数据量过多指标曲线会明显异常——经常是水位线设置过高导致的。合理设置水位线本身又是一个大学问这里不展开但要记住侧流数据量过大的时候优先检查watermark策略。7. 结尾给正在学SideOutput的人一句话我个人学习SideOutput的过程其实挺曲折的。最开始看官方文档觉得很简单不过就是一个输出标签加一个方法调用。真正集成到复杂业务里才发现水位线的微妙影响、类型信息的细节、并行度下的顺序问题这些才是真正需要花时间去理解的地方。所以这个demo虽然短但它麻雀虽小五脏俱全。把这段代码跑通之后强烈建议你往里面再塞两个场景试试一是把侧流数据再连到另一个窗口做统计二是用ProcessFunction手动实现侧流而不是用sideOutputLateData。自己动手改一改比看十篇博客都管用。踩过这几个坑之后SideOutput在你的工具箱里就真的变成顺手的那一把工具了。