1. 这不是又一篇“Hello World”式Flink教程——为什么你学了三遍 still 写不出能跑通的DataStream作业

Apache Flink 的入门资料铺天盖地，但绝大多数人卡在同一个地方：照着官网示例敲完代码，本地mvn clean compile exec:java能跑，一提交到集群就报ClassNotFoundException、NoClassDefFoundError，或者更隐蔽的——数据流压根没触发计算，print()里空空如也，日志里连个SourceReader初始化的痕迹都找不到。我见过太多某高校实验室的A同学，把《Flink权威指南》翻烂了，对着“WordCount”抄了七遍，第八次想加个自定义窗口函数时，发现连KeyedStream是怎么从DataStream变出来的都说不清楚。这不是学习态度问题，是绝大多数入门材料默认你已经理解了 Flink 运行时模型、序列化机制、并行度与算子链、以及 DataStream API 最核心的有状态流处理契约——而这些，恰恰是 DataStream API 和 Spark Streaming 或 Kafka Streams 最本质的区别。

这篇内容不讲“什么是流处理”，不堆概念图，也不复述官网那套“Source-Transform-Sink”三段论。它直接从一个真实可运行、可调试、可部署的最小闭环项目出发：用SocketTextStreamFunction模拟实时日志源，通过flatMap拆词、keyBy分组、window聚合、process自定义处理逻辑，最终输出到控制台和文件。每一步，我都告诉你为什么必须这么写，而不是“按文档做就行”。比如，为什么keyBy后面必须跟window才能调用sum()？为什么process函数里ctx.timerService().currentProcessingTime()返回的值，在本地 IDE 和集群上可能差出几秒？为什么env.setParallelism(1)在开发阶段是救命稻草，但上线前必须改掉？这些不是边角料，而是你后续写反欺诈、实时风控、IoT 设备告警等真实业务逻辑时，每天都要面对的底层约束。关键词就三个：DataStream API、状态一致性、并行执行模型——它们像三根钢筋，撑起了整个 Flink 流处理应用的骨架。如果你现在写的 Flink 作业还停留在“能跑就行”，那这篇就是为你量身定制的补丁。

2. 从零构建一个可调试、可验证的 DataStream 项目：环境、依赖与第一个“活”起来的流

2.1 项目骨架：Maven 依赖的取舍逻辑与版本陷阱

别急着写main()方法。先解决一个最基础却最容易翻车的问题：你的pom.xml里，到底该引入哪些依赖？网上很多教程一股脑贴出七八个flink-*的 dependency，结果编译时报错“冲突”、“循环依赖”。根本原因在于没分清 Flink 的运行时依赖和开发时依赖。我们只保留最精简、最安全的组合：

<properties> <flink.version>1.18.1</flink.version> <scala.binary.version>2.12</scala.binary.version> <java.version>11</java.version> </properties> <dependencies> <!-- 核心运行时：DataStream API 的基石 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>${flink.version}</version> </dependency> <!-- 本地测试必需：提供 MiniCluster，让你在 IDE 里就能启动一个嵌入式 Flink 集群 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-runtime-web</artifactId> <version>${flink.version}</version> <scope>test</scope> </dependency> <!-- 日志桥接：避免 SLF4J 绑定冲突，这是本地调试不报红的关键 --> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-simple</artifactId> <version>1.7.36</version> </dependency> </dependencies>

提示：绝对不要引入flink-clients或flink-table-api-java。前者是 JobClient 的封装，对初学者是干扰项；后者属于 Table API 层，和 DataStream API 是平行关系，混用会引发类加载器混乱。flink-runtime-web加了<scope>test</scope>，意味着它只在mvn test时生效，打包进生产 jar 包时会被自动剔除，避免和集群上的 Flink runtime 冲突。

2.2 第一个“活”的流：Socket 源 + 控制台 Sink 的完整闭环

现在，写一个真正能“动起来”的程序。不是System.out.println("Hello Flink")，而是让数据从网络端口进来，经过处理，再打到控制台。关键在于：所有操作必须在一个StreamExecutionEnvironment实例中完成，且execute()必须被调用。这是 Flink 的硬性契约，漏掉任何一点，程序都会静默退出。

