企业官网建设流程全解析

上个月帮一个团队看他们的离线报表作业，任务本身不复杂——读两张宽表，按天做聚合，再写回结果表。他们用的是 DataStream API，跑在 Flink 上，配置里全是流式默认值。作业能出结果，但一天的数据要跑四十多分钟，Checkpoint 每五分钟一次，中间频繁背压，偶尔还因为某个 TaskManager 抖动整条链路重跑一遍。我看完第一反应是：这个作业的数据明明是有界的，为什么非要按流的方式跑？

把runtime-mode改成BATCH，同样的逻辑、同样的代码，跑完只用了九分钟，Checkpoint 相关的开销直接消失，TaskManager 内存也不再频繁告警。这件事让我意识到，很多人在用 Flink 的时候，其实从来没有认真想过Flink 执行模式这个问题——默认给了流模式就一直用流模式，哪怕面对的是彻底有界的数据。

这篇就聊Flink 的流、批两种执行模式到底差在哪儿、runtime-mode怎么配、背后被改写了哪些机制。内容偏原理和机制拆解，适合已经能写 DataStream 作业、但对执行计划怎么被改掉还没什么概念的同学。看完你至少能判断出：手上这个作业，究竟该跑在哪种模式下。

1. 流与批在 Flink 里不是两套代码，而是两套执行计划

1.1 一次离线报表任务跑得异常慢的排查起点

回到开头那个作业。我先看了它的执行计划，问题一目了然：数据源读完之后，下游的聚合算子和上游是 pipelined 连接的，也就是上游每产出一条记录，下游立刻消费。对于一条持续不断的无界流，这是唯一正确的做法，因为数据永远不会结束，你没法等"全部到齐"。可对于一个明确有界的数据集，这种边生产边消费的方式意味着：没有任何机会做批级别的排序、合并、分阶段优化。

更麻烦的是容错。流模式下作业必须周期性做 Checkpoint，把每个算子的状态快照到远端存储。这个作业的状态不算小，一次快照要几秒，频率又高，等于持续在给集群施加额外压力。而如果按批模式跑，整个作业会被切成若干个阶段，每个阶段跑完把中间结果落盘，某个阶段失败只需要重跑那个阶段，压根不需要 Checkpoint 这套机制。

这就是 Flink 引入执行模式的核心动机：同一份 DataStream 代码，面对无界数据和有界数据时，应该走两条不同的执行路径。前者要求低延迟、持续计算、状态可恢复；后者要求高吞吐、可排序、阶段化调度。硬要用一条路径覆盖两种场景，必然有一边是别扭的。

1.2 同一个 DataStream 图，走两条不同的编译路径

很多人以为执行模式是个运行时开关，其实它在作业提交之前就已经把执行计划改完了。你在代码里写的map、keyBy、window、join，先被翻译成一张StreamGraph，这是一张逻辑图，跟模式无关。真正分叉发生在后面：StreamGraph要被翻译成JobGraph，再被调度器翻译成ExecutionGraph。

在这几步翻译里，执行模式至少影响三件事。第一是连接类型，两个算子之间的数据交换是流水线式的还是阻塞式的，这个在翻译阶段就定死了。第二是算子链的划分策略，流模式下为了减少序列化开销会尽量把算子串成一条链，批模式下要考虑阶段边界，串链的自由度不一样。第三是物理算子的选择，同样一个join，流模式大概率走哈希连接，批模式可能被改写成排序合并连接。

提示：这也是为什么你在批模式下打开 Web UI 看执行计划，会发现算子名字和你代码里写的对不上——不是你的代码有问题，是优化器把它换掉了。

理解这一点很关键：执行模式不是给运行时加个 if-else，而是在编译期就选择了另一套优化规则。所以"我在代码里加了判断，为什么批模式下没生效"这类问题，答案往往是你判断的位置在图生成之后了，那时候计划早定型了。

1.3 ExecutionMode 影响的是图之后的所有环节

把范围说得再宽一点，执行模式一旦确定，下面这些环节全部跟着变：调度器怎么切分任务、失败之后回滚的粒度、中间结果写内存还是写磁盘、内存模型里托管内存占多少比例、算子内部能不能用定时器、Checkpoint 配置是否还有意义。

我见过一个挺典型的误解：有人在批模式作业里认认真真配了execution.checkpointing.interval，还配了 RocksDB 状态后端，然后疑惑为什么从 Savepoint 恢复不了。道理很简单，批模式压根不走 Checkpoint 那套恢复逻辑，配置项还在，但不会产生你期待的效果。这种坑不是配置写错了，而是对模式边界没有概念。

