Flink DataStream API 实战：从环境搭建到状态管理与时间语义

1. 为什么从 DataStream API 入手

刚接触 Flink 那会儿，我也走过弯路。上来就啃 Table API 和 SQL，觉得写 SQL 多省事，结果遇到自定义窗口触发、状态过期策略、水位线对齐这些场景，SQL 表达起来捉襟见肘，回头还得补 DataStream 的课。后来带新人，我都建议先从 DataStream API 打起，原因很实在：它是 Flink 运行时最直接的映射，你写的每一个算子、每一条数据流，都能在 Web UI 上看到对应的任务链和背压情况，出了问题知道往哪儿查。

Flink 的 Java API 大致分三层。最底层是 ProcessFunction 那一套，能拿到时间戳、水位线、定时器，控制粒度最细；中间层是 DataStream API，也就是 map、filter、keyBy、window 这些日常写得最多的算子；最上层是 Table API 和 SQL，声明式，适合做 ETL 和报表。这三层不是互斥的，实际项目里经常混着用——主流程用 DataStream 保证灵活，维表关联和聚合用 SQL 提效。

这篇文章面向的是刚上手 Flink、准备用 Java 写第一个流处理作业的人，也适合写过一段时间但总觉得“能跑就行、原理说不清”的同行。我会把环境搭建、核心概念、算子实操、状态与时间语义这几块串起来讲，重点放在为什么这么设计和踩过的坑上，而不是照抄官方文档。代码基于 Flink 1.17 到 1.19 这几个稳定版本，1.18 之后有些 API 标了 deprecated，我会在对应位置提醒。

先说一个我自己的判断：学 Flink Java API，别一上来就追求“全”，先把一条数据从 Source 进来、经过转换、落到 Sink 的完整链路跑通，再逐步加状态、加时间语义、加容错。这条主线抓住了，剩下的都是在这个骨架上挂东西。

2. 环境搭建与项目骨架

2.1 依赖选型：别被版本号绕晕

Flink 的版本管理有个特点：flink-java和flink-streaming-java在 1.14 之后合并了，现在写流处理只需要引flink-streaming-java。我见过不少人还在 pom 里同时引这两个，结果类冲突排查半天。下面是我常用的最小依赖组合，基于 Flink 1.18：

<properties> <flink.version>1.18.1</flink.version> <java.version>11</java.version> <maven.compiler.source>11</maven.compiler.source> <maven.compiler.target>11</maven.compiler.target> </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> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka</artifactId> <version>3.1.0-1.18</version> </dependency> </dependencies>

这里有个关键点：flink-streaming-java和flink-clients用provided作用域。为什么？因为提交到集群运行时，集群自己带了这些 jar，你打进去反而可能和集群版本冲突。但本地 IDE 里跑又需要它们，provided在编译和测试阶段是可见的，正好满足。Kafka connector 则要打进去，因为集群通常不带。

注意：connector 的版本号和 Flink 版本是解耦的，格式是连接器版本-Flink版本。比如3.1.0-1.18表示 Kafka connector 3.1.0，适配 Flink 1.18。选错了会在运行时报NoSuchMethodError，这个坑我踩过不止一次。

Java 版本我选 11，这是目前生产环境最稳的选择。Flink 1.18 官方支持 Java 8、11、17，但 8 已经逐步退出，17 有些第三方 connector 还没跟上。11 是甜点区。

2.2 本地执行环境：一行代码的讲究

写第一个作业，通常从StreamExecutionEnvironment开始：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

这行代码看着简单，但它的行为取决于运行上下文。在 IDE 里直接跑，它返回一个LocalStreamEnvironment，并行度默认等于 CPU 核数；用flink run提交到集群，它返回RemoteStreamEnvironment，并行度取集群配置。我建议在本地开发时显式设置并行度：

env.setParallelism(1);

为什么？因为本地多并行度下，算子的输出顺序是不确定的，调试时打印出来的日志会交错，很难看清数据流转。设成 1 之后，所有数据串行处理，日志一目了然。等逻辑调通了，再去掉这行或者调大并行度做压测。

还有一个细节：env.execute()是触发作业执行的唯一入口。很多人写完算子忘了调它，程序直接退出，什么也不发生。这个方法的参数是作业名，会显示在 Web UI 上，起个有意义的名字，比如"user-behavior-analysis"，别用默认的"Flink Streaming Job"，不然集群上几十个作业你分不清谁是谁。

2.3 项目目录结构建议

我习惯把代码分成这么几层：

