1. 为什么我要从Java API重新理解Flink
刚接触Flink那会儿，我犯过一个很典型的错误：抱着官方文档啃了三天，DataSet、DataStream、Table API混着看，结果写出来的第一个作业连本地都跑不起来。后来复盘才发现，问题不在于Flink难，而在于我没有找到一个稳定的切入点。Flink的API体系其实分了好几层，Java API是最底层、最接近运行时的那一层，把这一层吃透了，上面那些SQL、Table的封装你才能看懂它到底在干什么。
这篇内容就是围绕Flink Java API展开的。我会从环境搭建讲到DataStream的核心算子，再到窗口、状态、时间语义这些真正决定作业能不能跑对的关键点。适合两类人看：一类是刚学Flink、被各种API绕晕的新手；另一类是用过Flink SQL但想往下钻、搞清楚底层执行逻辑的开发者。我不会只贴代码，每个参数、每个选择背后的原因我都会讲清楚，因为Flink这东西，光会抄模板是跑不出正确结果的。
先说清楚一个前提：Flink的Java API主要分两块历史遗产——老的DataSet API（批处理）和DataStream API（流处理）。从1.12开始官方就在推动流批一体，DataSet已经进入维护状态，新项目一律建议用DataStream API配合有界流来处理批数据。所以这篇内容的重心放在DataStream上，DataSet只在必要的地方提一句，避免你学了马上要过时的东西。
2. 环境搭建与第一个可运行的作业
2.1 依赖选型：别小看pom里那几行
很多人第一步就卡在依赖上。Flink的依赖有个坑：flink-java、flink-streaming-java、flink-clients这几个包的作用范围（scope）如果配错，本地IDE能跑，一打包提交就报ClassNotFound。我的习惯是这样配：
<properties> <flink.version>1.17.1</flink.version> <scala.binary.version>2.12</scala.binary.version> </properties> <dependencies> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> </dependencies>为什么用provided？因为集群上本身就有这些jar，你打包时再塞进去，轻则包体积爆炸，重则和集群版本冲突导致各种诡异的序列化异常。本地跑的时候IDE会把provided的依赖加进classpath，所以不影响调试。这个细节我踩过坑：有一次打出来的fat jar有200多MB，就是因为把Flink核心包全打进去了。
版本选择上，1.17是个比较稳的长期维护版本，1.18、1.19也陆续在推，但生产环境我一般不会追最新，等它出两三个小版本修完bug再上。Scala版本那个属性看着没用，其实有些传递依赖会用到，留着能避免一些莫名其妙的解析失败。
2.2 执行环境：getExecutionEnvironment到底做了什么
第一个作业永远是从这行开始的：
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();这行代码看着简单，但它内部做了个判断：如果你是在IDE里直接跑main方法，它返回一个LocalStreamEnvironment，作业在本地JVM里多线程模拟执行；如果你是把jar提交到集群，它返回的是RemoteStreamEnvironment，作业会被分发到集群的TaskManager上。这个自动判断机制很贴心，但也埋了个雷——本地和集群的行为不完全一致，尤其是并行度和资源相关的地方。
我建议在开发阶段显式设置并行度，别依赖默认值：
env.setParallelism(2);默认并行度是你机器的CPU核数，本地跑可能开8个线程，提交到集群如果没配好又变成别的值，结果就是同样的代码两边输出不一样。显式设置能让行为可预测，排查问题时少一个变量。
2.3 从Source到Sink：一个完整的词频统计
光说不练没意义，上个能跑的完整例子。这个例子从socket读数据，做词频统计，结果打到控制台：
public class SocketWordCount { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); DataStream<String> lines = env.socketTextStream("localhost", 9999); DataStream<Tuple2<String, Integer>> counts = lines .flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() { @Override public void flatMap(String line, Collector<Tuple2<String, Integer>> out) { for (String word : line.split("\\s+")) { if (!word.isEmpty()) { out.collect(Tuple2.of(word, 1)); } } } }) .keyBy(value -> value.f0) .sum(1); counts.print(); env.execute("Socket WordCount"); } }跑之前先在终端开个端口：nc -lk 9999，然后启动程序，随便敲几行字，控制台就会滚动输出每个单词的累计计数。
这里有几个点值得展开。keyBy是Flink里最核心的操作之一，它按key把数据分区，保证同一个key的数据一定流向同一个并行子任务，这是后面做状态和窗口的前提。sum(1)里的1是Tuple的字段下标，表示对第二个字段（计数）做累加。env.execute()是必须的，Flink的算子都是惰性的，不调用execute什么都不会发生——这点和Spark的惰性求值是一个道理。
注意：
socketTextStream只适合本地测试，它不能保证exactly-once，也不支持并行读取。生产环境的Source要用Kafka、文件系统这类带checkpoint能力的连接器。
3. DataStream核心算子逐个拆解
3.1 map、flatMap、filter：最基础的三件套
这三个算子几乎每个作业都会用到，但它们的区别和适用场景很多人说不清楚。
map是一对一转换，进来一条出去一条。比如把字符串转成对象、把温度值做单位换算。它的函数签名是MapFunction<T, R>，你只需要实现R map(T value)。
flatMap是一对多，进来一条可以出去零条、一条或多条。词频统计里用它就是因为一行文本要拆成多个单词。它的签名是FlatMapFunction<T, R>，实现void flatMap(T value, Collector<R> out)，通过out.collect()决定输出几条。
filter是过滤，返回boolean决定这条数据留不留。签名FilterFunction<T>，实现boolean filter(T value)。
选哪个的判断标准很简单：输出条数和输入条数相等用map，不等用flatMap，只做筛选用filter。有人喜欢用flatMap干所有事，能跑但可读性差，而且flatMap的序列化开销比map略大，没必要的地方别用。
3.2 keyBy与分区：数据到底去了哪个并行子任务
keyBy不是算子，它是分区转换，返回的是KeyedStream。它的作用是把流按key的哈希值重新分布到下游的各个并行子任务上。这里有个关键公式：
分区索引 = hash(key) % 下游并行度理解这个公式能解释很多现象。比如你设了并行度4，key是用户ID，那么同一个用户的数据永远落到同一个子任务，这就是状态能正确维护的基础。但如果你中途改了并行度，比如从4改成8，那么原来落在子任务0的数据，改完之后可能落到子任务0或4，状态就需要reshuffle，处理不好会丢数据或重复。
除了keyBy，Flink还有几个手动分区算子：
| 算子 | 作用 | 使用场景 |
|---|---|---|
| shuffle | 完全随机分区 | 负载均衡，但同一key不保证同分区 |
| rebalance | 轮询分区 | 数据倾斜时打散负载 |
| rescale | 局部轮询 | 上下游并行度成比例时效率更高 |
| forward | 一对一 | 上下游并行度相同时避免网络传输 |
| broadcast | 广播到所有分区 | 小表关联大表、配置分发 |
我实测下来，rebalance是解决数据倾斜最省事的办法，但它会破坏key的有序性，所以只能用在不需要按key聚合的场景。rescale比rebalance更轻量，因为它只在局部做轮询，不走全局协调，但要求上下游并行度有整数倍关系。
3.3 聚合算子：sum、max、min的隐藏陷阱
KeyedStream上可以直接调sum、max、min、maxBy、minBy这些聚合算子。看起来简单，但max和maxBy的区别坑过不少人。
假设数据是(商品ID, 价格, 库存)这样的三元组，你要找每个商品价格最高的那条记录：
max(1)：只把价格字段更新成最大值，其他字段（库存）保持第一条记录的值，输出的是个"缝合怪"。maxBy(1)：返回价格最大的整条记录，库存字段也是那条记录的真实值。
大部分业务场景你要的是maxBy，因为你要的是完整的那条数据。max只在"我只要这个最大值本身，其他字段无所谓"时才用。这个区别官方文档写得很含蓄，我第一次用的时候输出结果对不上，排查了半天才发现是算子选错了。
另外这些聚合算子内部都是有状态的，状态默认存在TaskManager的内存里。如果key的数量很大，比如按用户ID聚合几千万用户，内存会撑爆。这时候要么用keyBy之后接窗口限制状态范围，要么配置状态后端把状态放到RocksDB。
3.4 富函数：需要open和close时怎么办
普通的MapFunction、FlatMapFunction是无状态的工具类，但有时候你需要在算子启动时初始化一个数据库连接，结束时关闭它。这时候就要用RichFunction：
public class RichFlatMapExample extends RichFlatMapFunction<String, String> { private transient Connection conn; @Override public void open(Configuration parameters) throws Exception { conn = DriverManager.getConnection("jdbc:..."); } @Override public void flatMap(String value, Collector<String> out) throws Exception { // 使用conn } @Override public void close() throws Exception { if (conn != null) conn.close(); } }open在每个并行子任务启动时调用一次，close在任务结束时调用。这里有个大坑：连接对象不能是静态的，也不能在多个子任务间共享，因为每个子任务可能在不同的线程甚至不同的机器上。用transient修饰是因为这些字段不需要被序列化——它们是在运行时才创建的。
还有个更隐蔽的坑：open里创建的连接如果长时间不用会超时断开，但open只调用一次。所以生产环境要么加连接池和心跳检测，要么在flatMap里做重连判断。我见过一个作业跑了三天突然全挂，就是因为数据库连接被服务端掐了而代码没处理。
4. 窗口、时间与状态：让流处理真正可用
4.1 时间语义：Event Time和Processing Time怎么选
Flink有三种时间概念：
- Event Time：数据本身携带的时间戳，比如日志里的
timestamp字段。 - Ingestion Time：数据进入Flink Source的时间。
- Processing Time：数据被算子处理时的机器时间。
选哪个直接决定结果的正确性。Processing Time最简单，不用管乱序，但结果不可复现——同样的数据跑两次结果可能不一样。Event Time能处理乱序和延迟数据，结果稳定，但需要配合Watermark机制。
我的经验是：只要数据里带时间戳，一律用Event Time。只有那种实时性要求极高、对准确性要求不高的监控场景，才用Processing Time图省事。设置方式：
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);注意1.12之后这个API被标记过时了，新写法是在WatermarkStrategy里指定，但底层逻辑没变。
4.2 Watermark：怎么处理迟到的数据
Watermark是Flink处理乱序的核心机制。你可以把它理解成一个"进度声明"：当Watermark到达时间T时，系统认为T之前的数据都到齐了，可以触发窗口计算了。
生成Watermark最常用的是BoundedOutOfOrdernessTimestampExtractor，它允许数据迟到一段时间：
DataStream<Event> stream = env.addSource(...) .assignTimestampsAndWatermarks( WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((event, ts) -> event.getTimestamp()) );这里的5秒就是最大允许延迟。假设窗口是10秒，那么窗口结束时间到了之后，还会再等5秒才真正触发计算，给迟到的数据一个机会。这个值设多大有讲究：设太小，迟到数据被丢弃；设太大，结果输出延迟高。我一般先看数据的实际乱序程度，用监控统计P99延迟，然后取那个值再留点余量。
超过允许延迟的数据怎么办？默认是丢弃，但你可以用sideOutputLateData把它们收集到侧输出流，单独处理。这个功能很实用，我经常用它来监控有多少数据因为迟到被丢了，如果量太大就说明Watermark设小了。
4.3 窗口类型：滚动、滑动、会话的区别
窗口是把无界流切成有界块的手段，Flink提供了几种：
滚动窗口（Tumbling Window）：固定大小，不重叠。比如每5分钟统计一次，窗口是[0,5)、[5,10)、[10,15)。适合周期性报表。
滑动窗口（Sliding Window）：固定大小，可重叠。比如窗口大小10分钟、滑动步长5分钟，那么窗口是[0,10)、[5,15)、[10,20)。适合需要平滑过渡的指标，比如"最近10分钟的订单量"每5分钟更新一次。
会话窗口（Session Window）：按活动间隙切分。如果两条数据间隔超过设定值，就切一个新窗口。适合用户行为分析，比如"一次会话内的点击流"。
代码上：
// 滚动窗口，每5秒一个 stream.keyBy(...).window(TumblingEventTimeWindows.of(Time.seconds(5))); // 滑动窗口，大小10秒，步长5秒 stream.keyBy(...).window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5))); // 会话窗口，间隙5秒 stream.keyBy(...).window(EventTimeSessionWindows.withGap(Time.seconds(5)));窗口的触发时机由Watermark决定。滚动窗口在Watermark越过窗口结束时间时触发；滑动窗口因为重叠，一个数据可能属于多个窗口，每个窗口独立触发。这里要注意：窗口的并行度由keyBy决定，同一个key的窗口计算在同一个子任务里，所以窗口状态也是按key隔离的。
4.4 状态管理：ValueState、ListState和MapState
Flink的状态是它区别于普通流处理框架的核心能力。常用的状态类型：
- ValueState：单值状态，存一个对象。
- ListState：列表状态，存多个对象。
- MapState<K, V>：映射状态，存键值对。
- ReducingState / AggregatingState：带聚合逻辑的状态。
用状态必须通过RichFunction的RuntimeContext获取：
public class StatefulFlatMap extends RichFlatMapFunction<Event, String> { private ValueState<Long> countState; @Override public void open(Configuration parameters) { ValueStateDescriptor<Long> descriptor = new ValueStateDescriptor<>("count", Long.class); countState = getRuntimeContext().getState(descriptor); } @Override public void flatMap(Event event, Collector<String> out) throws Exception { Long current = countState.value(); if (current == null) current = 0L; current += 1; countState.update(current); out.collect("count=" + current); } }状态的关键在于它和key绑定。上面这段代码，每个key都有自己独立的countState，互不干扰。这就是为什么状态只能在KeyedStream上使用——非keyed的流没有key的概念，状态无法隔离。
状态后端的选择也很重要。默认的HashMapStateBackend把状态放内存，快但容量有限；EmbeddedRocksDBStateBackend把状态放磁盘，慢一些但能支撑TB级状态。判断标准是：状态总量能不能塞进内存。我一般超过几GB就切RocksDB，别等OOM了才想起来换。
5. 常见问题与排查实录
5.1 作业跑不起来？先看这几个地方
新手最常遇到的报错和排查方向，我整理成了一张表：
| 报错现象 | 大概率原因 | 排查方向 |
|---|---|---|
| ClassNotFoundException | 依赖scope配错 | 检查是否误用compile打包Flink核心包 |
| 序列化异常 | 自定义类没实现Serializable | POJO字段加transient或实现接口 |
| 作业提交后立即失败 | 并行度超过slot数 | 检查TaskManager的slot配置 |
| 结果一直不输出 | Watermark没推进 | 检查数据时间戳是否单调、是否为空 |
| 内存溢出 | 状态太大或数据倾斜 | 换RocksDB后端、加rebalance |
| checkpoint超时 | 状态太大或反压 | 看背压监控、调大超时时间 |
序列化那个坑特别常见。Flink要求所有在算子间传递的数据、所有状态里的对象都必须可序列化。Java的Serializable接口是最省事的，但Flink对POJO有更高效的序列化器，条件是：类是public的、有无参构造函数、字段是public或有getter/setter。满足这些条件Flink会自动用POJO序列化器，性能比Java序列化好很多。
5.2 数据倾斜：最难缠的性能问题
数据倾斜的表现是：大部分子任务很快跑完，个别子任务卡住不动，整体吞吐上不去。根因是某些key的数据量远大于其他key。
排查方法：在Flink Web UI上看各子任务的Records Sent/Received，如果差异超过一个数量级，基本就是倾斜了。
解决办法分两层。第一层是加盐：给热点key拼个随机前缀，把它打散到多个子任务，聚合完再去掉前缀做二次聚合。这个办法有效但代码复杂。第二层是两阶段聚合：先局部聚合再全局聚合，类似MapReduce的combiner思路。
我实际用下来，如果倾斜不严重（比如最大子任务是平均值的3倍以内），直接上rebalance最省事。如果倾斜严重到某个key占了80%的数据，那就得从业务上想办法，比如把大key拆成多个小key，或者单独处理。
5.3 反压：从下游往上游找原因
反压（Backpressure）是Flink的自我保护机制：下游处理不过来，上游就降速。Web UI上如果某个算子显示High背压，说明它是瓶颈。
排查顺序是从下游往上游走：先看Sink是不是写得太慢（比如数据库批量插入没开、每条都提交），再看中间算子是不是有耗时操作（同步IO、复杂计算），最后看Source是不是读太快。
我遇到过一次反压，查了半天发现是Sink端每条记录都单独提交事务，改成批量提交后吞吐直接翻了十倍。所以看到反压别急着调并行度，先定位瓶颈在哪，很多时候是代码写法问题而不是资源不够。
5.4 checkpoint失败：状态和存储的博弈
checkpoint是Flink容错的基础，但它失败的原因五花八门。最常见的几个：
- 超时：状态太大，上传到存储的时间超过配置的超时时间。解决方法是调大
execution.checkpointing.timeout，或者换更快的存储。 - 存储不可用：checkpoint的存储路径（比如HDFS、对象存储）连不上。检查网络和权限。
- 并发checkpoint冲突：上一个还没完成下一个就开始了。调大
execution.checkpointing.interval，或者开启execution.checkpointing.unaligned减少对齐时间。 - 状态太大：增量checkpoint能缓解，但前提是用了RocksDB后端。
我的经验是，checkpoint间隔别设太短，1到5分钟比较合理。设成10秒一次，大部分时间都花在checkpoint上，正常处理反而被拖慢。另外一定要配重启策略，不然checkpoint失败后作业直接挂了不会自动恢复：
env.setRestartStrategy(RestartStrategies.fixedDelayRestart( 3, Time.seconds(10) ));这段配置表示失败后重试3次，每次间隔10秒。生产环境一般用failureRateRestart，按失败率来决定是否重启，更灵活。
6. 一些让我少走弯路的实操心得
写Flink作业这几年，有几个习惯是踩坑踩出来的，分享给你。
第一，永远显式设置并行度和时间语义。别依赖默认值，默认值在本地和集群、不同版本之间都可能变。显式写出来，行为可预测，排查问题时少一个变量。
第二，本地测试用有界数据。用fromElements或fromCollection造一批测试数据，跑完自动结束，比socket流方便调试。等逻辑对了再换成真实Source。
第三，状态能不用就不用。状态是Flink强大的地方，也是复杂度和故障的来源。能用窗口解决的别用自定义状态，能用聚合算子的别自己维护ValueState。每多一个状态，就多一份checkpoint开销和恢复时间。
第四，监控比调试重要。作业上线后，checkpoint成功率、反压、吞吐量这三个指标要盯着。很多问题在爆发前都有征兆，比如checkpoint时间逐渐变长，往往意味着状态在膨胀或者存储变慢。
第五，版本升级要谨慎。Flink的API在小版本间相对稳定，但状态后端的兼容性、序列化器的行为偶尔会变。升级前先在测试环境跑一遍，确认checkpoint能正常恢复再上生产。我见过升级后老checkpoint读不出来的情况，只能从零开始重跑，损失很大。
最后说个容易被忽略的点：算子的链化（Operator Chaining）。Flink默认会把能合并的算子链在一起，减少线程切换和序列化开销。但有时候链化反而有害，比如某个算子特别耗CPU，链在一起会拖累整条链。这时候可以用disableChaining()或startNewChain()手动控制。这个优化属于进阶技巧，作业跑得动的时候不用管，遇到性能瓶颈时值得一试。