所以在动手改runtime-mode之前，先想清楚这个作业依赖了哪些能力：要不要状态恢复？要不要低延迟输出？数据边界是不是明确的？这三问答完，模式基本就定了。

2. runtime-mode 的三个取值和它们的生效顺序

2.1 STREAMING、BATCH、AUTOMATIC 各自承诺了什么

runtime-mode在 DataStream API 里对应枚举RuntimeExecutionMode，三个值：STREAMING、BATCH、AUTOMATIC。名字很直白，但每个值背后的承诺差别很大，值得逐个说清楚。

STREAMING是默认值，语义是"数据可能永远不结束"。它承诺低延迟、逐条处理、支持完整的状态和定时器、支持 Checkpoint 与从 Savepoint 恢复。代价是吞吐相对低、无法做全局排序类的优化、容错开销持续存在。

BATCH的语义是"数据一定有限，一定会结束"。它承诺阶段化执行、阻塞式数据交换、更强的优化空间、失败按阶段重跑。代价是不支持 Checkpoint 恢复、不支持定时器、某些算子的行为会和流模式不同。

AUTOMATIC是让框架替你选：作业提交时检查所有 Source 的有界性，全都有界就用批模式，只要有一个无界就退回流模式。听着很省心，但它有个隐含前提——你的 Source 必须正确声明自己的有界性，这个后面单独讲。

2.2 代码、命令行、配置文件：优先级与覆盖陷阱

设置方式有三种，优先级从高到低是：代码里显式设置、命令行动态参数、集群配置文件。

代码方式最直接，在构建环境之后立刻设置：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setRuntimeMode(RuntimeExecutionMode.BATCH);

命令行方式不用改代码，适合临时验证：

flink run \ -D execution.runtime-mode=BATCH \ -c com.example.ReportJob \ report-job.jar

配置文件方式是在flink-conf.yaml里加一行execution.runtime-mode: BATCH。我不太推荐这一种，原因是它作用于整个集群上提交的所有作业，很容易出现"某个作业莫名其妙变成了批模式"的排查惨案。除非你维护的是一个专门跑离线作业的独立集群，否则别往全局配置里写。

这里有个很容易踩的坑：代码里的setRuntimeMode会覆盖命令行参数。有人写了个工具类，在初始化环境的时候固定设成了STREAMING，结果运维在命令行传-Dexecution.runtime-mode=BATCH死活不生效，查了半天以为是版本问题。所以如果你希望作业的执行模式由提交方决定，代码里就别写死，留空交给命令行。

2.3 AUTOMATIC 的判定基于 Source 的 Boundedness

AUTOMATIC的判定逻辑不复杂：遍历作业里所有 Source，检查它们声明的有界性，全部是有界就选批，否则选流。关键在于"声明"这两个字——框架判断的是 Source 自己上报的属性，不是你心里认为的属性。

这带来两类常见误判。第一类是自定义 Source 忘了实现有界性声明，默认被当成无界，于是永远走流模式，你以为开了自动模式就自动优化了，其实没有。第二类是多个 Source 混合的作业，比如主数据来自消息队列（无界），维表来自文件（有界），整体判定结果一定是流模式，因为只要有一个无界，全图就不能按批处理。

我在一个数据回刷场景里遇到过一个变体：作业读的是历史分区文件，但用的是一个通用文件读取组件，这个组件在没做额外配置的情况下上报的是无界。改法很简单，在读取时明确指定只读一次、不监听新文件，或者在代码里干脆直接setRuntimeMode(BATCH)。后者更保险，因为语义明确，不依赖组件的实现细节。

注意：如果你只是"希望它自动变快"，而不是"确定数据有界"，那AUTOMATIC帮不了你。判定是机械的，不会猜你的意图。

3. BATCH 模式下被改写的三件事：调度、Shuffle、容错

3.1 调度从"全图并发"退化成"按 stage 推进"

流模式下调度的基本单位是整个作业图，所有算子同时部署、同时启动、同时运行，数据在算子之间流动。批模式完全不同，作业被切分成若干个 stage，stage 之间有明确的先后依赖，前一个 stage 全部跑完，后一个 stage 才能开始。

这种"按 stage 推进"的模型有个直接好处：资源可以分时复用。假设作业有五个阶段，每个阶段需要 20 个并行实例，流模式下你得常驻 20 个 slot 从头跑到尾，批模式下同一批 slot 可以被五个阶段轮流使用。对于离线场景，这意味着同样的集群规模能跑更大的作业。

