Flink ProcessFunction实战：侧输出、时间戳校正与状态边界
1. 判断该不该自己上 ProcessFunction 的三条信号
很多人第一次接触ProcessFunction是在翻 Flink 官方文档的时候，看到那一句“最底层的处理函数”，然后就没再往下看。实际上在流处理任务里，一旦你开始遇到“我需要在元素流过的时候顺手做点别的事情”这类需求，ProcessFunction迟早要摆到台面上。它最常被使用的场合包括：对上游分配好的事件时间不太放心，需要在函数里重新校正时间戳；需要对数据按规则打上标记，然后把不合格的数据分到另一条流里，并保持主链路逻辑干净；需要基于水位线做时效判断，而不是依赖窗口算子。这三个场景一旦出现，用map、flatMap、filter去拼凑会越拼越乱。本文围绕ProcessFunction这一类处理函数展开，讲清楚它的入参、上下文、输出通道各自能做什么、做不到什么，并用一个校验分流加时间戳校正的完整链路把代码跑通。适合已经写过 Flink 基础算子、想往底层走一步的同学，也适合被侧输出和定时器搞得一头雾水的同路人。
1.1 处理函数家族的全貌与它所在的位置
Flink 里的处理函数不是单一的一个类，而是一组接口和抽象类的集合。ProcessFunction是其中最基础的那一个，直接挂在DataStream上，作用于非 keyed 流。它的兄弟包括KeyedProcessFunction、CoProcessFunction、BroadcastProcessFunction，以及窗口场景下的ProcessWindowFunction、ProcessAllWindowFunction。全部这些类都继承自AbstractRichFunction，也就是说它们都拥有open、close、getRuntimeContext这套生命周期能力，可以读配置、拿累加器、注册指标，这一点和RichMapFunction是一脉相承的。区别体现在泛型签名和方法数量上：ProcessFunction<IN, OUT>只需要实现一个processElement，另有可选的onTimer；而KeyedProcessFunction<K, IN, OUT>需要额外传入 key 的类型参数，方法体里能访问到当前 key 和 keyed state。
把这组关系画成一句话：ProcessFunction是家族的门面，它把“逐元素处理 + 上下文访问 + 多路输出”这三件事打包成一个接口，而在它之上，KeyedProcessFunction额外挂了“按 key 分区”和“状态与定时器”两块能力。所以理解ProcessFunction是理解整个处理函数体系的第一步，跳过它直接去啃KeyedProcessFunction，很容易把两者的职责边界搞混。
1.2 ProcessFunction 相比普通算子的三样增量能力
第一个增量是上下文对象。map和filter拿了数据只给你数据，函数体里能用的只有元素本身。ProcessFunction的processElement会多塞给你一个Context，从中可以取当前元素的时间戳、当前的水位线值，还能拿到一个TimerService引用。这个上下文是只读快照，处理完当前元素就失效，你不能把它存起来以后再用，这一点要牢牢记住。
第二个增量是侧输出。普通算子输出的结果一律进同一个下游，想分流只能靠filter加多次计算，或者用split这种已经被标记为废弃的老 API。ProcessFunction通过ctx.output把数据送到带名字的旁路流里，主输出和旁路输出各走各的，下游用getSideOutput取出来即可。分流逻辑写在一个函数里，条件改动时只改一处，这对数据质量校验类的需求非常友好。
第三个增量是函数级的生命周期。因为继承了AbstractRichFunction，你可以在open里初始化连接池、加载字典、读取广播配置，并且在close里统一清理。这听起来和RichMapFunction差不多，但在处理函数里尤其重要，因为处理函数经常要持有外部资源做校验或补全，初始化时机选错会导致算子反复重连，后面的章节会细说。
1.3 一个反例：用 keyBy 加 flatMap 拼出“伪分流”的代价
我见过一个挺典型的写法：为了把异常数据拎出来，有人先用count或随机数给每条记录打标，再keyBy到不同分区，最后filter出异常值。这条链路能跑，但代价相当明显。首先，多一次keyBy就多一次全量 shuffle，数据量大时网络开销不便宜，而且这个 shuffle 完全不承载任何业务语义，纯粹是给分流让路。其次，分流规则分散在两个算子中，条件一改就得同时改两头，一旦漏改，异常数据会静默混进正流，排查起来非常痛苦。再者，这个写法没法复用上下文信息，比如“这条数据到达时水位线已经到了哪里”，普通算子根本拿不到。
ProcessFunction把这些动作收拢进一个方法里，分流规则、时间信息读取、输出路径全在同一屏代码内，改一处即可。这不是风格问题，是维护成本问题。写流处理任务久了会有一个感觉：凡是需要“看一眼上下文再决定去哪”的逻辑，都应该下沉到处理函数，交给上层算子只会越拼越拧。
2. processElement 的入参、上下文与输出通道逐项拆解
写ProcessFunction的日常工作量，九成集中在processElement一个方法里。这个方法有三个入参，各有分工，很多初学者把它们混着用，结果要么拿不到想要的信息，要么把输出路径写乱。这一段把三个参数拆开讲，顺带解释两个最容易踩的取值陷阱。
2.1 三个入参分别承载什么信息
第一个参数是元素本身，类型由泛型IN决定，它就是上游DataStream里流过来的那条数据。第二个参数是Context，这是一个只读的上下文对象，提供了timestamp()、timerService()、output()三个主要能力，后面会逐个展开。第三个参数是Collector<OUT>，代表主输出通道，调用out.collect(...)才能把数据送到下游主流，不调用就等于这条数据被丢弃了。这里有个新手常犯的错误：方法体里写了半天的逻辑，最后忘了collect，然后困惑为什么下游一条数据都收不到。空输出是合法的，编译器不会报错，单元测试如果不覆盖场景容易漏掉。
我的习惯是在方法开头就明确“默认走主输出”的意图，然后只在异常分支里提前return。这样写出来的代码阅读顺序比较顺：先校验，不合格就送侧输出并返回，合格的一路落到末尾再collect。中间增删条件时不需要反复找“到底哪里该输出”，逻辑链条始终清晰。
2.2 timestamp 与 currentWatermark 的读取时机和空值处理
ctx.timestamp()返回的是当前元素被赋予的事件时间戳，类型是Long，注意它是包装类型，可能是 null。如果上游没有调用assignTimestampsAndWatermarks，或者虽然调用了但分配逻辑没覆盖这条数据，这个方法会返回 null。直接拿它做比较或者拆箱会立刻抛空指针，而且这个异常发生在算子线程里，日志里往往只看到一段堆栈，很难第一时间联想到时间戳没分配这个原因。
ctx.timerService().currentWatermark()返回的是当前算子看到的水位线值，同样是Long。在没有任何水位线推进的情况下，它会返回Long.MIN_VALUE，也就是一个极大的负数。这一点在写时效判断时特别重要：直接用timestamp < currentWatermark判断迟到，在任务刚启动、水位线还没推起来的时候天然成立，会把正常数据误判成迟到。比较稳妥的方式是先判断currentWatermark是否等于Long.MIN_VALUE，是的话跳过迟到判断，只在确实有水位线推进之后再启用这条逻辑。
这两处取值都属于“看起来一行代码、实际上暗藏两个空值陷阱”的典型。我的做法是统一封装成一个工具方法，把 null 判断和最小值判断都塞进去，业务代码里只调用封装后的方法，避免每次手写都漏一处。
2.3 Collector 与 OutputTag 一起构成的分流通道
Collector只负责主输出，想分流必须借助OutputTag。声明方式有两种，一种是创建匿名子类，利用 Java 的匿名内部类保留泛型信息：
OutputTag<OrderEvent> dirtyTag = new OutputTag<OrderEvent>("dirty-order") {};另一种是显式传入类型信息：
OutputTag<OrderEvent> dirtyTag = new OutputTag<>("dirty-order", TypeInformation.of(OrderEvent.class));第一种写法更短，代价是每声明一次就多生成一个匿名类，量不大时可以接受。第二种在多处复用时更合适，可以做成静态常量，全工程共享。两种写法都要求 Tag 的名字在同一个作业内全局唯一，重名会在作业构建阶段直接抛异常，而不是运行时报错，这点反而比较友好。侧输出的数据在下游通过mainStream.getSideOutput(tag)取回，得到的是一条独立的DataStream，可以单独接sink或者再加工。
提示：
OutputTag的泛型参数决定了侧输出的序列化器，如果这里写成OutputTag<Object>，侧输出流拿到的类型信息就是通用类型，后面接sink时可能触发类型擦除相关的运行时错误。泛型一定要写具体。
3. 非 Keyed 场景下 ProcessFunction 的能力与硬边界
这一段是全文最值得细看的地方。很多关于ProcessFunction的困惑，根源都在于没搞清楚它在非 keyed 流上到底被允许做什么。有人照着KeyedProcessFunction的例子把定时器代码粘到ProcessFunction里，跑起来直接抛异常，然后开始怀疑人生。其实文档里写得很清楚，只是那句话容易被略过。
3.1 定时器注册为什么在这里直接被拒绝
ProcessFunction.Context.timerService()返回的是一个非 keyed 的定时器服务实现。你可以从中读取当前水位线，但一旦调用registerEventTimeTimer或registerProcessingTimeTimer，会立刻抛出UnsupportedOperationException。原因不复杂：Flink 的定时器是基于 key 存储的，每个 key 各自维护一批待触发的定时器，触发时还要能找回该 key 对应的状态。非 keyed 流没有 key 的概念，定时器挂在哪里、状态怎么回填都无从谈起，所以干脆禁掉。
这意味着ProcessFunction里覆写onTimer方法虽然语法上允许，但实际不会被触发，因为没有合法的注册入口。这个“能覆写但不会执行”的坑很隐蔽，编译不报错，运行时也没异常，只是那段代码永远静静地躺在那里。我见过有人在里面写了半天清理逻辑，上线之后发现从没生效，最后查了两天才定位到根源。所以记住一条：要用定时器，就得先keyBy，然后换KeyedProcessFunction。
3.2 状态访问失败的深层原因
同样的道理，ProcessFunction里通过getRuntimeContext().getState(...)拿 keyed state 也会失败。keyed state 的作用域是“当前 key”，没有 key 自然没有作用域。这时候如果你确实需要在整个算子实例范围内共享一点数据，可以用算子状态（Operator State）或者干脆在open里维护一个普通的成员变量。但要注意成员变量的生命周期和并行度强相关，任务扩缩容时不会自动迁移，也不参与 checkpoint 的一致性快照，只能放一些可重建的缓存，比如字典表、配置映射，不能放业务上不可丢失的数据。
这个边界看着有点扫兴，但换个角度想反而清爽：ProcessFunction被明确定义为“无状态、无 key、只做逐元素处理”的组件，职责单一，能力边界清楚。状态和定时器这类重活儿交给KeyedProcessFunction，两者各司其职，代码结构也更干净。
3.3 不用定时器也能做的两件时效性判断
虽然注册不了定时器，但ProcessFunction在时效判断上并非无能为力。第一件事是迟到元素识别：拿当前元素的时间戳和currentWatermark比较，如果水位线已经越过了这个时间点，说明这条数据来晚了。这个判断在窗口场景下同样有意义，因为窗口算子对迟到数据的默认处理是丢弃，如果你希望在丢弃之前先把它捞出来存档，用ProcessFunction放在窗口之前做一层拦截就正好。
第二件事是水位线单调性观测。currentWatermark在每条元素处理时都能读到，可以顺手记录水位线的推进幅度，配合指标系统观察数据源是否存在长时间不推进的情况。数据源空闲导致水位线卡住是很常见的线上问题，表现是窗口迟迟不触发、下游任务看起来“没数据”，而日志里一切正常。让处理函数定期吐一条水位线值出来，排查时能省很多时间。
这两件事都不需要 key，不需要状态，完全在ProcessFunction的能力范围内，非常适合放在数据清洗链路上做个顺手之举。
4. 完整实战：带校验、时间戳校正与脏数据分流的处理链路
理论讲到这里，该上代码了。这个 Demo 的场景很朴素：上游是一批订单事件，字段包含订单号、用户、事件类型、事件时间和金额，其中一部分数据存在字段缺失、金额为负、时间戳缺失等问题。需要一条链路做三件事——字段校验、时间戳兜底、脏数据分流，主链路只放行干净数据。全部逻辑集中在一个ProcessFunction里，正好把前面讲的能力串起来。
4.1 数据模型与拓扑结构设计
先定数据模型。用一个简单 POJO 承载订单事件，字段保持扁平，方便序列化：
public class OrderEvent { public String orderId; public String userId; public String eventType; // CREATE / PAY / CANCEL public long eventTs; // 事件实际发生时间，毫秒 public long amount; public long ingestTs; // 进入系统的时间，用于兜底 public OrderEvent() {} // 省略 getter / setter }拓扑设计上，数据源先进水位线分配，再进处理函数，处理函数输出主流和脏数据侧流两条线。水位线分配用有界乱序策略，容忍 5 秒的乱序窗口，时间戳提取直接用eventTs。主链路下游接消费逻辑，脏数据侧流接一个低频的落盘sink，方便人工复查。这个结构刻意保持扁平，避免在演示链路里堆太多算子干扰阅读。
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(2); DataStream<OrderEvent> raw = env.addSource(new OrderEventSource()); DataStream<OrderEvent> withTs = raw.assignTimestampsAndWatermarks( WatermarkStrategy.<OrderEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((e, ts) -> e.eventTs) ); OutputTag<OrderEvent> dirtyTag = new OutputTag<OrderEvent>("dirty-order") {}; SingleOutputStreamOperator<OrderEvent> clean = withTs .process(new OrderCleanseFunction(dirtyTag)); DataStream<OrderEvent> dirty = clean.getSideOutput(dirtyTag); clean.addSink(new CleanOrderSink()); dirty.addSink(new DirtyOrderSink());4.2 核心代码逐段拆解
处理函数本体分四段，每段职责独立，便于后续单独修改。
public class OrderCleanseFunction extends ProcessFunction<OrderEvent, OrderEvent> { private final OutputTag<OrderEvent> dirtyTag; private transient Counter dirtyCounter; public OrderCleanseFunction(OutputTag<OrderEvent> dirtyTag) { this.dirtyTag = dirtyTag; } @Override public void open(Configuration parameters) { dirtyCounter = getRuntimeContext() .getMetricGroup() .counter("dirty_order_total"); } @Override public void processElement(OrderEvent value, Context ctx, Collector<OrderEvent> out) { // 第一段：基础字段校验 if (value == null || value.orderId == null || value.eventType == null || value.amount < 0) { dirtyCounter.inc(); ctx.output(dirtyTag, value); return; } // 第二段：时间戳兜底与校正 Long assignedTs = ctx.timestamp(); if (assignedTs == null) { if (value.ingestTs <= 0) { dirtyCounter.inc(); ctx.output(dirtyTag, value); return; } value.eventTs = value.ingestTs; } // 第三段：基于水位线的迟到识别 Long watermark = ctx.timerService().currentWatermark(); if (watermark != null && watermark != Long.MIN_VALUE && value.eventTs < watermark) { dirtyCounter.inc(); ctx.output(dirtyTag, value); return; } // 第四段：放行主链路 out.collect(value); } }第一段是纯字段校验，不依赖任何上下文，写完就return，逻辑上最不需要解释。第二段处理的是时间戳缺失，这里用ingestTs做兜底，如果连兜底值都没有，说明这条数据确实没法用于事件时间计算，只能进脏数据流。之所以保留这个兜底分支，是因为我在实际数据里见过不少采集端漏填事件时间的样本，直接丢弃会损失可用的统计口径。
第三段是迟到判断。这里刻意加了双重防护：先判断watermark是否为 null，再判断是否等于Long.MIN_VALUE。刚启动时水位线是极小值，如果跳过这层判断，所有早期数据都会被误判为迟到，仪表盘上的脏数据计数会瞬间飙升，很容易让人误以为数据质量出了大问题。
第四段只有一行collect，但这行是整个方法的出口。所有校验通过的数据最终都走到这里，主链路和脏数据流的分界线就落在这个语句上。
4.3 本地跑通与验证手段
验证这个处理函数不需要搭集群。用env.fromElements(...)构造几条典型样本即可：一条完全正常、一条金额为负、一条事件时间为零且无兜底、一条事件时间远早于当前水位线。然后分别收集主流和侧流的结果，断言条数和内容。
DataStream<OrderEvent> input = env.fromElements( okEvent("A001", 100L), badAmount("A002", -1L), noTime("A003"), lateEvent("A004", 1L) ).assignTimestampsAndWatermarks(...); SingleOutputStreamOperator<OrderEvent> result = input.process(new OrderCleanseFunction(dirtyTag)); List<OrderEvent> mainOut = new ArrayList<>(); result.addSink(new CollectSink<>(mainOut)); List<OrderEvent> dirtyOut = new ArrayList<>(); result.getSideOutput(dirtyTag).addSink(new CollectSink<>(dirtyOut));用CollectSink这种把数据塞进静态集合的测试sink，跑完直接断言即可。这种写法在本地单测里足够用了，虽然它有并行度相关的局限，但对验证函数逻辑完全够。注意测试环境要显式设置并行度为 1，否则静态集合会被多个子任务写入，断言结果不稳定。
4.4 侧输出 Tag 的声明方式与序列化坑
侧输出 Tag 的声明前面提过两种写法，这里补一个实战中容易踩的坑：Tag 必须能在不同算子之间被正确识别，靠的是名字加类型信息。如果同一个作业里在 A 处用匿名子类声明、在 B 处用类型信息构造，只要名字和类型一致，仍然能对上；但如果在两处用了不同名字，或者一处写了泛型一处没写，取侧输出时会拿到一条空流，而且不报错。
我踩过一次特别隐蔽的：同一个作业里有两个数据处理函数都往名为error的侧输出写数据，但它们的泛型类型不同。作业能正常提交，运行时只报一个类型不匹配的告警，侧输出流里混进来的数据类型五花八门。从那以后我给所有 Tag 名字加了业务前缀，格式统一成“模块名-用途”，并且在类里声明为静态常量，全局唯一引用。
5. 从 ProcessFunction 切换到 KeyedProcessFunction 的改造路径
第四节的链路跑顺之后，需求往往会往前走一步：现在需要按订单号维护状态，比如同一个订单在 15 分钟内没有支付就自动标记为超时。这时候侧输出和时间戳校正都还在，但定时器与状态成了刚需，ProcessFunction就到顶了，得换成KeyedProcessFunction。
5.1 改造的触发条件判断
我的判断标准很简单：只要出现下面任意一条，就该考虑切换。需要按某个业务维度维持跨元素的状态，比如计数、去重、累积求和；需要在未来某个时间点主动触发逻辑，而不是等数据来了再处理；需要感知某个业务维度的水位线推进情况，而不是全局水位线。这三条都是KeyedProcessFunction的强项。
反过来，如果逻辑只跟单条数据有关，不关心历史也不关心未来，那就留在ProcessFunction里，没必要为了用定时器而硬造一个 key。key 本身会引入分区、乱序、状态规模这些额外成本，能不加就不加。
5.2 代码层面的具体迁移步骤
迁移分三步。第一步，在上游加keyBy，明确用哪个字段做 key。第二步，把类的继承关系从ProcessFunction<IN, OUT>改成KeyedProcessFunction<K, IN, OUT>，processElement的签名跟着变化，多出来的K key参数正是当前 key 值。第三步，有了key和上下文里的timerService，就能注册定时器并访问 keyed state，onTimer也从“摆设”变成了真正的回调。
public class OrderTimeoutFunction extends KeyedProcessFunction<String, OrderEvent, OrderEvent> { private transient ValueState<Long> tsState; @Override public void open(Configuration parameters) { ValueStateDescriptor<Long> desc = new ValueStateDescriptor<>("create-ts", Long.class); tsState = getRuntimeContext().getState(desc); } @Override public void processElement(OrderEvent value, Context ctx, Collector<OrderEvent> out) throws Exception { if ("CREATE".equals(value.eventType)) { tsState.update(value.eventTs); ctx.timerService().registerEventTimeTimer(value.eventTs + 15 * 60 * 1000L); } else if ("PAY".equals(value.eventType)) { tsState.clear(); } out.collect(value); } @Override public void onTimer(long timestamp, OnTimerContext ctx, Collector<OrderEvent> out) { if (tsState.value() != null) { OrderEvent timeout = new OrderEvent(); timeout.orderId = ctx.getCurrentKey(); timeout.eventType = "TIMEOUT"; out.collect(timeout); tsState.clear(); } } }注意processElement里的throws Exception：状态读写可能抛异常，签名必须放开，这一点和ProcessFunction有区别，ProcessFunction的默认实现里可以不写throws。迁移时如果漏了这个，编译直接过不去。
5.3 迁移过程中最容易漏的两处配置
第一处是状态后端的选型。ValueState在生产环境通常要配RocksDBStateBackend，否则状态量大时堆内存扛不住。默认的堆内状态后端在数据量小的时候没问题，一旦单个 key 的状态对象变大、或者 key 的数量到了百万级，GC 压力会明显抬头。迁移到KeyedProcessFunction之后，状态规模就从“零”变成了“要长期维护的东西”，这项配置必须提前规划。
第二处是定时器的清理。registerEventTimeTimer注册的定时器不会因为状态被清空而自动撤销。上面那段代码里，订单支付之后清理了状态，但定时器还挂着，到点触发时tsState.value()为 null，靠这个判断兜住了。如果业务上定时器触发本身还有别的副作用，比如发通知，那就必须在支付分支里显式调用deleteEventTimeTimer，否则会重复触发。定时器数量本身也是资源，长跑任务里长期不清理会导致定时器堆积，触发时扫描开销越来越大，这种问题往往要跑上几天才暴露出来。
6. 上线前必须过的检查清单与踩坑复盘
代码写完到上线之间隔着一段很起实际作用的路。这一段列几类我在本地联调和集群运行中真实遇到过的问题，都是处理函数这个场景下的典型症状。
6.1 本地单测与集群联调中遇到的四类问题
第一类是上下文对象被跨线程持有。有人把ctx塞进一个异步任务的闭包里，等回调触发时再去读ctx.timestamp()，结果拿到的是别的元素的时间戳，甚至直接报错。上下文只在当前元素处理期间有效，这个约束没有任何商量余地。需要异步处理就用AsyncDataStream或者自己管好数据拷贝，绝不能把上下文带出去。
第二类是侧输出流漏取。作业跑起来之后发现脏数据没地方去，查了半天配置，最后发现是getSideOutput忘了调用，侧输出写进去的数据被静默丢弃。这个行为不报任何错误，因为侧输出本来就是“写了可以不用”，算子不会做校验。我的习惯是在作业构建阶段统一做一次遍历，把定义了但没消费的 Tag 打日志提示。
第三类是序列化问题。处理函数本身会被序列化后分发到各个子任务，类里的成员变量必须可序列化。Tag 字段没问题，但如果在里面塞了一个数据库连接对象，作业提交阶段直接抛NotSerializableException，堆栈指向的却常常是某个不起眼的字段。所有外部资源都在open里初始化，并且声明成transient，这条规则在处理函数里没有例外。
第四类是并行度变化导致的行为漂移。本地测试用的是 1 并行度，侧流里的数据顺序符合预期；上了集群跑到 4 并行度，同一批数据在不同分区里的处理顺序变了，如果下游逻辑依赖顺序，行为就会不一致。处理函数本身不保证跨分区顺序，这一点在设计时就得认清楚，需要全局顺序的场景要另想办法。
6.2 性能相关的三条经验
处理函数处在数据流的正中间，它的执行效率会直接影响整条链路的吞吐。第一条经验是别在processElement里做同步阻塞调用。查一次外部服务的耗时如果是 5 毫秒，在每秒几万条的流量下就是灾难。需要外部查询的，走异步 IO 或者提前用广播把维表发下来。
第二条经验是控制单条数据的处理时长。反压往往不是算子整体慢，而是某一个分支里出现了长尾。比如某条脏数据触发了特别复杂的分支逻辑，或者某次查询超时了 30 秒，这 30 秒内整个子任务的处理都被卡住。给外部调用设超时、给分支逻辑设上限，是处理函数里必须有的防御。
第三条经验是合理设置并行度和资源配置。处理函数的并行度跟上下游保持一致最省心，避免额外的数据重分布。如果处理函数特别重，可以考虑单独拆成一个算子并提高并行度，但这会引入额外的网络传输，要看瓶颈到底在哪。我一般的做法是先看背压监控，定位到具体算子之后再调，而不是一上来就全局加并行度。
6.3 与 checkpoint、重启策略的配合
处理函数本身不持有需要快照的业务状态，但如果它依赖open里加载的缓存，就要考虑任务重启后的重建成本。缓存可以通过广播流定期刷新，也可以放在open里加载一次。前者能应对源端数据变更，后者启动更快但数据会陈旧。选择哪种看业务对数据新鲜度的要求，没有统一答案。
重启策略方面，处理函数里如果捕捉了大量异常并吞掉，会导致任务明明有问题却一直“健康”运行，指标上看不出异常。我的做法是把异常分为两类：可恢复的（比如单条脏数据）走侧输出并计数，不可恢复的（比如外部依赖不可用）直接抛出，交给 Flink 的重启机制处理。让框架知道出了问题，比自己在函数里硬扛要可靠得多。
最后分享一个我在实际项目里养成的小习惯：处理函数写完之后，先在open里打一行日志把当前子任务编号和关键配置打印出来，联调时能一眼看出数据落在哪个分区、配置有没有按预期生效。这行日志上线后可以降级成 debug 级别，但排查问题时它往往是最先派上用场的那条信息。