Flink侧输出流实战：迟到数据兜底、多路分流与脏数据审计

做流计算的朋友应该都遇到过这种尴尬：任务上线前什么都测好了，结果某天一根异常数据进来，算子直接抛异常，整个作业重启；或者事件时间窗口一闭合，后面姗姗来迟的数据就那么被吞了，业务对账对不齐，连个“残骸”都找不到。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：一条旁路的“身份证”

侧输出流的入口是OutputTag<T>。它的作用有两个：一是给旁路起名字，二是声明旁路数据类型。声明时几乎总要用到匿名子类：

static final OutputTag<UserAction> LATE_TAG = new OutputTag<UserAction>("late-actions") {};

注意最后的那对花括号。Java 的泛型在运行时会被擦除，直接new OutputTag<UserAction>("late")写出来，等运行时只能读到Object类型，类型信息全没了。匿名子类会把UserAction这个泛型参数“焊死”在类信息里，Flink 才能正确推断并序列化。如果不想要匿名子类，也可以用显式类型：

OutputTag<UserAction> tag = new OutputTag<>("late-actions", TypeInformation.of(UserAction.class));

两种写法等价，但无论如何，你都必须在声明时就敲定类型，因为侧流后续的序列化、分区、下游 schema 全靠它。很多人忽略了这一点，等到运行时报序列化异常才回头找原因，白白浪费时间。

2.2 主流和旁路的分工与拓扑

一个 ProcessFunction 算子内部有Collector和Context两个出口。out.collect(value)是主流，数据进入正常的下游链路；ctx.output(tag, value)是旁路，数据进入对应的侧流。一个算子可以有多个侧输出，也可以往同一个侧输出多次写，规则很自由。

关键点在于这两条路并不是对立的。你完全可以在一个 ProcessFunction 里既往主流吐数据，又往侧流吐数据；也可以只往侧流吐、主流什么都不发。侧流拿回来之后，要从对应的SingleOutputStreamOperator上这样取：

DataStream<UserAction> lateStream = mainStream.getSideOutput(LATE_TAG);

getSideOutput是定义在SingleOutputStreamOperator上的方法，所以只能从“单个输出算子”身上取侧流，而不是随便从哪个 DataStream 上取。这条侧流和主流一样，之后可以 keyBy、开窗口、接 sink，一切照旧。拓扑上相当于同一个算子长出了好几条向下游延伸的分支，每条分支的语义完全独立。

2.3 哪些算子支持侧输出

很多初学者只以为 ProcessFunction 能用，其实支持面比想象中大，我整理了一个速查：

普通map、flatMap这类算子没有内置的Context.output，可以先.process(...)转成 ProcessFunction 再用侧输出，所以实际操作中我几乎都是无脑 process。

还有一点容易忽略：侧输出不改变输入流的 key 状态，也就是说在 KeyedProcessFunction 里往侧流吐数据时，数据本身不会自动带着“key 分组信息”落到下游；下游如果还需要按 key 聚合，记得重新 keyBy。

3. 三个实战场景的完整实现

3.1 场景一：窗口迟到数据的兜底收集

先串一下完整流程：进入窗口算子的数据，经过allowedLateness和sideOutputLateData两层筛选之后，迟到数据走上旁路。这个组合里两个配置缺一不可：allowedLateness决定“宽限多久”，sideOutputLateData决定“宽限过后往哪送”。