src/main/java/com/example/flink/ ├── source/ // 各种 Source 实现 ├── transform/ // 转换算子 ├── sink/ // 输出实现 ├── function/ // 自定义 Function 类 ├── model/ // POJO 数据模型 └── JobMain.java // 作业入口

这么分不是为了好看，是为了后面加状态、加 checkpoint 时，改动范围可控。比如你要给某个算子加状态，只需要动function/下的类，不用满项目找。POJO 单独放也有讲究——Flink 对 POJO 有特殊优化，能推断出字段类型，用 Kryo 泛型序列化会慢很多，后面讲序列化时会细说。

3. 核心概念：数据流、算子与执行图

3.1 DataStream 到底是什么

DataStream<T>这个名字容易让人误解，以为它是个装数据的容器。其实它更像一张“施工图纸”——你调用map、filter的时候，并没有真的处理数据，只是在图纸上画了一笔。真正干活是在env.execute()之后，Flink 把这张图纸翻译成执行图，分发到 TaskManager 上跑。

理解这一点很重要，因为它解释了很多“反直觉”的现象。比如你在map里写了个System.out.println，发现作业没提交时什么都不打印；再比如你在main方法里写了个循环，想每处理一条数据就更新一个变量，结果发现变量根本没变——因为main方法在提交作业后就结束了，真正处理数据的是另一批线程。

我常用一个类比：DataStream是菜谱，execute()是开火。菜谱写得再详细，不开火菜也不会自己熟。

3.2 算子分类：转换、分组、聚合

Flink 的算子按语义可以分几类，我按使用频率排一下：

单流转换：map、flatMap、filter。这三个是日常用得最多的。map是一进一出，flatMap是一进多出（包括零出），filter是一进零或一出。flatMap最灵活，但别滥用——能用map表达的逻辑用map，因为map的语义更清晰，Flink 也能做更多优化。

分组：keyBy。这是流处理的分水岭，一旦keyBy，数据流就从“一条”变成“多条并行的子流”，后续的算子会在每个 key 上独立执行。keyBy接受一个KeySelector，返回 key 的值。这里有个坑：key 必须是可哈希的，而且同一个 key 必须始终路由到同一个并行子任务。如果你用对象做 key 但没正确实现hashCode和equals，会出现同一个逻辑 key 被分到不同子任务的情况，聚合结果就错了。

聚合：reduce、aggregate、sum、max等。这些算子都是基于 keyed stream 的，会在每个 key 上维护累加器。reduce最通用，接受一个ReduceFunction，两两合并；aggregate更高效，因为它用AggregateFunction做增量聚合，不需要保存所有历史数据。

窗口：window、windowAll。窗口是把无界流切成有界块的手段，后面单独讲。

分流与合流：split（已废弃）、sideOutput、union、connect。sideOutput是现在做分流的标准方式，配合ProcessFunction的OutputTag使用。

3.3 执行图的三个层次

Flink 把用户代码翻译成执行图，中间经过三次变形：

算子链优化是 Flink 性能的关键。默认情况下，source -> map -> filter -> sink如果并行度相同且没有 shuffle，会被合并成一个任务，在一个线程里执行，避免线程切换和序列化开销。但keyBy会打断算子链，因为数据要重新分区。

你可以用env.disableOperatorChaining()全局禁用算子链，或者在单个算子上调startNewChain()、disableChaining()精细控制。我一般只在调试时禁用，生产环境保持默认，因为链合并带来的性能提升很可观。

4. Source 与 Sink 实操

4.1 从集合和文件读数据

最简单的 Source 是从集合读：

DataStream<String> stream = env.fromElements("a", "b", "c");

或者从文件读：

DataStream<String> stream = env.readTextFile("path/to/file");

这两个在测试时很方便，但生产环境基本不用。fromElements的数据全在内存里，readTextFile是一次性读取，不支持增量。它们的作用是让你快速验证逻辑，别拿它们做真实数据源。

4.2 Kafka Source：生产环境的主力

真实项目里，Kafka 是最常见的 Source。Flink 1.17 之后推荐用新的KafkaSource，替代老的FlinkKafkaConsumer：

KafkaSource<String> source = KafkaSource.<String>builder() .setBootstrapServers("localhost:9092") .setTopics("user-behavior") .setGroupId("flink-group") .setStartingOffsets(OffsetsInitializer.latest()) .setValueOnlyDeserializer(new SimpleStringSchema()) .build(); DataStream<String> stream = env.fromSource( source, WatermarkStrategy.noWatermarks(), "kafka-source" );

