企业官网建设流程全解析

从环境搭建到编程模型，这个系列终于写到了第七篇。前几篇我们把 Flink 的部署、核心概念和基础流程理了一遍，到这篇就该正儿八经地把 API 摊开看了。说实话，API 这个话题在 Flink 里特别容易被讲成“文档复读机”——把官方列表抄一遍，读者看完还是不知道用哪个。所以这篇我换一种讲法：以 1.11 版本为主，从做项目的角度出发，讲清楚每一类 API 到底解决什么问题、在什么场景下用、用的时候最容易在哪里栽跟头。

Flink 1.11 是一个比较有意思的版本，既保留了经典的 DataStream API 写法，又把 Blink Planner 定为默认，Table / SQL 的可用性大幅提升。所以这篇会同时覆盖 DataStream 和 Table / SQL 两套主线，再补上 Process Function、状态、侧输出这些底层能力。内容偏向已经在跑 Flink 作业、或者正在写第一个线上作业的同学，看完能直接对着代码模板去改。

1. 先看懂 API 家族：用哪套 API 完成哪种需求

很多新手一上来就纠结“我要学 DataStream 还是学 SQL”，其实不用把它们当成对立面。Flink 的 API 是分层的，从底层到顶层分别是 Process Function、DataStream API、Table / SQL API，层级越低控制力越强，层级越高开发效率越高。1.11 版本里你甚至可以在一份代码里混用它们，不是非此即彼的关系。

1.1 DataStream 与 DataSet：别再用错编程入口

在 1.11 里，DataStream API 依然是处理有界无界数据的主力，而 DataSet API 主要是历史原因留下来的批处理接口。官方其实早就开始弱化 DataSet 的地位，到后面几个版本就彻底和 DataStream 合并了。所以新项目我建议直接忘掉 DataSet，统一用 DataStream 入口。你在 1.11 里写StreamExecutionEnvironment走 DataStream，跑批处理也是通过setRuntimeMode或者直接读取有界数据源，DataSet 那套单独的执行环境不要再碰。

具体写法上，最常见的入口代码就是：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(4);

这里有个细节，getExecutionEnvironment在本地环境会拿到 LocalStreamEnvironment，提交到集群会自动识别成 RemoteStreamEnvironment。很多同学在本地测试没问题，打包丢到集群跑却发现并行度不对，多半是没设置并行度或者集群配置覆盖了本地默认值。

1.2 Table / SQL：1.11 的亮点到底在哪儿

1.11 版本的 Blink Planner 成为默认，这是 Table API 和 SQL 在 Flink 里走向成熟的一个里程碑。和 1.9、1.10 那种实验性质的体验不同，1.11 里你可以在流上直接执行比较复杂的 SQL，包括双流 JOIN、窗口聚合、TopN 这类操作，语法已经非常接近传统数据库。这个版本还正式完善了和 Hive 的集成，能直接复用 Hive 的 Catalog 和 metastore，对已经用 Hive 做离线数仓的团队很友好。

我个人的感觉是，如果业务逻辑能用 SQL 表达，尽量用 SQL，维护成本比 DataStream 低一个数量级。之前带过一个项目，用 DataStream 写了二三十个算子串成一条链路，交接的时候每个人都看得很痛苦，后来重构了一版 SQL，整个链路缩成了几个查询，功能一样，可读性天差地别。

1.3 项目里我习惯怎么选型

我给自己总结了一套很粗糙但好用的判断标准，遇到需求就对着套：

这个表格不是绝对标准，但基本能帮你在 80% 的场景下快速定位。剩下 20% 都是特殊情况，需要两套 API 搭配，后面我会专门讲怎么从 Table 切回 DataStream。

2. 数据进出 API：Source 与 Sink 的常见搭配和自定义方式

一个 Flink 作业的结构说穿了就是 source 进、transformation 算、sink 出。这一节先把两端聊清楚，因为实际开发里很多问题都出在数据接入和写出的配置上，而不是中间算子写错。