代价是延迟。批模式下你在 Web UI 里看不到"数据一点点流过来"的效果，只有某个阶段整体完成后，下游才开始动。如果作业的某个阶段特别慢，整个链路就得等它。这也是为什么批模式的调优重点在于找出那个最慢的阶段，而不是像流模式那样盯着背压和反压指标。

顺带提一个实际体验：较新版本里批作业支持自适应调度，会根据上游输出的实际数据量动态决定下游的并行度。这个特性对数据倾斜很友好，但也会带来"我明明设了并行度，为什么 UI 上显示的不是这个数"的困惑。遇到这种情况先去看自适应调度有没有开，别急着怀疑配置写错。

3.2 Shuffle 落地成磁盘文件带来的延迟与稳定

流模式里两个算子之间的数据交换默认是流水线式的，上游产出的数据直接通过网络推给下游，中间不做持久化。批模式下这个行为被改成了阻塞式，上游把结果写成一批文件放在本地磁盘，下游再去拉取。

这个变化带来的影响是双向的。延迟上肯定是变差了，因为要等文件写完。但稳定性上明显更好：首先中间结果落盘之后，下游如果失败重跑，不用再让上游重算一遍；其次阻塞式交换天然形成了反压边界，上游跑得再快也不会把下游冲垮；第三是排序和分组类的操作有了物理基础，只有在数据能落盘、能按 key 排序的前提下，排序合并连接这类算子才成立。

同一份作业，流模式跑可能内存占用很高（因为要维持大量算子状态来缓存未匹配的数据），批模式跑内存占用会平稳很多（数据在磁盘上排好序，按顺序扫）。我第一次把一个大 join 从流切到批，最直观的感受就是 TaskManager 的堆内存曲线从锯齿状变成了平的。

3.3 Checkpoint 在批模式下基本失去意义

批模式的容错逻辑是另一套：某个阶段失败，就重跑这个阶段，凭的是上游阶段已经落盘的中间结果。既然数据都在磁盘上，失败重算的代价是可控的，那就不需要周期性做全量状态快照了。

所以批模式下 Checkpoint 配置不会产生你期待的效果。这里我要强调"期待的效果"这几个字，因为配置项本身还在，日志里也可能不报错，但作业失败时不会从 Checkpoint 恢复，你也别指望用 Savepoint 把批作业的状态迁到另一个作业里。

这一条对迁移工作影响很大。如果你手上的作业原来跑在流模式下，依赖 Savepoint 做版本升级时的状态迁移，那改成批模式之后这条路径就断了。要么保留流模式，要么接受"批作业本身无状态、每次全量重算"这个事实。我在做模式切换评估时，一定会先问一句：这个作业升级的时候，需不需要保留历史状态？需要的话，批模式基本可以直接排除。

3.4 内存账本要重算

流模式和批模式对内存的使用偏好不一样。流模式偏重堆内存，因为算子状态、网络缓冲都在堆上或者托管内存里，比例相对均衡。批模式大量依赖排序、哈希、落盘，托管内存的比例通常要调高一些，否则容易出现"数据量不大但频繁溢写"或者直接 OOM 的情况。

我一般的做法是：切到批模式之后，把托管内存占比往上提一档，同时把网络缓冲适当调小（因为阻塞式交换不需要那么大的常驻缓冲）。这两个方向是相反的，所以别一起猛调，改一个观察一个。最直接的观察指标是任务的重启次数和 GC 时间，如果频繁 Full GC，说明堆不够；如果磁盘 IO 打满但 CPU 很闲，说明托管内存偏小、溢写太多。

这块没有放之四海皆准的数值，跟数据特征强相关。但有一条经验可以复用：批模式作业的内存调优，先看溢写，再看 GC。溢写是批处理的第一性能杀手，往往调完托管内存就能解决大半问题，比反复调堆参数有效得多。

4. 有界流上算子语义的漂移：容易算错的地方

4.1 Watermark 不再推进，窗口靠排序收尾

这是我认为最需要提醒的一点。流模式下窗口能不能触发，取决于 Watermark 有没有越过窗口结束时间。水位线不推进，窗口就一直挂着，数据被缓存在状态里。而批模式下，Watermark 这套机制基本不参与，窗口算子换了一种实现：把属于同一个 key 的数据按时间戳排序，然后一次性把窗口算完。

