1. 侧输出流到底是什么为什么要关注它做实时计算的同行应该都有过这种体验一条主流数据进来情况太复杂了既想保留主链路继续往下游算又想把某些特殊数据单独拎出来处理。以前的做法无非是filter两遍、split分流或者干脆复制流再处理代码难看不说性能损耗也让人头大。直到Flink的侧输出流Side Output出现这个场景才有了一个像样的标准解法。侧输出流可以理解成给主数据流打了一个“侧门”。它允许你在处理主流的任意节点上通过一个带标签的输出口把符合特定条件的数据投递到另一条独立的旁路流里。旁路流和主流共用同一个作业拓扑但下游算子各自独立互不干扰。你不需要复制数据流也不需要额外维护一套分发逻辑代码结构自然就清爽了。我最早接触这个特性是因为一个实时风控项目正常的订单数据要进指标计算但可疑订单需要立刻拉到旁路做人工审核。当时用filter写了两套逻辑主流和旁流各算一遍状态还不好共享调试起来格外费劲。后来翻官方文档看到侧输出流一下子思路就通了——主链路按原有逻辑继续走异常订单通过侧输出流单独成流下游接一个独立的告警算子就好。这个特性适合谁正在用Flink DataStream API做实时数仓、实时风控、实时推荐的同学特别是遇到“主流逻辑必须保持完整但部分数据需要差异化处理”这类场景的时候基本就是为你量身定做的。还有就是想把“脏数据隔离”和“业务主链路”彻底解耦的团队侧输出流能帮你省掉很多不必要的Kafka topic和中间逻辑。2. 侧输出流的设计思路拆解2.1 为什么主流需要“旁路”能力实时计算里一个非常尴尬的场景是数据种类杂、质量参差不齐但主业务又要求一个统一处理入口。举个例子你接了一个用户行为日志的Kafka主题里面既有正常的浏览、点击事件也有格式错误、字段缺失或者明显超时的事件。主链路只关心正常事件异常事件又不能直接丢掉——审计要用、重放要用、监控也要用。传统做法是filter分流正常事件filter出去算A异常事件filter出去算B最后两路再分别处理。这个方案能跑但问题很明显——每条数据都被遍历了多遍开销白白翻倍而且一旦主逻辑需要改所有分支都要跟着动维护成本直线上升。侧输出流的思路完全不一样你只在主处理逻辑里判断一次正常的走return或者collector往下游发射异常的通过ctx.output(tag, value)扔到侧输出流。它不改变主流的数据走向只是额外开启了一条“溢出通道”。2.2 对比split、filter和侧输出流Flink老版本有个split算子专门做分流。但split有个硬伤split之后的下游算子只能通过select(tag)选择某一路而且split后的流无法直接作为主流继续做多级处理这种“分裂式”的设计让它在复杂拓扑里非常笨重。Flink 1.12之后split已经被标记废弃官方推荐就是侧输出流。filter分流的问题在于N路分支就需要N个filter算子每条数据要被完整遍历N遍。实时作业的瓶颈很多就在这种重复计算上数据量一上来CPU消耗就藏不住了。侧输出流的优势很清晰单次处理、单次判断想分多少路就分多少路且每一路都是真正的DataStream可以继续connect、keyBy、窗口计算、sink到任意存储。它等于把“分流”和“主处理”这两件事在同一个算子内合并了性能和代码可读性都占优。2.3 侧输出流的典型适用场景从我接触过的真实需求来看侧输出流在下面几类场景里是真正的“杀手锏”迟到数据兜底事件时间窗口关闭后迟到的数据直接进侧输出流单独攒起来做延迟统计或修正计算不污染主窗口的结果。脏数据隔离反序列化失败或者字段校验不通过的数据统一进侧输出流落盘或者发到告警队列主链路永远干净。超时事件检测比如订单15分钟未支付通过定时器触发后把订单ID输出到侧流下游接短信或者推送服务。多级分流处理同一份数据既要做核心指标计算又要拆分出不同类型的明细进入不同的存储侧输出流能做到一份输入、多路输出。3. 核心API与底层原理3.1 OutputTag、process函数和Context先记住三个关键要素OutputTag是侧输出流的“身份证”process function是触发侧输出的地方Context是拿到侧输出出口的钥匙。OutputTag的定义方式如下// 侧输出流标签必须匿名内部类泛型是侧流数据类型 final OutputTagString abnormalTag new OutputTagString(abnormal) {};这里有个细节值得注意OutputTag要求用匿名内部类的方式创建因为Flink需要利用TypeInfoGeneration来保留泛型类型信息。如果你直接new OutputTagString(abnormal)不带花括号类型信息会被擦除运行时就可能报类型相关的异常。侧输出必须在ProcessFunction或者KeyedProcessFunction里完成因为只有这两个函数能拿到Context。普通FlatMapFunction和MapFunction不具备这个能力。伪代码大概长这样DataStreamOrder mainStream source .process(new KeyedProcessFunctionString, Order, Order() { Override public void processElement(Order value, Context ctx, CollectorOrder out) throws Exception { if (value.isAbnormal()) { // 侧输出到异常订单旁路 ctx.output(abnormalTag, value.getOrderId()); } else { // 正常订单走主流 out.collect(value); } } }); // 获取侧输出流继续做下游处理 DataStreamString abnormalStream mainStream.getSideOutput(abnormalTag);processElement的第三个参数out是主输出流存的是正常数据。侧输出流不经过out而是通过ctx.output打到指定的tag上。这里有一个容易混淆的细节getSideOutput是从处理后的主流上获取不是从source上直接获取必须先对process之后的结果流调用getSideOutput。3.2 为什么是ProcessFunction而不是MapFunction很多刚接触侧输出流的同学会问我就是在map函数里做个判断想顺手分个流为什么不行原因在于MapFunction的接口设计——它的map方法只有一个out参数Collector没有暴露Context。而ProcessFunction的processElement方法把Context一并交给了开发者Context内部持有OutputTag到Output的映射关系。这个设计是有意为之的Flink把“主流输出”和“旁路输出”的能力边界划得很清楚只有最强的ProcessFunction层级才具备旁路能力避免API被滥用。实际开发中如果需要做状态读写、定时器注册那就必须用KeyedProcessFunction如果纯粹就是做个分流不做状态操作也可以用ProcessFunction配合算子状态。总之从ProcessFunction起步是标准姿势。3.3 侧输出流的技术实现旁路输出如何不干扰主链路从实现层面看侧输出流的数据和主输出流的数据在同一个算子内生成但走的通道是分开的。算子内部维护了一个MapOutputTag, Output的结构每个tag对应一个独立的输出缓冲区。数据在主输出和侧输出之间是“各走各的”不会把旁路的数据灌回主流。这也意味着下游如果只看主流数据完全感知不到侧输出流的存在只有显式调用getSideOutput(tag)的算子才会拿到旁路数据。这种“惰性取流”的设计让旁路输出的附加成本几乎为零——不需要处理旁路的数据就不产生额外的序列化、网络传输开销。对比之前的filter方案性能差别在这种场景下非常明显。4. 侧输出流的实战订单超时监控与异常隔离4.1 场景设定实时订单超时监控我以订单支付超时监控为例拆解一个完整流程。假设订单系统会把创建事件实时发到Kafka字段包括订单号、用户ID、创建时间、订单金额、订单状态等。业务需求是订单创建后15分钟内未支付需要触发超时告警正常的支付事件继续走主链路做GMV统计。这个需求有两个关键点一是“超时”是个时间维度概念需要为每个订单注册定时器二是超时告警不能打扰主链路的统计逻辑。侧输出流配合KeyedProcessFunction几乎是这个场景最顺滑的解法。4.2 完整代码拆解基于事件时间和处理时间的定时器设计如果订单创建时间可以作为事件时间我们可以按事件时间注册定时器。但如果数据延迟到达事件时间定时器可能早就触发了所以很多生产环境里超时监控用的是处理时间。下面用处理时间版本演示。public class OrderTimeoutMonitor extends KeyedProcessFunctionString, Order, Order { // 侧输出Tag超时未支付订单 private final OutputTagString timeoutTag new OutputTagString(order-timeout) {}; // 侧输出Tag数据异常 private final OutputTagOrder errorTag new OutputTagOrder(order-error) {}; private ValueStateLong timerState; Override public void open(Configuration parameters) throws Exception { timerState getRuntimeContext().getState( new ValueStateDescriptor(timer-state, Long.class)); } Override public void processElement(Order value, Context ctx, CollectorOrder out) throws Exception { // 字段校验脏数据进错误旁路 if (value.getOrderId() null || value.getCreateTime() null) { ctx.output(errorTag, value); return; } // 订单状态为待支付注册15分钟后的定时器 if (CREATED.equals(value.getStatus())) { long timeoutTime ctx.timerService().currentProcessingTime() 15 * 60 * 1000L; ctx.timerService().registerProcessingTimeTimer(timeoutTime); timerState.update(timeoutTime); out.collect(value); } else { // 已支付等状态直接走主流 if (timerState.value() ! null) { ctx.timerService().deleteProcessingTimeTimer(timerState.value()); timerState.clear(); } out.collect(value); } } Override public void onTimer(long timestamp, OnTimerContext ctx, CollectorOrder out) throws Exception { // 定时器触发说明订单仍未支付 ctx.output(timeoutTag, ctx.getCurrentKey()); } }这里有几个要点我踩过坑值得单独说明定时器注册后如果订单在超时前支付了需要删除定时器否则会产生无效告警。所以我把定时器时间戳存到了ValueState里支付事件进来时取出并删除。为什么用ValueState而不是直接把定时器时间戳存在本地变量因为定时器回调时可能在不同的实例上执行作业重启、并行度变化等本地变量不持久状态才能保证可靠性。KeyedProcessFunction的onTimer触发时通过ctx.output可以拿到侧输出这里利用它输出超时订单。4.3 下游消费侧流多路输出对接不同存储拿到了主处理和旁路处理的结果后下游各自消费DataStreamOrder mainStream orders .keyBy(Order::getUserId) .process(new OrderTimeoutMonitor()); // 主流继续做统计 DataStreamOrder normalOrders mainStream .filter(order - PAID.equals(order.getStatus())); normalOrders .keyBy(order - order.getUserId()) .timeWindow(Time.hours(1)) .aggregate(new GmvAggregate()) .addSink(gmvClickhouseSink); // 超时侧流直接告警 DataStreamString timeoutStream mainStream.getSideOutput(timeoutTag); timeoutStream.addSink(new AlertSink()); // 异常侧流进Kafka死信队列 DataStreamOrder errorStream mainStream.getSideOutput(errorTag); errorStream.addSink(new KafkaDeadLetterSink());注意一个细节getSideOutput之后返回的是一个全新的DataStream可以继续做keyBy、加窗口、做聚合能力上跟主流没有任何区别。这里超时流没有再keyBy是因为告警只需要把订单信息发出去如果要按用户维度统计超时次数那就继续keyBy。4.4 实战心得从一次“告警风暴”里学到的经验我第一次跑这个作业的时候上线第一天就爆发了告警风暴几乎所有待支付订单在15分钟后都触发了超时告警。排查下来发现是开发环境测试数据的时间戳用的历史时间处理时间定时器一注册当前时间减去事件时间早已超过15分钟立刻触发。这提醒了我超时监控类的需求要明确用处理时间还是事件时间二者在测试和生产环境的数据形态上会有很大差异。如果你确定要按事件时间来算超时有一个额外注意事项Flink中的事件时间定时器依赖watermark触发而watermark的推进意味着数据有序。对乱序严重的订单数据事件时间超时容易出现“该触发的不触发”的怪现象排查难度远比处理时间大。非严格需要建议先用处理时间。5. 侧输出流的进阶技巧与常见坑5.1 OutputTag的序列化和类型擦除问题OutputTag必须用匿名内部类创建这件事很多同学不理解。Flink在运行时通过反射获取匿名内部类的泛型父类信息来推断侧流的数据类型。如果你直接new OutputTagString(tag)泛型信息在编译后就是Object侧流拿到数据后无法正确反序列化作业会报类型相关的异常而且这类异常通常在运行时才出现排查成本不低。多个侧流可以共用一个OutputTag吗不行。每个侧流必须有独立的tag实例tag的id字段字符串要全局唯一。如果两个侧流共用一个taggetSideOutput的时候只能拿到其中一个而且作业可能直接报冲突。我习惯在id命名上用“业务-用途”的格式比如order-timeout、order-error一眼能看懂。5.2 侧输出流是否参与checkpoint、如何保证一致性侧输出流本身不额外消费状态但写入侧流的数据和主流的处理在同一个operator里checkpoint机制的保障是处理到某条数据时产生的所有输出包括主流和侧流作为一个整体被记录在checkpoint中。也就是说只要你的作业开了checkpoint侧输出的数据不会丢也不会有重复精确一次语义下。但是要注意侧流数据如果直接sink到外部系统外部系统的写入要满足幂等性否则Flink的重放机制会导致侧流数据重复。比如告警侧流一次重启可能重发告警生产上要考虑告警去重或者接受“至少一次”的语义。5.3 迟到的数据怎么处理侧输出流与Allowed Lateness的配合窗口计算中事件时间语义下晚到的数据默认会被丢弃。如果不想丢弃可以给窗口设置allowedLateness在窗口关闭后再等一段时间。但allowedLateness的时间窗关闭后有数据来了窗口已经无法更新结果这时俗称“回撤”的一个常见处理办法就是把这些迟到数据送入侧输出流。实现方式是在窗口函数比如ProcessWindowFunction里通过Context的sideOutput方法输出迟到数据。这里有个容易混淆的点ProcessWindowFunction的Context同样有output方法所以窗口内也能做侧输出不只在KeyedProcessFunction里才能做。下边是一个简化写法DataStreamOrder windowed orders .keyBy(Order::getUserId) .timeWindow(Time.minutes(5)) .allowedLateness(Time.minutes(1)) .process(new ProcessWindowFunctionOrder, OrderStats, String, TimeWindow() { private final OutputTagOrder lateTag new OutputTagOrder(late-order) {}; Override public void process(String s, Context context, IterableOrder elements, CollectorOrderStats out) throws Exception { if (context.window() ! null) { // 窗口正常输出 } } // 注意迟到数据是通过LateDataOutputTag机制在外部处理ProcessWindowFunction本身要显式定义 });严格说迟到数据分流有三种做法一是窗口带allowedLateness且窗口函数内判断二是通过SideOutputLateData机制新版API里是withLateData三是通过测流配合。实际项目中建议在窗口声明时用.withLateData(lateTag)这是Flink专门为迟到数据设计的侧输出出口语义最明确。5.4 单独升级多级侧输出流嵌套Flink是支持“侧输出的侧输出”的——也就是说从getSideOutput得到的旁路流如果再用process函数处理依然可以再定义新的侧输出流。这在多级分流的场景下很有用。比如第一级分流出异常订单第二级再在异常订单里分出“风控命中”和“规则校验失败”两个子类。每一级的OutputTag在本级定义即可下一级拿到的仍然是一个标准DataStream能力不受限制。层级多了注意命名别乱我建议tag的id直接体现层级比如level1-timeout、level2-fraud。不过有一点提醒多级侧输出会让拓扑的调试难度上升数据流向不再是一条直线。建议把侧输出流的定义和拉取集中放在一起管理不要散落在各个函数里否则运维同事接手时容易一头雾水。5.5 侧输出流和广播流、维表关联的配合实时计算里经常要动态更新规则比如风控规则、阈值配置。这类动态规则一般用广播流BroadcastStream下发规则变了不用重启作业。侧输出流和广播流结合时通常是在KeyedBroadcastProcessFunction的processBroadcastElement里更新规则状态在processElement里做判断命中规则的进侧流。这里有个细节KeyedBroadcastProcessFunction同样继承自BaseKeyedProcessFunction所以也能拿Context做侧输出能力上没有缺失。但广播状态的数据结构一旦定义后续变更比较麻烦建议规则版本化能演化不推翻。5.6 中国式场景压测和流量高峰下的侧输出流每次大促压测我都会特意观察侧输出流的吞吐表现。侧输出流在算子内部和主流共用同一个执行线程所以它不会增加额外的线程开销。但如果侧流的下游是另一个吃资源的算子比如写库、调外部API一旦旁路数据量突增会反过来拖慢主链路的处理因为它俩在同一个TaskManager的slot上运行。有个经验做法给侧输出流的下游单独设置并行度和主流错开资源分配。比如主流并行度是16侧流告警下游的并行度可以调到4避免旁路任务把主流资源挤爆。Flink作业的资源隔离在这个场景下很重要。6. 横向对比侧输出流 vs 多路Sink、复制流方案6.1 侧输出流 vs 复制数据流复制数据流就是把source的数据同时发给多个下游算子每个算子各自处理。看起来也能实现“一份数据多处用”但问题在于每一路都重复消费同一份数据source的读取吞吐被放大N倍对Kafka等上游系统造成不必要的压力。侧输出流不会这样的源只读一次算子在内部把数据分发到主流和旁路。尤其当旁路的数据量远小于主流时侧输出流的模式节省了整整一路上游消费的开销。6.2 侧输出流 vs 多Kafka Topic分流有些团队习惯在接入层就把不同数据写到不同Kafka topic比如正常订单一个topic、异常订单一个topic。然后Flink分别消费。这种模式的好处是存储层已经分离坏处是接入层逻辑变重实时性也差了一截——异常判定必须在上游完成如果Flink里的规则更灵活上游就无能为力了。侧输出流的优势是判定逻辑紧跟Flink作业本身规则更新只需要改作业。等侧输出流往下游sink的时候再决定是落Kafka还是落其他存储灵活得多。6.3 侧输出流 vs window的late data机制有人会把allowedLateness和侧输出流混在一起其实它们解决的维度不一样。allowedLateness是让窗口“多等一会儿”侧输出流是多出来的数据有地方去。窗口计算场景下最佳实践是两者配合先allowedLateness容忍正常延迟窗口关闭后把仍迟到的数据输出到侧流做修正或监控。只开allowedLateness不接侧流迟到的数据最终还是会丢只是丢得晚一点侧流才是数据的最终保障。7. 侧输出流排查思路与排错手记7.1 侧输出流数据始终为空这是最高频的问题。可能是你定义完getSideOutput之后下游没有接任何数据也可能是侧输出流的tag用的不是同一个实例。下面是排查清单检查process函数里有没有真正调用ctx.output有没有可能打到了别的条件分支上。确认拿侧流的dataStream是从process后的结果流上getSideOutput的而不是从原始source上。确认OutputTag泛型和ctx.output传人的数据类型一致不一致时编译能过但运行期可能静默丢数据。在process函数里本地打日志用log.info把输出到侧流的数据打印出来确认数据确实进了侧流。7.2 作业重启后侧流数据重复前面提过侧流数据跟主流走同一个checkpoint重启后如果外部sink不是幂等的就会出现重复。告警、短信、推送这类场景尤其要留意。解决办法是在侧流下游做去重比如用去重状态、Redis set、或者在告警文本里带唯一ID接收端做幂等。7.3 窗口场景下侧流不触发事件时间窗口配allowedLateness时如果watermark一直不推进比如某个上游key的数据一直不来定时器就不会触发侧流也不会输出。这个问题看起来是侧流没生效实际是watermark没推进。检查source的watermark策略尤其是空闲分区场景要配withIdleness否则一个分区没数据会把整个watermark卡住。7.4 反序列化报错导致旁路数据丢失当侧流数据的类型不是Flink内置类型也没有自定义Serializer时作业可能报出序列化相关的异常。这个坑不好排查因为有时候批流都能跑但一旦侧流下游做了特殊操作就炸。我的经验是自定义POJO类型尽量用public字段或者标准的getter/setter不要依赖内部类复杂类型干脆把TypeInformation显式传给OutputTagfinal OutputTagOrder tag new OutputTagOrder(order, TypeInformation.of(Order.class)) {};显式传TypeInformation可以绕开因泛型擦除导致的类型推断失败问题这个写法在复杂嵌套泛型场景下更稳。8. 关于侧输出流我认为最重要的一条建议侧输出流不是万能的但它解决了实时计算里一个非常高频的“旁路需求”。用过一段时间后我最大的感想是你会发现真正的复杂度不在侧输出流本身而在于你怎么组织主链路和旁路之间的边界。旁路太多了代码一样会乱旁路太少主逻辑会被异常分支塞满。我的个人实践习惯是主链路永远保持“主干干净”所有特殊情况一律侧输出宁可多定义两个tag也不要在主处理逻辑里堆if-else。这类代码后续维护起来尤其当团队里有新同学接手时带来的清晰度价值远大于多写几行API的成本。最后分享一个小技巧侧输出流的tag对象建议用静态final常量定义放在算子类的外部或者内部静态类里。这样多个算子引用同一个侧流tag时不会出现命名不一致导致的“抢不到数据”问题。这个细节我踩过希望大家别再踩。