2.1 内置连接器先认全

1.11 版本自带的常用连接器大致有这几类：Kafka、JDBC、Elasticsearch、FileSystem、Kinesis、Redis 这种第三方提供的。我自己用得最多的是 Kafka 和 JDBC，前者是数据管道的主力，后者是结果入库最直接的方式。FileSystem 适合做批量落盘或者备份，Elasticsearch 一般配合可视化检索场景。

选连接器有个原则：优先用官方维护的，不要急着自己写。官方连接器经过大量生产环境验证，很多边角问题已经被踩平了。但也要注意版本匹配，1.11 的 Kafka connector 和不同 Kafka 版本的兼容性有差异，最好参照当时官方文档提供的版本矩阵来选，不然碰到序列化或提交 offset 的奇怪问题会比较头疼。

2.2 Kafka Source：老写法和新参数

1.11 里连接 Kafka 的常规写法还是FlinkKafkaConsumer。这里要注意，不要拿后来新版本的KafkaSource写法去套 1.11，KafkaSource是后面几个大版本才推荐的新接口，早期版本并不存在。用 1.11 时老老实实写老接口：

Properties props = new Properties(); props.setProperty("bootstrap.servers", "localhost:9092"); props.setProperty("group.id", "flink-demo-group"); FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>( "input-topic", new SimpleStringSchema(), props ); consumer.setStartFromLatest(); DataStream<String> stream = env.addSource(consumer);

这段代码看起来简单，但有几个细节值得留意。setStartFromLatest和setStartFromEarliest只在第一次消费时起作用，如果 Kafka 消费组里已经记录了 offset，后面的启动策略由已提交的 offset 决定。所以你想重置消费位点，得去改 Kafka 的 offset 或者在代码里用setStartFromSpecificOffsets指定，不是简单改代码就行。

另外，SimpleStringSchema只处理字符串，如果消息是 JSON，建议直接传 String 然后在 map 里解析，不要用自带的 JSON 序列化器，因为错误提示太抽象，出现脏数据时定位很麻烦。

2.3 自定义 Source：如果不想依赖现成连接器

有些业务场景确实需要自己写 Source，比如从一个不支持官方连接器的消息中间件读数据，或者本地文件测试时想生成模拟数据。自定义 Source 的写法从 1.11 到现在变化不大，核心是继承SourceFunction或者RichSourceFunction，实现run和cancel两个方法。

public class RandomStringSource extends RichSourceFunction<String> { private volatile boolean running = true; @Override public void run(SourceContext<String> ctx) throws Exception { while (running) { ctx.collect("msg-" + System.currentTimeMillis()); Thread.sleep(500); } } @Override public void cancel() { running = false; } }

这段代码有一个容易被忽略的点：cancel方法只是把标志位设为 false，如果run方法里卡在某个阻塞操作上，线程不会立刻退出，还会继续占用资源。所以正规做法是在run里捕获中断异常，主动退出循环。之前帮人排查过一个作业，停任务的时候一直显示 CANCELING，状态迟迟不结束，最后发现就是run里的Thread.sleep被中断后没有退出逻辑。

2.4 Sink 侧的幂等性处理

写 Sink 时最容易踩的坑就是“重复写入”。Flink 的 exactly-once 语义在 source 端和算子端做得比较完善，但到了外部系统，如果没有为它设计去重机制，理论上可能重复写入。要解决这个问题，方案通常是两种：一是让外部系统支持幂等写入，比如数据库用主键冲突更新；二是在 Flink 作业里自己维护去重逻辑，比如记录最新一条写到哪里。

我用 JDBC Sink 的做法是每条都写成 upsert 语句，这样即使重复执行也不会产生脏数据。1.11 里 JDBC Sink 的写法是：