这意味着两件事。好消息是，批模式下你不需要操心乱序和水位线延迟，所有数据都到齐了才计算，结果一定是完整的。坏消息是，如果作业里有依赖水位线推进来触发的自定义逻辑，比如自己在ProcessFunction里注册的定时器，批模式下这些定时器不会被触发，相关逻辑直接失效，而且往往不报错，就是静默不执行。

我遇到过一次，一个作业用定时器做过期数据清理，切到批模式之后清理逻辑再也没跑过，但因为批作业本身生命周期短，问题被掩盖了很久，直到有人发现输出结果里有一批本该被过滤掉的记录。所以切换模式之前，全局搜一遍onTimer、registerEventTimeTimer、registerProcessingTimeTimer这类调用，有一个算一个，都得单独评估。

还有一个容易忽略的点：批模式下的窗口结果是"最终结果"，不会像流模式那样对同一个窗口做增量输出。如果你的下游依赖窗口的中间触发结果做展示，切批之后展示逻辑会出现空窗期，直到整个作业跑完才一次性出来。

4.2 Keyed State 与 Timer 在批模式的可用边界

这里要区分两个概念：算子内部使用的状态，和你能通过 API 访问的 Keyed State。

算子内部为了完成计算而缓存的数据（比如排序时的缓冲区）当然还是存在的，否则批处理没法干活。但通过KeyedStateStore显式声明的状态，在批模式下的语义是有折扣的：它能读能写，但不能通过 Checkpoint 持久化，也就无法跨作业复用。换句话说，你可以把它当成作业内部的临时存储，但别把它当成长久资产。

定时器的限制更硬一些。批模式下时间语义本身就很模糊——所有数据都已经到齐了，"现在几点"对于计算没有意义。所以定时器不被触发是合理的设计，不是 bug。

实际影响最大的是那些把业务逻辑写进定时器回调的作业。这类逻辑通常长这样：数据进来先缓存，注册一个几秒后的定时器，定时器触发时把缓存的数据批量处理掉。这种"攒批"模式在流里很常见，但在批里完全没必要，因为数据本身就已经是一批了。迁移的时候，这段逻辑要整体重写，而不是简单改个配置就能过去。

4.3 Join、聚合、去重的物理算子替换

物理算子被替换这件事，多数时候是好事，但有几种情况会让结果发生变化。

先说连接。流模式的双流 join 通常走哈希连接，一侧数据常驻状态，另一侧流过来做匹配。批模式更可能走排序合并连接：两边都按 key 排序，然后归并扫描。后者的内存效率高得多，但前提是数据可排序。如果你的 key 类型是自定义对象且没有合理的比较语义，批模式下可能出现排序行为不符合预期，结果顺序和流模式不一致。对于结果集不要求顺序的作业这无所谓，但如果下游依赖顺序，就得留个心眼。

再说聚合。流模式是增量聚合，来一条算一条。批模式可能先分组排序，再对每组做一次聚合。对于sum、count这类可结合的操作结果一致；但对于依赖输入顺序的聚合，比如"取第一条"、"取最后一条"这类，两种模式的语义就可能对不上。这类写法本身在流模式下就不太稳妥，批模式下更明显。

去重也是类似的情况。流模式靠状态记录已见过的 key，批模式靠排序后跳过相邻重复项。结果集相同，但资源消耗特征完全不同，调优的切入点也不一样。

4.4 一份可复现的对照实验

想亲眼看看差异，跑一个最小对照实验就够了。下面这段代码做的事很朴素：造一批有界数据，按 key 分组求和，然后输出。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setRuntimeMode(RuntimeExecutionMode.BATCH); env.setParallelism(2); DataStream<Tuple2<String, Integer>> source = env.fromElements( Tuple2.of("a", 1), Tuple2.of("b", 2), Tuple2.of("a", 3), Tuple2.of("c", 4), Tuple2.of("b", 5), Tuple2.of("a", 6) ); source.keyBy(v -> v.f0) .sum(1) .print(); env.execute("mode-demo");

把setRuntimeMode分别改成STREAMING和BATCH各跑一次，打开 Web UI 对比执行计划，你会看到两个明显不同：一是算子之间的连线类型变了，二是批模式下多了排序相关的算子。再对比一次运行耗时和中间结果的落盘情况，感受会更直观。

如果想更清楚地观察阶段划分，可以在keyBy后面加一个rebalance()，人为制造一次数据重分布，然后在批模式的执行计划里看它把作业切成了几段。这个实验花不了十分钟，但对理解"阶段化执行"这件事的帮助，比读十篇文档都大。