// 迟到的兜底标签 static final OutputTag<UserAction> LATE_ACTION_TAG = new OutputTag<UserAction>("late-action") {}; SingleOutputStreamOperator<WindowResult> result = source .keyBy(UserAction::getUserId) .window(TumblingEventTimeWindows.of(Time.minutes(5))) .allowedLateness(Time.minutes(1)) .sideOutputLateData(LATE_ACTION_TAG) .aggregate(new ActionAggregate()); // 取旁路：迟到的数据单独落盘 DataStream<UserAction> lateActions = result.getSideOutput(LATE_ACTION_TAG); lateActions.map(UserAction::toJson) .sinkTo(FileSink.forRowFormat( new Path("hdfs://.../late-actions/"), new SimpleStringEncoder<String>("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 ProcessFunction<LogRecord, LogRecord> { private final OutputTag<LogRecord> warnTag = new OutputTag<LogRecord>("warn") {}; private final OutputTag<LogRecord> errorTag = new OutputTag<LogRecord>("error") {}; @Override public void processElement(LogRecord log, Context ctx, Collector<LogRecord> out) { switch (log.getLevel()) { case WARN: ctx.output(warnTag, log); break; case ERROR: ctx.output(errorTag, log); break; default: out.collect(log); // INFO 走主流 } } } SingleOutputStreamOperator<LogRecord> infoStream = logs.process(new LogLevelSplit()); DataStream<LogRecord> warnStream = infoStream.getSideOutput(warnTag); DataStream<LogRecord> errorStream = infoStream.getSideOutput(errorTag);

注意warnTag和errorTag是 ProcessFunction 实例的成员变量。Flink 在把算子分发到 TaskManager 时会序列化这个函数对象，所以 OutputTag 字段必须final修饰，且类型本身可序列化。再提醒一句：同一个标签对象在“生产侧”和“消费侧”要保持一致，最稳的办法是声明成static final，或者通过构造器传入同一个实例；如果代码里各写各的new OutputTag(...)，ID 一样通常也能匹配上，但没必要给自己留这个风险。如果以后还想把 ERROR 侧流和 WARN 侧流合并起来做统一告警，直接.union(warnStream, errorStream)就行，两边类型一致，非常好接。

3.3 场景三：脏数据审计与链路监控

数据处理任务里，脏数据的处理最能体现一个作业的工程成熟度。以前遇到解析失败的记录，要么直接抛异常让任务重启，要么打个日志默默丢掉。用侧输出流，可以把这些记录原封不动地转进旁路，然后做三件事：计数、落盘、告警。

SingleOutputStreamOperator<ValidRecord> valid = rawRecords .process(new ProcessFunction<RawRecord, ValidRecord>() { private final OutputTag<RawRecord> dirtyTag = new OutputTag<RawRecord>("dirty") {}; @Override public void processElement(RawRecord raw, Context ctx, Collector<ValidRecord> out) { try { out.collect(parser.parse(raw)); } catch (Exception ex) { ctx.output(dirtyTag, raw); // 解析失败，走旁路 } } }); DataStream<RawRecord> dirty = valid.getSideOutput(dirtyTag); dirty.map(DirtyCounter::count) .process(new AlertOnHighRate()) // 脏数据比例超阈值自动告警 .sinkTo(alertSink);

这个模式最值钱的地方在于：主链路不用 break。脏数据再也不会因为一个解析异常就把整个作业拖垮，异常记录被完好保留，出问题时有据可查。我在实际项目里经常把“审计侧流”连同任务名、算子名等埋点一起记录，出事之后十分钟就能定位是哪条数据哪一步解析挂了。这种“旁路审计”的思路，本质上是把异常从“系统故障”降级成“普通业务数据”，对整个作业的稳定性和可观测性都是质的提升。

4. 常见问题与排查实录

4.1 类型擦除与序列化：OutputTag 为什么必须匿名类

这是新手最常见的问题：写new OutputTag<String>("mytag")去掉花括号，然后运行时报出序列化相关异常，或者下游getSideOutput拿回来的流元素类型不对、反序列化报错。原因就是泛型擦除。Java 运行时不知道String是什么，OutputTag里的TypeInformation被推断成了GenericType甚至变成Object，Flink 只好走 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 RichMapFunction<RawRecord, 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 的时候，顺手把侧流加上，哪怕暂时只是打印出来，也比裸奔强得多。