stream.addSink(JdbcSink.sink( "INSERT INTO user_stats(user_id, cnt) VALUES (?, ?) ON DUPLICATE KEY UPDATE cnt = VALUES(cnt)", (ps, event) -> { ps.setString(1, event.userId); ps.setLong(2, event.cnt); }, new JdbcExecutionOptions.Builder().withBatchSize(500).build(), new JdbcConnectionOptions.JdbcConnectionOptionsBuilder() .withUrl("jdbc:mysql://localhost:3306/test") .withDriverName("com.mysql.cj.jdbc.Driver") .withUsername("root") .withPassword("your_password") .build() ));

注意 1.11 的JdbcSink需要显式配置withDriverName，否则可能在提交到集群后找不到驱动。另外withBatchSize不是越大越好，我实测过 500 到 1000 是性价比最高的区间，再大容易出现单批次执行时间过长，影响端到端延迟。

3. 转换算子 API 详解：真正让数据“流”起来的部分

中间这一大块是 DataStream API 的核心转换算子，也是出现频率最高的代码段。很多初学者把 map、flatMap、keyBy、window 背得滚瓜烂熟，但真到项目里就发现问题不断，这一节我就挑最容易出错的几个点展开讲。

3.1 map / filter / flatMap：基础算子里的几个坑

这三个算子算是 Flink 里的“三小只”，基础到不能再基础，但坑反而都在细节里。第一个坑是匿名内部类捕获外部变量导致序列化失败。Flink 集群会对算子做序列化分发，如果 map 的匿名内部类里捕获了一个非序列化的对象，作业启动时会直接报序列化异常。我见过有人在这个问题上头疼一整天的：

// 错误示例 String prefix = "prefix-"; DataStream<String> result = stream.map(s -> prefix + s);

这段代码在有些环境能跑，在有些环境会报异常，原因是 Lambda 表达式里捕获了prefix，它默认被当作非静态内部类处理。解决办法很简单，要么把prefix声明为static final并确保可序列化，要么自定义一个实现MapFunction的类，把变量作为成员传进去。

第二个坑是flatMap不好好用。很多人在flatMap里自己也搞不清楚到底要输出几条数据，随手往 collector 里塞。其实flatMap最适合做“一对多”或“一对零”的转换，它天然承载了 filter 和 map 两块能力。如果你只是把元素过滤掉，写 filter 更清晰；如果你要把一条 JSON 拆成多条记录，那才轮到flatMap上场。

3.2 keyBy 到底做了什么

keyBy是 DataStream API 里最容易产生误解的算子，没有之一。很多人以为keyBy之后数据就“分组”了，后续操作都在组内进行，这个理解不能算错，但不够精确。keyBy本质上是一个分区操作，它通过 key 的哈希值把相同 key 的数据分到同一个子任务上。只有到了同一个子任务里，后面跟着的状态和窗口才是按 key 隔离的。

这里有三个细节值得记住：

• keyBy之后如果直接跟 map，map 是看不出分组的，它依然是一条条处理数据，只是同一 key 的数据会汇聚到同一并行子任务里。
• keyBy的 key 如果为 null，会发生数据倾斜，因为 null 的哈希值是一样的，所有 null 都会涌向同一个子任务。
• keyBy使用的 key 类型必须正确实现hashCode和equals，否则相同的业务 key 会被当成不同的 key，状态和窗口全部失效。

实际项目中数据倾斜大多出现在keyBy这一步。比如日志里的 user_id 字段大部分为空，或者集中在几个大用户上，导致某个子任务积压。缓解办法是先看 key 分布，再考虑加随机前缀打散，但要注意打散后就不适合做精确聚合了，得两阶段聚合。

3.3 窗口聚合：Window API 怎么选

窗口是流处理的灵魂。在 1.11 版本里，窗口分为时间窗口和计数窗口，时间窗口又按照 event time、processing time 区分。写代码时你面对的选择主要是四种：TumblingEventTimeWindow、SlidingEventTimeWindow、SessionWindow，以及它们的 processing time 变体。