5. 选型与迁移：把作业放到正确的模式上

5.1 三条判断线：数据边界、时效要求、状态依赖

我判断一个作业该跑哪种模式，基本就看三条线，按顺序问下来基本不会错。

第一条，数据边界。Source 会不会自己结束？读文件、读固定分区、读一次性快照，都是有界；读消息队列、读持续写入的表，都是无界。有一个无界，全图就只能流模式，这条是硬约束。

第二条，时效要求。结果需要多久出来一次？如果是天级、小时级的离线产出，批模式完全够用，延迟换成吞吐很划算。如果是秒级、毫秒级的实时响应，流模式是唯一选择，别为了省资源硬上批。

第三条，状态依赖。作业需不需要从历史状态恢复？需不需要跨版本迁移状态？需不需要依赖定时器做业务逻辑？只要有一个"是"，就得慎重，因为批模式在这三件事上都不支持或者支持得不完整。

三条都过了，就可以放心切批。有一条卡住，留在流模式也不丢人——流模式跑有界数据只是浪费，不是错误。

5.2 从 DataSet 迁移过来的作业要盯紧什么

早期用 DataSet API 写的作业，现在基本都在往 DataStream + 批模式迁移。这条路我走过几次，有几个地方必须提前看。

首先要看的是隐式类型推断。DataSet API 大量依赖类型信息做序列化和排序，迁移到 DataStream 之后，如果泛型被擦除或者用了Object类型，批模式下的排序和分组可能直接失败，报错信息还不一定指向根因。我的习惯是迁移时保留显式类型声明，宁可多写几个TypeInformation，也别让框架去猜。

其次看迭代。DataSet 里的迭代是批处理语义，DataStream 的迭代默认是流式语义。批模式下有界迭代是支持的，但写法和参数不一样，需要显式声明最大迭代次数。如果原作业里有迭代逻辑，这块要单独测。

再看分区和排序。DataSet 里的partitionBy、sortPartition这类操作，在 DataStream 里没有一一对应的 API，要么用keyBy加排序算子重组，要么接受批模式自动做的排序优化。别指望能逐行翻译，得按新模型的思路重新组织。

最后一点容易被忽略：DataSet 作业通常没有配置并行度之外的东西，迁到 DataStream 之后如果有人顺手加了 Checkpoint 配置，在批模式下就是无效配置。清理掉，减少误解。

5.3 一份可以直接抄的配置与自查清单

下面这套配置是我在批模式作业里常用的起点，具体数值视集群规模调整，但结构可以照搬。

execution.runtime-mode: BATCH execution.batch-shuffle-mode: ALL_EXCHANGES_BLOCKING taskmanager.memory.managed.fraction: 0.5 taskmanager.memory.network.fraction: 0.1 parallelism.default: 8

说明一下为什么这么配。execution.batch-shuffle-mode明确指定阻塞式交换，避免依赖版本默认值带来的不确定性（较新版本引入了混合模式，虽然在多数场景表现更好，但如果你的作业对中间结果落盘有明确预期，显式指定更稳）。托管内存提到 0.5 是给排序和哈希留空间，网络内存降到 0.1 是因为阻塞式交换不需要大缓冲。

配套的自查清单：

• 全局搜索registerEventTimeTimer、registerProcessingTimeTimer、onTimer，确认没有业务逻辑挂在上面
• 检查是否有KeyedStateBackend相关配置，批模式下这些配置无效，清掉避免误导
• 确认作业不依赖 Savepoint 恢复，或者已经准备了全量重算的方案
• 检查 Source 的有界性声明是否准确，尤其自定义 Source 和多 Source 混合的场景
• 检查自定义 key 类型的比较语义，避免排序结果异常
• 跑一次小数据量验证，对比流、批两种模式的结果集是否一致

提示：清单里的最后一条别省。我见过不止一次，模式切换后结果对不上，最后发现是某个聚合逻辑依赖了输入顺序，流模式下碰巧是对的，批模式下排序一变就露馅了。小数据量对照跑一遍，成本极低，收益很大。

执行模式这块的机制差异，讲到这里基本把"为什么会不一样"这层说清楚了。至于具体到每个算子在两种模式下的行为对照、批模式的性能调优切入点、以及有界无界混跑的场景该怎么拆，涉及的细节还比较多，放到后面一篇展开更合适。如果你现在就着急动手，建议先把第 5 节那份清单过一遍，别急着改配置——很多问题不是配置能解决的，是模型选错了。