做流计算的朋友应该都遇到过这种尴尬任务上线前什么都测好了结果某天一根异常数据进来算子直接抛异常整个作业重启或者事件时间窗口一闭合后面姗姗来迟的数据就那么被吞了业务对账对不齐连个“残骸”都找不到。Flink 侧输出流side output干的就是这件事在同一个算子内部让数据在主流和若干旁路之间分流旁路是一条独立的 DataStream可以单独消费、单独落库、单独监控绝不干扰主链路。适合谁凡是玩过 Flink 窗口、ProcessFunction或者被“一条流要拆成多条流”困扰过的同学这篇文章都值得读完我会从 API 原理一路讲到踩坑实录代码拿走就能改。1. 侧输出流解决的三个典型痛点1.1 迟到数据不是只能扔掉事件时间窗口处理的核心矛盾是窗口到底什么时候收盘。全收任务等不起不收内存顶不住。所以 Flink 默认窗口触发一次就算数晚到的数据直接丢弃。业务上最怕的不是丢而是“丢了没痕迹”。侧输出流的第一个典型用途就是窗口迟到数据兜底allowedLateness允许窗口在收盘后再等一段时间这段时间内的迟到数据走主流正常参与统计sideOutputLateData再把连这段宽限期都没赶上的数据单独放到一条侧流里。也就是说迟到数据不再等于“消失”而是换了一条路走。你可以把侧流接到一个外部存储里存档也可以把它单独计数做成监控指标。比如单量统计的场景当天 10 点的单客户 11 点半才报上来主窗口早就关了但如果有一条按事件时间兜底的侧流这类订单至少能被记录下来归档、补偿、重算都有了依据。这是以前只用窗口算子时根本做不到的。再退一步说哪怕你暂时还没想好迟到数据拿来干嘛先把它留在侧流里打印出来也比让它静默消失强一百倍。1.2 一条流拆多条流别再 filter 硬切很多人的第一反应是要拆流是吧filter(条件A)拿一路filter(条件B)拿一路。问题在于一条数据流被 filter 两遍意味着每条数据至少被处理了两遍CPU 和序列化开销直接翻倍更麻烦的是filter 只能“留下符合的”无法捕获“不符合的”——被过滤掉的数据去哪儿了没有答案。这种时候你需要的是旁路而不是筛子。侧输出流的做法是在一个 ProcessFunction 里根据判断条件分别ctx.output(tag, value)主流只保留默认路径其他分支各自进侧流。数据只经过一次算子计算成本最小而且每个分支都是类型安全的 DataStream后续算子随便接。我见过一个项目要把流量按“新客、老客、异常会话”拆成三路最早用三个 filter 硬切后来高峰期 CPU 直接拉满改成侧输出流之后同样的数据量 CPU 降了快三分之一差异就是这么明显。1.3 为什么 split/select 被废弃了Flink 早期版本提供过DataStream.split()select()做流拆分用起来的确简单但它的设计有几个硬伤第一split 只能按数据内容打标签没有类型信息select 回来以后全是同一个类型第二标签信息没有跨算子传递拆完的流很难再被其他算子描述第三官方早已将其标记过时新版本里基本就是个“历史包袱”。侧输出流用OutputTag携带类型信息每个旁路本质是一条独立的、强类型的 DataStream什么都能接这正好补上了 split 的所有短板。所以新代码一律用侧输出流别在旧 API 上浪费时间。2. 核心原理与 API 拆解2.1 OutputTag一条旁路的“身份证”侧输出流的入口是OutputTagT。它的作用有两个一是给旁路起名字二是声明旁路数据类型。声明时几乎总要用到匿名子类static final OutputTagUserAction LATE_TAG new OutputTagUserAction(late-actions) {};注意最后的那对花括号。Java 的泛型在运行时会被擦除直接new OutputTagUserAction(late)写出来等运行时只能读到Object类型类型信息全没了。匿名子类会把UserAction这个泛型参数“焊死”在类信息里Flink 才能正确推断并序列化。如果不想要匿名子类也可以用显式类型OutputTagUserAction tag new OutputTag(late-actions, TypeInformation.of(UserAction.class));两种写法等价但无论如何你都必须在声明时就敲定类型因为侧流后续的序列化、分区、下游 schema 全靠它。很多人忽略了这一点等到运行时报序列化异常才回头找原因白白浪费时间。2.2 主流和旁路的分工与拓扑一个 ProcessFunction 算子内部有Collector和Context两个出口。out.collect(value)是主流数据进入正常的下游链路ctx.output(tag, value)是旁路数据进入对应的侧流。一个算子可以有多个侧输出也可以往同一个侧输出多次写规则很自由。关键点在于这两条路并不是对立的。你完全可以在一个 ProcessFunction 里既往主流吐数据又往侧流吐数据也可以只往侧流吐、主流什么都不发。侧流拿回来之后要从对应的SingleOutputStreamOperator上这样取DataStreamUserAction lateStream mainStream.getSideOutput(LATE_TAG);getSideOutput是定义在SingleOutputStreamOperator上的方法所以只能从“单个输出算子”身上取侧流而不是随便从哪个 DataStream 上取。这条侧流和主流一样之后可以 keyBy、开窗口、接 sink一切照旧。拓扑上相当于同一个算子长出了好几条向下游延伸的分支每条分支的语义完全独立。2.3 哪些算子支持侧输出很多初学者只以为 ProcessFunction 能用其实支持面比想象中大我整理了一个速查算子入口侧输出用法ProcessFunction/KeyedProcessFunction/CoProcessFunctionContext.output(tag, value)ProcessWindowFunctionContext.output(tag, value)ProcessAllWindowFunctionContext.output(tag, value)KeyedProcessFunction配合定时器定时器回调里也能output窗口算子WindowedStream.sideOutputLateData(tag)直接兜底迟到数据普通map、flatMap这类算子没有内置的Context.output可以先.process(...)转成 ProcessFunction 再用侧输出所以实际操作中我几乎都是无脑 process。还有一点容易忽略侧输出不改变输入流的 key 状态也就是说在 KeyedProcessFunction 里往侧流吐数据时数据本身不会自动带着“key 分组信息”落到下游下游如果还需要按 key 聚合记得重新 keyBy。3. 三个实战场景的完整实现3.1 场景一窗口迟到数据的兜底收集先串一下完整流程进入窗口算子的数据经过allowedLateness和sideOutputLateData两层筛选之后迟到数据走上旁路。这个组合里两个配置缺一不可allowedLateness决定“宽限多久”sideOutputLateData决定“宽限过后往哪送”。// 迟到的兜底标签 static final OutputTagUserAction LATE_ACTION_TAG new OutputTagUserAction(late-action) {}; SingleOutputStreamOperatorWindowResult result source .keyBy(UserAction::getUserId) .window(TumblingEventTimeWindows.of(Time.minutes(5))) .allowedLateness(Time.minutes(1)) .sideOutputLateData(LATE_ACTION_TAG) .aggregate(new ActionAggregate()); // 取旁路迟到的数据单独落盘 DataStreamUserAction lateActions result.getSideOutput(LATE_ACTION_TAG); lateActions.map(UserAction::toJson) .sinkTo(FileSink.forRowFormat( new Path(hdfs://.../late-actions/), new SimpleStringEncoderString(UTF-8)) .withRollingPolicy(DefaultRollingPolicy.builder() .withRolloverInterval(Time.minutes(5).toMilliseconds()) .withInactivityInterval(Time.minutes(1).toMilliseconds()) .withMaxPartSize(128 * 1024 * 1024) .build()) .build());这里有个容易看漏的点sideOutputLateData返回的仍然是WindowedStream所以它的位置必须在.window()之后、真正的窗口计算函数.aggregate()/.apply()/.process()之前这样窗口算子才会在内部把迟到数据“分拣”出来。调用顺序错了比如放在.aggregate()之后编译器直接报错。3.2 场景二日志等级分流这个场景在日志处理任务里很常见一条日志流按 WARN/ERROR/INFO 拆成三路三个下游各自处理主流只保留 INFO。用一个自定义 ProcessFunction 就能搞定private static class LogLevelSplit extends ProcessFunctionLogRecord, LogRecord { private final OutputTagLogRecord warnTag new OutputTagLogRecord(warn) {}; private final OutputTagLogRecord errorTag new OutputTagLogRecord(error) {}; Override public void processElement(LogRecord log, Context ctx, CollectorLogRecord out) { switch (log.getLevel()) { case WARN: ctx.output(warnTag, log); break; case ERROR: ctx.output(errorTag, log); break; default: out.collect(log); // INFO 走主流 } } } SingleOutputStreamOperatorLogRecord infoStream logs.process(new LogLevelSplit()); DataStreamLogRecord warnStream infoStream.getSideOutput(warnTag); DataStreamLogRecord errorStream infoStream.getSideOutput(errorTag);注意warnTag和errorTag是 ProcessFunction 实例的成员变量。Flink 在把算子分发到 TaskManager 时会序列化这个函数对象所以 OutputTag 字段必须final修饰且类型本身可序列化。再提醒一句同一个标签对象在“生产侧”和“消费侧”要保持一致最稳的办法是声明成static final或者通过构造器传入同一个实例如果代码里各写各的new OutputTag(...)ID 一样通常也能匹配上但没必要给自己留这个风险。如果以后还想把 ERROR 侧流和 WARN 侧流合并起来做统一告警直接.union(warnStream, errorStream)就行两边类型一致非常好接。3.3 场景三脏数据审计与链路监控数据处理任务里脏数据的处理最能体现一个作业的工程成熟度。以前遇到解析失败的记录要么直接抛异常让任务重启要么打个日志默默丢掉。用侧输出流可以把这些记录原封不动地转进旁路然后做三件事计数、落盘、告警。SingleOutputStreamOperatorValidRecord valid rawRecords .process(new ProcessFunctionRawRecord, ValidRecord() { private final OutputTagRawRecord dirtyTag new OutputTagRawRecord(dirty) {}; Override public void processElement(RawRecord raw, Context ctx, CollectorValidRecord out) { try { out.collect(parser.parse(raw)); } catch (Exception ex) { ctx.output(dirtyTag, raw); // 解析失败走旁路 } } }); DataStreamRawRecord dirty valid.getSideOutput(dirtyTag); dirty.map(DirtyCounter::count) .process(new AlertOnHighRate()) // 脏数据比例超阈值自动告警 .sinkTo(alertSink);这个模式最值钱的地方在于主链路不用 break。脏数据再也不会因为一个解析异常就把整个作业拖垮异常记录被完好保留出问题时有据可查。我在实际项目里经常把“审计侧流”连同任务名、算子名等埋点一起记录出事之后十分钟就能定位是哪条数据哪一步解析挂了。这种“旁路审计”的思路本质上是把异常从“系统故障”降级成“普通业务数据”对整个作业的稳定性和可观测性都是质的提升。4. 常见问题与排查实录4.1 类型擦除与序列化OutputTag 为什么必须匿名类这是新手最常见的问题写new OutputTagString(mytag)去掉花括号然后运行时报出序列化相关异常或者下游getSideOutput拿回来的流元素类型不对、反序列化报错。原因就是泛型擦除。Java 运行时不知道String是什么OutputTag里的TypeInformation被推断成了GenericType甚至变成ObjectFlink 只好走 Kryo性能差且不稳定。判断方法也简单IDE 里悬停看new OutputTag...的类型声明或者直接看日志里有没有Kryo字样的序列化告警。解决方案就两条匿名子类或显式传TypeInformation。这点小差异影响整个作业的稳定性和吞吐别偷懒。如果用的是自定义 POJO顺手给它实现toString、equals、hashCode并保证有无参构造器后续序列化和调试都会省很多事。4.2 侧流不是“时光机”再聚合要小心水位线侧流里的迟到数据虽然本身带时间戳但侧流下游看到的水位线仍然是原算子传下去的水位线不会因为侧流里出现一条“更早”的数据就倒退。也就是说你如果天真地把迟到侧流再开一个事件时间窗口重新聚合窗口大概率已经过期触发逻辑会很尴尬要么要等下一个水位线才触发要么一直不触发。我的建议是侧流拿到之后优先做“面向结果的处理”直接写外部存储、转成维度更新信息、或者触发补偿任务而不是再次做事件时间窗口聚合。如果确实要聚合请把它当作一个完全独立的新输入源设计不要指望它继承上游的窗口语义。这个坑我踩过一次之后就再也不敢随意在侧流后面挂窗口了。4.3 背压、checkpoint 与分区语义侧流不是免费的。它本质上是同一个算子多开了一条发射通道下游消费慢照样会把背压传回上游主链路。曾经有个作业主流写到 Kafka 很快侧流却接了一个性能很差的外部监控接口结果整个作业吞吐掉了一半排查半天才找到元凶。所以侧流下游的 sink 一定要按和主流同等级别对待必要时给侧流接缓冲队列或者降级丢弃策略。checkpoint 方面侧流数据本身不会额外占用状态存储它和主流一样是算子正常输出的记录但如果你在侧流下游做了聚合、开了窗口那这些下游算子该有的状态照样会有清理侧流本身不需要额外配置。分区语义上侧流不会天然保持 key 的分布。你从 keyBy 之后的算子拿到侧流侧流里的数据并不保证按原 key 分区需要 keyBy 就再 keyBy 一次。4.4 调试与监控怎么落地侧流有一个很讨厌的特点看起来“存在”但你不知道它到底有没有数据。我常用的三招第一招在getSideOutput之后临时接一个打印节点mainStream.getSideOutput(LATE_ACTION_TAG) .map(a - { System.out.println([late] a.toJson()); return a; });定位问题非常快调试完删掉即可。第二招用自定义 RichFlatMapFunction 给侧流加指标比如脏数据量、迟到数据量注册成 task 级 Metric在监控看板上直接看趋势dirty.map(new RichMapFunctionRawRecord, RawRecord() { private transient Counter dirtyCounter; Override public void open(Configuration parameters) { dirtyCounter getRuntimeContext().getMetricGroup().counter(dirty_count); } Override public RawRecord map(RawRecord raw) { dirtyCounter.inc(); return raw; } });第三招在日志里打印水位线确认迟到数据是不是真的“迟到”。很多时候侧流没数据不是侧输出代码写错而是上游水位线压根没推进所有数据都还滞留在窗口里。这三招配合起来侧流基本处于“裸奔可观测”状态。我个人这几年的体会是侧输出流这套机制真正改变的其实是流处理任务的容错观它让你不再把“异常数据”和“迟到数据”当作系统故障去处理而是当成正常业务路径的一部分去看待。数据分流、审计、补偿、监控这些以前需要外部系统配合才能做到的事现在一个算子内部就闭环了。如果你还没在项目里用过它下次写窗口或者 ProcessFunction 的时候顺手把侧流加上哪怕暂时只是打印出来也比裸奔强得多。