对应到代码层面，最常见的窗口聚合是这种方式：

DataStream<Event> streamWithWatermark = stream .assignTimestampsAndWatermarks( WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((event, ts) -> event.getTimestamp()) ); DataStream<CountResult> windowed = streamWithWatermark .keyBy(Event::getUserId) .window(TumblingEventTimeWindows.of(Time.minutes(1))) .aggregate(new CountAggregate()) .name("user-count-window");

这里我建议优先使用aggregate而不是apply。aggregate是增量聚合，来一条数据算一条，内存占用低；apply是先把整个窗口的数据都攒下来，到触发时再统一处理，窗口越大越危险。有人图省事，窗口设为 1 天，然后用apply做统计，结果内存直接爆掉。增量聚合解决不了的问题再去想ProcessWindowFunction，它适合需要访问窗口元数据和全局状态的情况。

3.4 union / connect / coGroup：多流操作怎么区分

多流操作在 DataStream API 里也有好几个选择，很多同学搞不清楚。这里我按自己的理解整理了一个判定流程：

• 多个流的类型相同，且只要做简单合并，用union。
• 多个流的类型不同，需要各自处理后再合并，或者需要共享状态做 join，用connect，配合CoProcessFunction。
• 多个流要做类似 SQL join 的操作，按 key 匹配后做窗口内合并，用coGroup。

union是最省力的，但注意要求流的类型完全一致。connect则灵活得多，两个流各管各的，可以通过CoProcessFunction互相传递数据，经常用在双流匹配的场景。至于coGroup，它的语义比connect更接近传统 join，使用时要指定同一个 key 和一个共同的窗口，适合做对账类的需求。

4. 进阶核心 API：Process Function、状态与侧输出

走到这一步，你已经把 Flink 的“常用 API”用得差不多了。但如果只停留在 DataStream 的现成算子，很多复杂业务还是写不出来。Process Function 才是 Flink 真正放开手脚让你自定义处理的入口。

4.1 为什么要学 Process Function

Process Function 和 map 这类算子的最大差别是：它把流处理里的三样东西直接暴露给你——事件本身、当前算子的状态、以及定时器服务。你可以把 Process Function 理解成一辆手动挡的车，map 是自动挡，帮你把换挡逻辑封装好了；Process Function 让你自己踩离合、挂挡、给油，代价是操作复杂，上限也更高。

我最常用 Process Function 的场景是“等待某个外部事件然后超时”。比如支付系统中，用户下单后等待支付成功事件，如果 5 分钟没等到，就输出一条超时告警。这个逻辑用 map 做不了，因为你不知道什么时候该判断超时，必须依赖定时器。用ProcessFunction就能很自然地在收到下单事件时注册一个 5 分钟后的定时器，到时间了去状态里查一下支付状态。

4.2 KeyedProcessFunction 的定时器用法

定时器在 Flink 里分为处理时间定时器和事件时间定时器，分别对应registerProcessingTimeTimer和registerEventTimeTimer。注意，只有KeyedStream上才能注册定时器，普通非 keyed 流的 Process Function 里不提供TimerService。

下面这是一个常用的超时判断模板，我自己在项目里改过很多遍：

public class OrderTimeoutFunction extends KeyedProcessFunction<String, OrderEvent, String> { private ValueState<OrderEvent> orderState; @Override public void open(Configuration parameters) { orderState = getRuntimeContext().getState( new ValueStateDescriptor<>("order-state", OrderEvent.class) ); } @Override public void processElement(OrderEvent value, Context ctx, Collector<String> out) throws Exception { orderState.update(value); ctx.timerService().registerProcessingTimeTimer( ctx.timerService().currentProcessingTime() + 300_000L ); } @Override public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception { OrderEvent event = orderState.value(); if (event != null && "CREATED".equals(event.getStatus())) { out.collect("订单超时: " + event.getOrderId()); } orderState.clear(); } }