public class SocketWordCount { public static void main(String[] args) throws Exception { // 1. 创建执行环境：这是整个 DataStream 程序的“容器” final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 2. 设置并行度为 1：开发阶段的黄金法则，避免多线程调试地狱 env.setParallelism(1); // 3. 定义 Source：从本地 9999 端口读取文本行 // 注意：这里用的是 flink-streaming-java 内置的 SocketTextStreamFunction // 不需要自己写 Netcat，也不需要额外依赖 DataStream<String> textStream = env.socketTextStream("localhost", 9999); // 4. 定义 Transform：拆词、转小写、过滤空字符串 DataStream<Tuple2<String, Integer>> wordCountStream = textStream .flatMap(new LineSplitter()) // 自定义 FlatMapFunction .keyBy(value -> value.f0) // 按单词分组 .sum(1); // 对每个 key 的第二个字段（count）求和 // 5. 定义 Sink：打印到控制台 wordCountStream.print(); // 6. 触发执行：没有这行，上面所有代码只是“定义”，不会运行！ env.execute("Socket WordCount Job"); } // 自定义 FlatMapFunction：将一行文本拆成多个 (word, 1) 元组 public static class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> { @Override public void flatMap(String value, Collector<Tuple2<String, Integer>> out) { String[] words = value.toLowerCase().split("\\W+"); for (String word : words) { if (!word.isEmpty()) { out.collect(new Tuple2<>(word, 1)); } } } } }

注意：env.socketTextStream("localhost", 9999)这行代码背后，Flink 会自动启动一个SocketTextStreamFunction，它内部封装了一个SocketClient，持续监听端口。你不需要手动开nc -lk 9999，Flink 的socketTextStream会帮你搞定连接和重试。但前提是，你的本地localhost:9999端口没有被其他程序占用。如果报Connection refused，请先检查端口占用情况。

2.3 本地调试：如何让 Flink 在 IDE 里“呼吸”起来

很多人以为 Flink 只能在 YARN/K8s 上跑，其实它的MiniCluster是为本地调试而生的。在 IDEA 中运行上述main()方法，你会看到控制台疯狂滚动日志，其中最关键的一行是：

INFO org.apache.flink.runtime.minicluster.MiniCluster - Starting Flink Mini Cluster.

这表示一个完整的、单 JVM 进程内的 Flink 集群（JobManager + TaskManager）已经启动。此时，打开另一个终端，执行：

echo "hello world hello flink" | nc -w 1 localhost 9999

立刻，你的 IDEA 控制台就会输出类似：

1> (hello,2) 1> (world,1) 1> (flink,1)

这里的1>表示这是并行度为 1 的 task 的输出。如果你把env.setParallelism(1)改成2，输出会变成1>和2>交替出现，而且同一个单词（如hello）可能被分到不同 subtask，导致计数不准——这就是为什么keyBy是必须的：它确保相同 key 的数据一定被发送到同一个 subtask 进行处理。

3. 理解 DataStream 的“状态”与“时间”：为什么window不是可选项，而是必选项

3.1 无界流的本质困境：没有window，sum()就是个死循环

DataStream处理的是无界流（Unbounded Stream），这意味着数据永远不会“结束”。那么，sum()操作要对什么求和？对从作业启动到现在的所有数据？这显然不现实，内存会爆，结果也无法落地。window就是 Flink 给出的答案：它把无限的数据流，按照某种规则（时间或数量）切分成有限的、可管理的“桶”（Bucket）。sum()操作的对象，不是整个流，而是每一个独立的 window。

在上一个例子中，sum(1)看似简单，但它隐含了一个前提：Flink 默认使用GlobalWindow，即一个覆盖所有数据的、永不关闭的窗口。这在socketTextStream这种测试场景下“似乎”能工作，但一旦换成 Kafka 或 Pulsar 这样的真实消息队列，GlobalWindow会导致状态无限增长，最终 OOM。所以，我们必须显式定义一个有生命周期的窗口。

3.2 时间语义的选择：Processing Time vs Event Time，一次选错，全盘皆输

Flink 支持三种时间语义：Processing Time（处理时间）、Event Time（事件时间）、Ingestion Time（摄入时间）。对于初学者，Processing Time 是唯一安全的起点。它的定义极其简单：System.currentTimeMillis()。Flink 的TimerService会基于这个时间来触发窗口关闭。

// 修改上例中的 Transform 部分，加入 5 秒滚动窗口 DataStream<Tuple2<String, Integer>> wordCountStream = textStream .flatMap(new LineSplitter()) .keyBy(value -> value.f0) .window(TumblingProcessingTimeWindows.of(Time.seconds(5))) // 关键：5秒滚动窗口 .sum(1);

这段代码的意思是：每 5 秒，Flink 就会把这 5 秒内收到的所有(word, 1)数据，按word分组，然后对每个word的count字段求和。窗口是“滚动”的（Tumbling），意味着窗口之间不重叠，5-10秒的窗口和10-15秒的窗口是完全独立的。

提示：TumblingProcessingTimeWindows.of(Time.seconds(5))中的Time.seconds(5)并不是一个简单的整数 5，它是一个Time对象，内部封装了毫秒值和时间单位。Flink 用这种方式保证了跨平台（如 Windows/Linux）的时间精度一致性。如果你直接传5000，编译会失败。

3.3 窗口触发的“心跳”：如何观察窗口是否真的在工作

光写代码不够，你得亲眼看到窗口在“跳动”。最简单的方法，是在sum()后加一个print()，然后用nc持续发数据：

# 在一个终端持续发送 while true; do echo "hello"; sleep 1; done | nc -w 1 localhost 9999

在 IDEA 控制台，你会看到输出不是连续的，而是每隔 5 秒“爆发”一次：

1> (hello,5) # 这是第1个5秒窗口的汇总 1> (hello,5) # 这是第2个5秒窗口的汇总 1> (hello,5) # 这是第3个5秒窗口的汇总

这证明窗口机制正在工作。如果输出是(hello,1)、(hello,1)连续不断，说明window没生效，大概率是keyBy写错了位置，或者window调用在了sum()之后。

4. 从print()到生产级 Sink：文件输出、状态后端与容错保障的实操细节

4.1 文件 Sink 的坑：为什么StreamingFileSink不能直接用printToTextFile

很多初学者想把结果写到文件，第一反应是stream.writeAsText("output.txt")。这是 Flink 1.11 之前的旧 API，早已废弃。新 API 是StreamingFileSink，但它比print()复杂得多，因为文件系统是有状态的、需要事务保证的。

// 正确的文件输出方式（仅限本地文件系统） StreamingFileSink<String> fileSink = StreamingFileSink .forRowFormat(new Path("file:///tmp/flink-output"), new SimpleStringEncoder<String>("UTF-8")) .withRollingPolicy( DefaultRollingPolicy.builder() .withRolloverInterval(TimeUnit.MINUTES.toMillis(15)) // 15分钟切一个文件 .withInactivityInterval(TimeUnit.MINUTES.toMillis(5)) // 5分钟没数据就切 .withMaxPartSize(1024 * 1024 * 1024) // 1GB .build() ) .build(); wordCountStream.map(tuple -> tuple.f0 + ": " + tuple.f1).addSink(fileSink);

注意：file:///tmp/flink-output这个路径，前面的file://协议头绝对不能省略。Flink 会根据协议头选择不同的FileSystem实现。省略后，Flink 会尝试用 HDFS 的FileSystem去访问本地文件，导致UnknownHostException。另外，StreamingFileSink默认使用Buckets模式，会在/tmp/flink-output下生成part-0-0、part-0-1等文件，这是正常行为，不是 bug。

4.2 状态后端的抉择：MemoryStateBackend为什么只适合“玩具”

Flink 的状态（State）是容错的核心。keyBy后的sum()计数，本质上就是一种ValueState<Integer>。Flink 必须把状态存起来，才能在故障恢复时重建计算结果。状态后端（State Backend）就是负责这件事的组件。

• MemoryStateBackend：把状态存在 JVM Heap 里。优点是快，缺点是无法用于生产，因为 Heap 大小有限，且无法做 Checkpoint。
• FsStateBackend：把状态快照（Checkpoint）存到文件系统（如 HDFS、S3、甚至本地磁盘）。这是生产环境的标配。
• RocksDBStateBackend：把状态存在本地 RocksDB（一个嵌入式 KV 数据库）里，Checkpoint 存到远程文件系统。它支持超大状态（TB 级），是大型作业的首选。

对于本地开发，我们用FsStateBackend指向本地目录即可：

// 在 env 创建后，execute() 之前添加 env.setStateBackend(new FsStateBackend("file:///tmp/flink-checkpoints")); env.enableCheckpointing(10_000); // 每10秒做一次 Checkpoint

提示：enableCheckpointing(10_000)的参数单位是毫秒，不是秒。写成10是无效的，Flink 会忽略。10_000是 Java 8 引入的数字字面量分隔符，等价于10000，但可读性更好。

4.3 容错的“最后一公里”：如何验证 Checkpoint 真的在工作

光配置还不够，得看到 Checkpoint 成功的证据。在 IDEA 的控制台日志里，搜索Checkpoint completed，你应该能看到类似：

INFO org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed checkpoint 1 for job 00000000000000000000000000000000.

这表示 Checkpoint 1 已成功写入/tmp/flink-checkpoints目录。你可以手动ls -l /tmp/flink-checkpoints/，会看到一个以chk-1命名的子目录，里面就是本次 Checkpoint 的二进制快照文件。如果日志里只有Starting checkpoint，却没有Completed checkpoint，说明 Checkpoint 被阻塞了，最常见的原因是StreamingFileSink的rolling policy设置得太激进（如withRolloverInterval设为 100ms），导致文件频繁切换，拖慢了 Checkpoint 进程。

5. 从开发到部署：打包、提交与集群日志定位的全流程避坑指南

5.1 Maven 打包：shade插件的正确用法与 ClassLoader 陷阱

本地能跑，不等于集群能跑。最大的鸿沟在于类路径（Classpath）。Flink 集群有自己的lib/目录，里面放着flink-streaming-java-*.jar等核心依赖。如果你的 jar 包里也包含了这些依赖，就会发生ClassLoader 冲突，导致NoSuchMethodError或IncompatibleClassChangeError。

解决方案是：只打包你的业务代码，不打包 Flink 的依赖。maven-shade-plugin就是干这个的：

<plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <version>3.4.1</version> <executions> <execution> <phase>package</phase> <goals> <goal>shade</goal> </goals> <configuration> <artifactSet> <excludes> <!-- 排除所有 flink-* 依赖，它们由集群提供 --> <exclude>org.apache.flink:*</exclude> <!-- 排除 slf4j，避免日志桥接冲突 --> <exclude>org.slf4j:*</exclude> </excludes> </artifactSet> <transformers> <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"> <mainClass>SocketWordCount</mainClass> </transformer> </transformers> </configuration> </execution> </executions> </plugin>

执行mvn clean package后，生成的target/*.jar文件大小应该只有几百 KB，而不是上百 MB。如果看到flink-streaming-java-1.18.1.jar被打包进去了，说明excludes配置没生效，必须修正。

5.2 提交到 Standalone 集群：flink run命令的参数详解

假设你已经启动了一个 Flink Standalone 集群（一个 JobManager + 两个 TaskManager），其 Web UI 地址是http://localhost:8081。提交作业的命令是：

flink run \ --jobmanager localhost:8081 \ --class SocketWordCount \ target/flink-demo-1.0-SNAPSHOT.jar

• --jobmanager：指定集群地址，格式是host:port，不是 URL。
• --class：指定主类，必须和MANIFEST.MF里的Main-Class一致。
• target/flink-demo-1.0-SNAPSHOT.jar：你的 fat jar。

提示：flink run命令本身不输出作业日志，它只返回一个JobID。真正的日志在集群的log/目录下，或者通过 Web UI 的Jobs -> YourJob -> Logs查看。如果作业启动失败，Web UI 的Overview页面会显示红色错误信息，点进去就是堆栈。

5.3 日志定位：当NoClassDefFoundError出现时，你该看哪几行

这是集群上最经典的错误。假设你看到：

Caused by: java.lang.NoClassDefFoundError: org/apache/flink/api/common/functions/FlatMapFunction

不要慌。这不是你的代码错了，而是flink-streaming-java依赖没被正确排除。定位步骤如下：

1. 在 Web UI 的Jobs -> YourJob -> Exceptions标签页，找到完整的堆栈。
2. 找到Caused by:这一行，它指明了缺失的类。
3. 往上翻，找到at SocketWordCount$LineSplitter.<init>(SocketWordCount.java:XX)这一行，确认是哪个类触发了加载。
4. 最关键的一步：在Configuration标签页，查看Classpaths部分，确认flink-streaming-java-*.jar是否出现在User JARs列表里。如果出现了，说明maven-shade-plugin的excludes配置失败，必须回退修改pom.xml。

注意：Flink 的日志级别默认是INFO，很多关键的类加载信息是DEBUG级别的。如果需要更详细的信息，可以在conf/log4j2.properties里临时加上logger.flink.name = org.apache.flink和logger.flink.level = debug，然后重启集群。

6. 一个真实场景的完整复现：实时统计每分钟的热门搜索词

6.1 需求拆解：从模糊描述到精确的 DataStream 操作链

假设需求是：“实时统计最近一分钟内，用户搜索次数最多的前 10 个关键词”。这听起来简单，但落到 DataStream API 上，需要拆解成一系列精确的操作：

• Source：Kafka Topicsearch-log，每条消息是 JSON 格式{"user_id":"u123","keyword":"flink","timestamp":1717023456}。
• 时间语义：必须用Event Time，因为搜索日志的timestamp字段才是业务时间，Processing Time 会因网络延迟、消费延迟而失真。
• Watermark：需要生成 Watermark 来处理乱序事件。假设最大乱序时间为 5 秒。
• Window：TumblingEventTimeWindows.of(Time.minutes(1))，滚动一分钟窗口。
• Aggregation：先keyBy(keyword)，再window，然后aggregate（用AggregateFunction实现count）。
• TopN：窗口结束后，需要对所有keyword的 count 做排序取 Top 10。这不能在window内完成，必须用ProcessWindowFunction。

6.2 完整代码实现：EventTime、Watermark与ProcessWindowFunction的协同

public class HotSearchTopN { public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); env.enableCheckpointing(10_000); env.setStateBackend(new FsStateBackend("file:///tmp/flink-checkpoints")); // 1. 从 Kafka 读取 JSON 日志（此处简化为 Socket，原理相同） DataStream<String> jsonStream = env.socketTextStream("localhost", 9999); // 2. 解析 JSON，提取 keyword 和 event time，并分配 timestamp & watermark DataStream<SearchLog> logStream = jsonStream .map(json -> { ObjectMapper mapper = new ObjectMapper(); JsonNode node = mapper.readTree(json); return new SearchLog( node.get("keyword").asText(), node.get("timestamp").asLong() ); }) .assignTimestampsAndWatermarks( WatermarkStrategy.<SearchLog>forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((event, timestamp) -> event.timestamp) ); // 3. KeyBy + Window + Process DataStream<String> topNStream = logStream .keyBy(log -> log.keyword) // 按 keyword 分组 .window(TumblingEventTimeWindows.of(Time.minutes(1))) .aggregate(new CountAgg(), new TopNProcessFunction(10)); topNStream.print(); env.execute("Hot Search TopN Job"); } // 自定义聚合函数：累加计数 public static class CountAgg implements AggregateFunction<SearchLog, Long, Long> { @Override public Long createAccumulator() { return 0L; } @Override public Long add(SearchLog value, Long accumulator) { return accumulator + 1; } @Override public Long getResult(Long accumulator) { return accumulator; } @Override public Long merge(Long a, Long b) { return a + b; } } // 自定义 ProcessWindowFunction：收集所有 key 的 count，排序取 TopN public static class TopNProcessFunction extends ProcessWindowFunction<Long, String, String, TimeWindow> { private final int topSize; public TopNProcessFunction(int topSize) { this.topSize = topSize; } @Override public void process(String key, Context context, Iterable<Long> counts, Collector<String> out) { List<Tuple2<String, Long>> list = new ArrayList<>(); for (Long count : counts) { list.add(new Tuple2<>(key, count)); } list.sort((a, b) -> b.f1.compareTo(a.f1)); // 降序 list.stream().limit(topSize).forEach(t -> out.collect(t.f0 + ": " + t.f1)); } } // 数据 POJO public static class SearchLog { public String keyword; public long timestamp; public SearchLog(String keyword, long timestamp) { this.keyword = keyword; this.timestamp = timestamp; } } }

6.3 关键点解析：为什么ProcessWindowFunction是 TopN 的唯一解

keyBy后的window操作，天然地将数据按key分散到了不同的 subtask。CountAgg的getResult()方法，只能拿到当前key的最终 count。但 TopN 需要的是所有 key 的 count 的全局视图。ProcessWindowFunction的process()方法，其参数Iterable<Long> counts是一个包含该窗口内所有 key 的 count 值的集合（注意，不是所有 key 的 count，而是该 subtask 负责的 key 的 count）。所以，TopNProcessFunction的逻辑是：先在每个 subtask 内部做局部 TopN，然后通过rebalance()或global()等算子，把所有局部 TopN 结果汇聚到一个 subtask，再做一次全局 TopN。上面的代码为了简化，假设了keyBy后只有一个 subtask（setParallelism(1)），所以list里就是所有 key 的 count。在生产环境中，你必须增加rebalance()步骤：

DataStream<String> topNStream = logStream .keyBy(log -> log.keyword) .window(TumblingEventTimeWindows.of(Time.minutes(1))) .aggregate(new CountAgg(), new TopNProcessFunction(10)) .rebalance() // 关键：打散，让所有结果均匀分布到下游 subtask .keyBy(value -> "global") // 用一个固定 key，把所有数据路由到同一个 subtask .window(GlobalWindows.create()) .trigger(CountTrigger.of(10)) // 当收到 10 个元素时触发 .process(new GlobalTopNProcessFunction(10));

这个过程揭示了 DataStream API 的一个核心哲学：它不隐藏分布式系统的复杂性，而是提供一套原语，让你清晰地表达数据在集群中如何流动、如何分区、如何聚合。理解这一点，你就不再觉得 Flink “难”，而是觉得它“诚实”。

7. 我踩过的那些坑：关于序列化、并行度与状态清理的血泪经验

7.1 序列化之痛：为什么你的自定义 POJO 总是报NotSerializableException

Flink 的keyBy、window、state等所有操作，都要求你的数据类型（POJO）是可序列化的。Java 的Serializable接口只是第一步，更重要的是，Flink 有自己的序列化框架（TypeInformation），它需要在编译期就知道你的 POJO 结构。

// ❌ 错误示范：用了非静态内部类，持有外部类引用 public class BadExample { private String config = "default"; public class SearchLog { // 非静态内部类，隐式持有 BadExample.this public String keyword; } } // ✅ 正确示范：用 static class 或顶级类 public static class SearchLog { // static class，不持有外部引用 public String keyword; public long timestamp; // 必须有 public 无参构造函数 public SearchLog() {} // 所有字段必须是 public，或有 public getter/setter }

经验：在main()方法里直接new SearchLog()是没问题的，但一旦SearchLog出现在DataStream<SearchLog>的泛型里，Flink 就需要对其进行序列化。如果SearchLog里有一个ThreadLocal字段，或者一个lambda表达式，Flink 的序列化器会直接崩溃。最稳妥的办法，是让 POJO 只包含基本类型、String、List、Map等标准集合，并且所有字段都是public。

7.2 并行度的“双刃剑”：setParallelism(1)是开发利器，也是生产毒药

我在某公司的一个实时推荐项目里，曾把env.setParallelism(1)一直留到上线。结果上线第一天，流量高峰时，单个 TaskManager 的 CPU 打满，延迟飙升到分钟级。setParallelism(1)的好处是：所有数据都在一个线程里处理，调试时变量值、日志顺序、状态变更，全部是确定性的，没有竞态条件。坏处是：它彻底放弃了 Flink 的并行处理能力，把所有压力压在一个核上。

正确的做法是：开发阶段用1，测试阶段逐步提高到2、4，上线前根据集群资源和业务 SLA，设定一个合理的值（如8或16）。并且，parallelism的设置必须和keyBy的 key 数量匹配。如果key只有 3 个（如user_id只有 u1/u2/u3），那么parallelism=16就是浪费，因为有 13 个 subtask 是空闲的。

7.3 状态清理：为什么clearAllState()不是万能的，以及如何优雅地“重启”作业

Flink 的状态是持久化的，即使作业cancel了，CheckPoint 文件还在磁盘上。下次start时，Flink 会自动从最新的 CheckPoint 恢复，继续上次的状态。这很好，但有时你需要“清零”重来，比如修复了一个状态更新的 bug，想让所有历史状态失效。

flink cancel --savepointPath /tmp/savepoints <jobId>可以生成一个 Savepoint，然后flink run -s /tmp/savepoints/savepoint-xxx从 Savepoint 启动。但这只是“跳过”了旧状态，旧的 Checkpoint 文件还在。真正清理，需要手动删除/tmp/flink-checkpoints/和/tmp/savepoints/目录。

最后一个小技巧：在main()方法开头，加一行System.setProperty("log4j.configurationFile", "log4j2-dev.xml");，然后在resources/下放一个log4j2-dev.xml，把org.apache.flink的日志级别设为DEBUG。这样，你在本地调试时，能看到 Flink 内部的每一步调度、Task 分配、State 加载的详细日志，这是理解 Flink 运行时模型最直接的窗口。