这里有几个参数值得说。setStartingOffsets决定从哪儿开始消费，latest()从最新开始，earliest()从头开始，committedOffsets()从提交的位点开始。生产环境通常用committedOffsets()，配合 checkpoint 做精确一次。setGroupId是消费者组，同一个组内的多个 Flink 并行子任务会分摊分区。

WatermarkStrategy.noWatermarks()表示不生成水位线，适合不需要事件时间语义的场景。如果需要，换成WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5))之类。

注意：Kafka Source 的并行度不要超过分区数，否则多出来的子任务会空闲。比如 topic 有 6 个分区，Source 并行度设 6 正好，设 12 就有 6 个闲着。

4.3 自定义 Source：RichSourceFunction

有时候数据源不是 Kafka，比如从数据库轮询、从 HTTP 接口拉。这时候要实现SourceFunction，推荐用RichSourceFunction，因为它有open和close生命周期方法：

public class MySource extends RichSourceFunction<String> { private volatile boolean running = true; private transient Connection conn; @Override public void open(Configuration parameters) throws Exception { conn = DriverManager.getConnection("jdbc:..."); } @Override public void run(SourceContext<String> ctx) throws Exception { while (running) { // 查询数据 String data = queryData(conn); ctx.collect(data); Thread.sleep(1000); } } @Override public void cancel() { running = false; } @Override public void close() throws Exception { if (conn != null) conn.close(); } }

关键点：running必须是volatile，因为cancel()由另一个线程调用，要保证可见性。conn用transient修饰，因为它不可序列化，Flink 在分发算子时会序列化对象，不标记会报错。open里初始化连接，close里释放，这是资源管理的标准姿势。

4.4 Sink：数据往哪儿写

Sink 的选择取决于下游。写 Kafka 用KafkaSink：

KafkaSink<String> sink = KafkaSink.<String>builder() .setBootstrapServers("localhost:9092") .setRecordSerializer(KafkaRecordSerializationSchema.builder() .setTopic("output-topic") .setValueSerializationSchema(new SimpleStringSchema()) .build()) .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE) .build(); stream.sinkTo(sink);

setDeliveryGuarantee有三个选项：NONE、AT_LEAST_ONCE、EXACTLY_ONCE。精确一次需要 Kafka 事务支持，会带来额外开销，按业务需求选。写数据库的话，通常自定义RichSinkFunction，在invoke里执行 insert，配合批量提交提升吞吐。

我个人的经验是：Sink 这一端最容易成为瓶颈。因为下游系统（数据库、消息队列）的写入速度往往跟不上 Flink 的处理速度，背压会一路传回 Source。所以 Sink 的批量大小、并发连接数要仔细调，别让整个作业卡在最后一步。

5. 转换算子深入与状态管理

5.1 map、flatMap、filter 的取舍

这三个算子看着简单，但用错了会影响性能。map是一进一出，Flink 知道输出数量等于输入数量，可以做更激进的优化。flatMap是一进多出，Flink 无法预知输出数量，每次collect都要走一遍序列化。所以能用map就别用flatMap。

举个例子，解析 JSON 字符串成对象，如果解析失败要丢弃，用flatMap：

stream.flatMap(new FlatMapFunction<String, User>() { @Override public void flatMap(String value, Collector<User> out) { try { out.collect(parse(value)); } catch (Exception e) { // 丢弃 } } });

但如果解析失败要报错，用map更合适：

stream.map(new MapFunction<String, User>() { @Override public User map(String value) throws Exception { return parse(value); // 抛异常由 Flink 处理 } });

filter的注意点是：过滤掉的数据越早越好。如果你在map里做了复杂计算，然后filter掉大部分数据，那是浪费。把filter往前放，先筛掉不需要的，再计算。

5.2 keyBy 的底层逻辑

keyBy是流处理里最重要的算子，没有之一。它的作用是把一条流按 key 分成多条逻辑子流，后续算子在每个 key 上独立执行。底层实现是：对 key 做哈希，然后对并行度取模，决定这条数据去哪个子任务。

stream.keyBy(new KeySelector<User, String>() { @Override public String getKey(User user) { return user.getUserId(); } });

这里有个性能陷阱：如果 key 分布不均匀，会出现数据倾斜。比如按用户 ID 分组，某个大 V 的用户 ID 占了 80% 的数据，那处理这个 key 的子任务就会成为瓶颈，其他子任务闲着。解决办法是加盐——给 key 拼一个随机后缀，先做局部聚合，再去掉后缀做全局聚合。这个技巧在实时大屏场景很常用。

还有一个坑：keyBy之后的数据流是KeyedStream，不是DataStream。很多算子只能用在KeyedStream上，比如reduce、sum、window。如果你keyBy之后想再map，是可以的，但map之后又变回DataStream，状态就丢了。所以keyBy的位置要规划好。

5.3 状态类型：Keyed State 与 Operator State

Flink 的状态分两种：Keyed State 和 Operator State。Keyed State 只能在KeyedStream上用，每个 key 一份状态，Flink 自动管理 key 和状态的对应关系。Operator State 绑定在算子实例上，每个并行子任务一份，需要自己实现CheckpointedFunction来管理。

Keyed State 有五种：

用ValueState举个例子，统计每个用户的访问次数：

public class CountFunction extends RichFlatMapFunction<User, 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(User user, Collector<String> out) throws Exception { Long count = countState.value(); if (count == null) count = 0L; count++; countState.update(count); out.collect(user.getUserId() + " : " + count); } }

open里注册状态描述符，flatMap里读写。注意countState.value()可能返回 null，因为第一次访问时状态还没初始化。这个 null 判断是必须的，忘了会 NPE。

5.4 状态过期：别让状态无限膨胀

状态是存在内存和磁盘上的，如果 key 无限增长，状态会撑爆。Flink 提供了状态 TTL：

StateTtlConfig ttlConfig = StateTtlConfig .newBuilder(Time.days(7)) .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) .build(); ValueStateDescriptor<Long> descriptor = new ValueStateDescriptor<>("count", Long.class); descriptor.enableTimeToLive(ttlConfig);

setUpdateType决定什么时候刷新过期时间，OnCreateAndWrite表示每次写都刷新，OnReadAndWrite表示读也刷新。setStateVisibility决定过期但未清理的状态是否返回，NeverReturnExpired表示不返回，更安全。

TTL 的清理是惰性的，不是到期立刻删，而是在读写时顺带检查。所以状态的实际大小可能比预期大一些。如果对内存敏感，可以开启后台清理，但会增加 CPU 开销。

6. 时间语义与窗口机制

6.1 三种时间：事件时间、摄入时间、处理时间

Flink 支持三种时间语义。事件时间是数据自带的时间戳，最准确，但需要处理乱序。摄入时间是数据进入 Flink 的时间，介于事件时间和处理时间之间。处理时间是算子处理数据的机器时间，最简单但不准确，因为网络延迟、背压都会影响。

生产环境我强烈建议用事件时间。原因很简单：处理时间的结果不可复现，同样的数据跑两次结果可能不同，出了问题没法排查。事件时间虽然要处理乱序，但结果是确定的。

用事件时间需要设置水位线：

DataStream<User> stream = env .addSource(new MySource()) .assignTimestampsAndWatermarks( WatermarkStrategy.<User>forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((user, ts) -> user.getEventTime()) );

forBoundedOutOfOrderness表示允许 5 秒的乱序，水位线 = 最大事件时间 - 5 秒。withTimestampAssigner从数据里提取时间戳。这两个必须配套，缺一不可。

6.2 水位线：事件时间的“进度条”

水位线是 Flink 用来判断“某个时间点之前的数据都到齐了”的机制。它是一个特殊的数据记录，插入在数据流里，随着数据一起流动。当算子收到水位线，就知道不会再有比这个时间更早的数据了，可以触发窗口计算。

水位线的传播有个特点：多输入算子取所有输入的最小值。比如一个算子有两个输入流，一个水位线到了 10 秒，另一个到了 8 秒，算子取 8 秒。这保证了不会因为某个流慢而提前触发计算。

注意：如果某个流长时间没有数据，水位线不会推进，窗口永远不会触发。这种情况要么设置withIdleness让空闲流被忽略，要么在 Source 端定期发送心跳数据。

6.3 窗口类型与选择

窗口是把无界流切成有界块的手段。Flink 的窗口分几类：

滚动窗口：固定大小，不重叠。比如每 5 分钟统计一次。

stream.keyBy(...) .window(TumblingEventTimeWindows.of(Time.minutes(5))) .sum("count");

滑动窗口：固定大小，可重叠。比如每 1 分钟统计过去 5 分钟的数据。

stream.keyBy(...) .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1))) .sum("count");

会话窗口：按活动间隙切分。比如用户 30 秒没操作就认为会话结束。

stream.keyBy(...) .window(EventTimeSessionWindows.withGap(Time.seconds(30))) .sum("count");

选哪种窗口取决于业务。滚动窗口适合固定周期的报表，滑动窗口适合平滑的监控指标，会话窗口适合用户行为分析。滑动窗口的计算量比滚动窗口大，因为数据会被多个窗口重复计算，滑动步长越小开销越大。

6.4 窗口函数：增量聚合与全量聚合

窗口函数分两类。增量聚合每来一条数据就计算一次，只保存中间结果，内存占用小。全量聚合把窗口内所有数据缓存起来，等窗口触发时一次性计算，内存占用大但能做更复杂的逻辑。

ReduceFunction和AggregateFunction是增量聚合，ProcessWindowFunction是全量聚合。两者可以结合：

stream.keyBy(...) .window(TumblingEventTimeWindows.of(Time.minutes(5))) .aggregate(new MyAggregateFunction(), new MyProcessWindowFunction());

这样AggregateFunction做增量聚合，ProcessWindowFunction拿到聚合结果后再加工，兼顾效率和灵活性。这是我最常用的组合。

7. 常见问题与排查技巧

7.1 作业启动就报序列化错误

这是新手最常见的错误，报错信息通常是The implementation of the ... is not serializable。原因是你在算子函数里引用了不可序列化的对象，比如数据库连接、SimpleDateFormat。

解决办法：把不可序列化的字段用transient修饰，在open方法里初始化。SimpleDateFormat是线程不安全的，每个并行子任务要独立一份，正好放在open里。

7.2 数据倾斜导致个别子任务卡死

现象是 Web UI 上某个子任务的背压是红色，其他是绿色。原因是 key 分布不均。排查方法是看每个子任务处理的记录数，差异大的就是倾斜。

解决办法：加盐。给 key 拼一个随机前缀，先做局部聚合，再去掉前缀做全局聚合。代价是要多一轮 shuffle，但能解决倾斜。

7.3 水位线不推进，窗口不触发

现象是窗口一直不输出结果。原因通常是某个 Source 没有数据，水位线卡住了。排查方法是看 Web UI 上每个 Source 的水位线值。

解决办法：给WatermarkStrategy加withIdleness(Duration.ofMinutes(1))，让空闲的 Source 被忽略。或者在 Source 端定期发送心跳数据。

7.4 Checkpoint 失败或超时

现象是作业频繁重启，日志里有Checkpoint expired或Checkpoint failed。原因可能是状态太大、下游 Sink 不支持事务、网络抖动。

排查思路：先看 Checkpoint 的大小和耗时，如果状态太大，考虑加 TTL 或换状态后端。如果 Sink 不支持事务，检查DeliveryGuarantee设置。如果是网络问题，调大超时时间。

7.5 常见问题速查表

7.6 我踩过的几个坑

第一个坑：keyBy之后用map，以为状态还在，结果状态丢了。keyBy产生KeyedStream，map之后变回DataStream，状态上下文没了。要保留状态，得用process或者flatMap这类能访问RuntimeContext的算子。

第二个坑：ValueState忘了判空。第一次访问时value()返回 null，直接+1会 NPE。这个错误在测试时不容易发现，因为测试数据往往从 0 开始，生产环境第一次遇到新 key 就崩了。

第三个坑：并行度和 Kafka 分区数不匹配。Source 并行度大于分区数，多出来的子任务空闲，浪费资源；小于分区数，一个子任务消费多个分区，可能成为瓶颈。最好设成相等。

第四个坑：env.execute()忘了调。代码写得再漂亮，不调这个方法作业不会跑。而且execute()是阻塞的，调了之后main方法不会退出，直到作业结束。

8. 从单机到集群的部署要点

本地调通之后，下一步是提交到集群。打包时注意provided的依赖不要打进去，用maven-shade-plugin打一个 fat jar，只包含 connector 和业务代码。提交命令是flink run -c 主类名 your-job.jar。

集群上跑有几个和本地不一样的地方。并行度取集群配置，不再是 CPU 核数。状态后端默认是 HashMapStateBackend，状态存在 TaskManager 内存里，大状态要换成 RocksDBStateBackend，状态落磁盘。Checkpoint 要配置持久化存储，否则作业重启后状态恢复不了。

我个人的经验是：本地开发和集群运行的差异，主要集中在并行度、状态后端、Checkpoint 配置这三块。本地用默认值就行，集群上要根据数据量和可用资源仔细调。调参没有银弹，都是压测出来的——先设一个保守值，观察背压和 Checkpoint 耗时，再逐步调整。

最后分享一个小技巧：Web UI 是你最好的朋友。背压、水位线、Checkpoint、算子链，所有运行时信息都在上面。遇到问题先看 UI，比翻日志快得多。我刚开始学的时候总想着看日志，后来发现 UI 上一眼就能看出的问题，日志里要翻半天。