这个代码有几个关键点：open里初始化状态只能在富函数里做；processElement里既更新状态又注册定时器；onTimer触发时读取状态判断业务条件。如果你发现定时器触发时状态已经丢了，优先检查是不是忘了设置setStateTtlConfig或者状态后端有问题，而不是怀疑 Flink 的定时器机制本身。

4.3 状态 API：ValueState / ListState / MapState

状态的合理使用是 Flink 进阶的重头戏。1.11 提供了几种状态原语，最常用的三种是ValueState、ListState、MapState。简单来说：

• ValueState：存一个对象，适合保存一条关键的中间结果。
• ListState：存一个列表，适合积攒多条数据。
• MapState：存一个 Map，适合按子键保存多种不同维度的信息。

很多人会纠结什么时候用MapState什么时候用ListState。一个经验法则：如果你需要按某个子键去更新或查询，就用MapState；如果你只需要按顺序全部遍历，用ListState。比如统计一个用户最近 10 次点击，用ListState就够了；但要按商品维度分别统计，肯定要上MapState。

另外要提醒一句，状态不会自动清理，除非你配置了 TTL。1.11 里设置 TTL 的方式是：

StateTtlConfig ttlConfig = StateTtlConfig .newBuilder(Time.days(3)) .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) .build(); ValueStateDescriptor<String> desc = new ValueStateDescriptor<>("state", String.class); desc.enableTimeToLive(ttlConfig);

TTL 的机制是在读写状态时检查过期时间，不是后台定时扫描，所以过期数据在下次访问时才会被清理或者过滤掉。往状态里塞大量无界数据而不设 TTL，是 Flink 作业内存持续上涨的常见原因之一。

4.4 Side Output：把一条流拆成多条

一条流有时候需要按条件拆成好几条子流，比如日志中把正常日志、异常日志、告警日志分别送到不同的下游。Flink 的 side output 正是干这个的工具，它让你在一个算子内部通过ctx.output输出到不同的旁路，主路还能继续往下流。

用法分三步：先定义OutputTag，再在 Process Function 里输出，最后从主流取旁路：

OutputTag<String> alarmTag = new OutputTag<String>("alarm") {}; DataStream<String> output = stream .process(new ProcessFunction<Event, String>() { @Override public void processElement(Event value, Context ctx, Collector<String> out) { if (value.isError()) { ctx.output(alarmTag, value.toString()); } else { out.collect(value.getName()); } } }); DataStream<String> alarmStream = output.getSideOutput(alarmTag);

这里有个重要细节：OutputTag必须用匿名内部类的形式加一对花括号，保留泛型类型信息，不加大括号会让 Flink 在运行时拿不到具体类型，经常在序列化时出问题。另外，你可以在 Process Function 的onTimer方法里也调用ctx.output，所以旁路输出还能用来发定时告警，非常灵活。

5. Flink 1.11 的 Table / SQL API：流批一体实战

DataStream API 讲完之后，咱们回过来看 Table / SQL。1.11 出的最大价值就是让 SQL 在流处理里真正可用，这一节给一些可直接参考的落地写法。

5.1 动态表的基础认知

Table API / SQL 处理流数据时的核心抽象叫动态表。流的每一条数据进入动态表，相当于往表里插入一行记录；当这个表是更新表时，来一条数据还可能更新之前某一行。理解这一点对后续操作很重要，因为同样是 SQL 查询，不同的表模式会影响你最后怎么把结果转回 DataStream。

如果上游是一个追加流，也就是只会插入不会更新，那查询结果大概率也是追加模式。但如果上游是更新流，比如 Kafka 里存的是 binlog，或者你的查询里有聚合、双流 JOIN，那么结果就是更新流。这时你要把它转回 DataStream，就不能用追加模式那套接口。

5.2 实战一个 SQL 查询和窗口

假设我们有一个user_log表，包含 user_id、url、ts 字段，想统计每十分钟每个用户的访问次数。用 Table API 创建视图再来 SQL 查询，1.11 的写法如下：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); EnvironmentSettings settings = EnvironmentSettings.newInstance() .useBlinkPlanner() .inStreamingMode() .build(); StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings); DataStream<String> rawStream = env.addSource(kafkaSource); Table events = tEnv.fromDataStream(rawStream, "user_id, url, ts"); tEnv.createTemporaryView("user_log", events); Table result = tEnv.sqlQuery( "SELECT user_id, COUNT(*) AS cnt, " + "TUMBLE_START(ts, INTERVAL '10' MINUTE) AS win_start, " + "TUMBLE_END(ts, INTERVAL '10' MINUTE) AS win_end " + "FROM user_log " + "GROUP BY user_id, TUMBLE(ts, INTERVAL '10' MINUTE)" );

这里用到的TUMBLE_START、TUMBLE_END就是 1.11 里 tumble 窗口的写法。要注意fromDataStream后面带的是字段名列表，如果字段名和上游 DataStream 里的 POJO 属性不一致，得在 as 里改名或在这里指定。在 1.11 里，SQL 窗口还不支持后面新版本那种 Table Valued Function 语法，不要混用。

查询出来的结果如果是更新流，转回 DataStream 要用toRetractStream：

DataStream<Tuple2<Boolean, Row>> retractStream = tEnv.toRetractStream(result, Row.class);

Boolean字段表示这条数据是新增还是撤销，加 1 和减 1 的语义都要处理，否则下游累加的时候数字会不对。如果只是做简单过滤、字段映射这种必然追加的操作，才适合用toAppendStream。

5.3 双流 JOIN 和时态表

1.11 的 SQL 已经可以做双流 JOIN，但要区分窗口 JOIN 和时态表 JOIN。窗口 JOIN 要求两条流的时间窗口能对上，时态表 JOIN 更适合“维表关联”的场景。比如订单流要和实时汇率表关联，汇率是一个会变动的表，这时候需要取下单时间那一刻的汇率，也就是时态表语义。

时态表关联的典型 SQL 是这样的：

SELECT o.order_id, o.amount, r.rate FROM orders o LEFT JOIN currency_rate FOR SYSTEM_TIME AS OF o.proc_time AS r ON o.currency = r.currency

这个写法的意思是：订单流里的每条数据，都去关联currency_rate表在它处理时间当时的状态。这里有一个很大的坑，就是FOR SYSTEM_TIME AS OF的字段如果是处理时间，意味着关联只会查到当时已经到达的汇率，晚到的汇率变更不会影响已经关联过的订单。如果业务上需要严格的“事件时间版本”，要在上游生成 watermark 并指定事件时间字段，复杂度会明显上升，建议先确认需求再决定。

5.4 从 Table 再切回 DataStream

说了这么多 SQL，还有一个常见的场景是：想用 SQL 做一部分处理，但后面还要用 Process Function 做定时器、状态这些 SQL 不方便表达的逻辑。这时候就要把 Table 转回 DataStream。

前面说过，追加流用toAppendStream，更新流用toRetractStream。收到结果后你可以继续对它做 keyBy、再调用 Process Function，等于整个作业是“DataStream -> Table -> DataStream”的一条混合链路。我觉得这种混用的模式特别适合复杂项目，既能享受 SQL 的高效表达，又不至于被 SQL 的限制卡死。

6. 常见问题与排查技巧实录

这部分是我最想写的。Flink 写起来容易，跑起来难，好多问题都是生产环境踩出来的。这里汇总了一些高频率问题，每个都是我或者合作团队亲身碰到过的。

6.1 本地跑不起来：入手最快的几个错误

本地跑 Flink 作业最常见的三个错误是：依赖冲突、端口占用、环境变量没配。依赖冲突一般出在flink-connector-kafka带进来的 kafka-clients 版本和项目里其他组件冲突，解决办法是统一版本号，不要各引各的。端口占用会报“无法分配端口”之类的错误，跑多个本地任务时尤其明显。环境变量主要是JAVA_HOME和 Hadoop 相关依赖，在没有 Hadoop 环境的机器上跑，要把相关依赖设置成 provided 或者手动补齐。

另外，本地默认并行度是你机器核数，几核就几个并行度。如果代码里用了setParallelism(1)但 watermarks 分配在 source 之后，你会发现数据不流动，这个和并行度没关系，多半是 watermark 策略没生效，见下面一条。

6.2 水位线不生效怎么办

我见过不少同学的作业，明明 TUMBLE 窗口设了事件时间，但窗口就是不触发。排查第一步先看 watermark 是否被正确赋值，最简单的办法是在界面上看 watermark 值，或者临时打印一下。1.11 里可以在 ProcessFunction 里拿到ctx.timerService().currentWatermark()然后打日志，看它是Long.MIN_VALUE还是有正常递增。

如果 watermark 一直是最大值或最小值，就检查两个地方：一是assignTimestampsAndWatermarks是否真的执行了，二是并行度下 watermark 的对齐机制。并行度大于 1 时，Flink 的 watermark 是在每个 subtask 独立生成的，下游要等所有分区的 watermark 都过了阈值才触发窗口。如果某个 source 分区一直没有新数据，它的 watermark 就会停留在初始值，拖住整个窗口不触发。这种场景要用WatermarkStrategy.withIdleness(Duration.ofSeconds(30))，让空闲分区自动推进 watermark。

6.3 状态与 checkpoint：越调越稳

状态没配好很容易出问题。首先确认你的状态后端，1.11 默认是内存后端，内存不够直接报 OOM。生产环境我会用 RocksDBStateBackend，然后配合 HDFS 做 checkpoint 持久化：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setStateBackend(new HashMapStateBackend()); env.getCheckpointConfig().setCheckpointStorage("hdfs://namenode:8020/flink-checkpoints"); env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500); env.getCheckpointConfig().setCheckpointTimeout(60000); env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

注意 1.11 里HashMapStateBackend和旧版MemoryStateBackend是共存关系，写法上可以按上面的新写法来。设置setMinPauseBetweenCheckpoints很重要，不加这个参数，两个 checkpoint 可能挤在一起，导致磁盘 IO 打满。如果 checkpoint 一直失败，最常见的两个原因是状态太大和作业频繁发生 failover，后者又和反压、数据倾斜有很强的关联。

6.4 背压和数据倾斜怎么查

背压是 Flink 里最让人头疼的现象之一，它表现为某个算子处理不过来，导致上游被按住，整个任务延迟上升。排查背压先在 Web UI 上看每个算子的 Back Pressure 状态，颜色发红的就是瓶颈。最常见的瓶颈在外部 IO，比如写入数据库太慢、维表查询太慢，再不然就是算子并行度不够。

数据倾斜和背压类似，但它更偏向某个 key 或某个子任务特别忙。处理思路分两步：先定位是不是 key 分布问题，然后决定用打散 key 还是调整并行度。如果是窗口聚合导致倾斜，可以考虑两阶段聚合，先加随机前缀局部聚合，再去前缀全局聚合。Java 代码写起来会复杂一点，但效果立竿见影。之前我用这个方法把一个 6 分钟延迟的作业降到了 40 秒以内。

6.5 一份快速定位的报错速查表

这张表没有覆盖所有问题，但遇到最多的是这几类。每次排查时先按类别缩小范围，不要一上来就怀疑框架本身。

最后再分享一个小习惯：不管用什么 API，我都建议给关键算子都加上.name("xxx")，方便在 Web UI 上直视作业拓扑。很多同学觉得这是额外功夫，等到了排查问题那一步才会意识到它的价值。Flink 作业越复杂，这个习惯越值钱。