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.18properties flink.version1.18.1/flink.version java.version11/java.version maven.compiler.source11/maven.compiler.source maven.compiler.target11/maven.compiler.target /properties dependencies dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version${flink.version}/version scopeprovided/scope /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-clients/artifactId version${flink.version}/version scopeprovided/scope /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-kafka/artifactId version3.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 到底是什么DataStreamT这个名字容易让人误解以为它是个装数据的容器。其实它更像一张“施工图纸”——你调用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 把用户代码翻译成执行图中间经过三次变形层次名称说明第一层StreamGraph直接映射用户 API 调用每个算子一个节点第二层JobGraph做算子链优化把能合并的算子合并成一个任务第三层ExecutionGraph按并行度展开每个并行子任务一个节点算子链优化是 Flink 性能的关键。默认情况下source - map - filter - sink如果并行度相同且没有 shuffle会被合并成一个任务在一个线程里执行避免线程切换和序列化开销。但keyBy会打断算子链因为数据要重新分区。你可以用env.disableOperatorChaining()全局禁用算子链或者在单个算子上调startNewChain()、disableChaining()精细控制。我一般只在调试时禁用生产环境保持默认因为链合并带来的性能提升很可观。4. Source 与 Sink 实操4.1 从集合和文件读数据最简单的 Source 是从集合读DataStreamString stream env.fromElements(a, b, c);或者从文件读DataStreamString stream env.readTextFile(path/to/file);这两个在测试时很方便但生产环境基本不用。fromElements的数据全在内存里readTextFile是一次性读取不支持增量。它们的作用是让你快速验证逻辑别拿它们做真实数据源。4.2 Kafka Source生产环境的主力真实项目里Kafka 是最常见的 Source。Flink 1.17 之后推荐用新的KafkaSource替代老的FlinkKafkaConsumerKafkaSourceString source KafkaSource.Stringbuilder() .setBootstrapServers(localhost:9092) .setTopics(user-behavior) .setGroupId(flink-group) .setStartingOffsets(OffsetsInitializer.latest()) .setValueOnlyDeserializer(new SimpleStringSchema()) .build(); DataStreamString 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 自定义 SourceRichSourceFunction有时候数据源不是 Kafka比如从数据库轮询、从 HTTP 接口拉。这时候要实现SourceFunction推荐用RichSourceFunction因为它有open和close生命周期方法public class MySource extends RichSourceFunctionString { 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(SourceContextString 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 用KafkaSinkKafkaSinkString sink KafkaSink.Stringbuilder() .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 字符串成对象如果解析失败要丢弃用flatMapstream.flatMap(new FlatMapFunctionString, User() { Override public void flatMap(String value, CollectorUser out) { try { out.collect(parse(value)); } catch (Exception e) { // 丢弃 } } });但如果解析失败要报错用map更合适stream.map(new MapFunctionString, 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 KeySelectorUser, 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 StateFlink 的状态分两种Keyed State 和 Operator State。Keyed State 只能在KeyedStream上用每个 key 一份状态Flink 自动管理 key 和状态的对应关系。Operator State 绑定在算子实例上每个并行子任务一份需要自己实现CheckpointedFunction来管理。Keyed State 有五种类型说明适用场景ValueState单值存储某个 key 的累计值ListState列表存储某个 key 的多个值MapState映射存储某个 key 的键值对集合ReducingState归约存储归约后的单值AggregatingState聚合存储聚合后的单值用ValueState举个例子统计每个用户的访问次数public class CountFunction extends RichFlatMapFunctionUser, String { private ValueStateLong countState; Override public void open(Configuration parameters) { ValueStateDescriptorLong descriptor new ValueStateDescriptor(count, Long.class); countState getRuntimeContext().getState(descriptor); } Override public void flatMap(User user, CollectorString 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 提供了状态 TTLStateTtlConfig ttlConfig StateTtlConfig .newBuilder(Time.days(7)) .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) .build(); ValueStateDescriptorLong descriptor new ValueStateDescriptor(count, Long.class); descriptor.enableTimeToLive(ttlConfig);setUpdateType决定什么时候刷新过期时间OnCreateAndWrite表示每次写都刷新OnReadAndWrite表示读也刷新。setStateVisibility决定过期但未清理的状态是否返回NeverReturnExpired表示不返回更安全。TTL 的清理是惰性的不是到期立刻删而是在读写时顺带检查。所以状态的实际大小可能比预期大一些。如果对内存敏感可以开启后台清理但会增加 CPU 开销。6. 时间语义与窗口机制6.1 三种时间事件时间、摄入时间、处理时间Flink 支持三种时间语义。事件时间是数据自带的时间戳最准确但需要处理乱序。摄入时间是数据进入 Flink 的时间介于事件时间和处理时间之间。处理时间是算子处理数据的机器时间最简单但不准确因为网络延迟、背压都会影响。生产环境我强烈建议用事件时间。原因很简单处理时间的结果不可复现同样的数据跑两次结果可能不同出了问题没法排查。事件时间虽然要处理乱序但结果是确定的。用事件时间需要设置水位线DataStreamUser stream env .addSource(new MySource()) .assignTimestampsAndWatermarks( WatermarkStrategy.UserforBoundedOutOfOrderness(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 常见问题速查表问题可能原因排查方向序列化错误引用了不可序列化对象检查算子函数字段用 transient数据倾斜key 分布不均看子任务记录数加盐窗口不触发水位线不推进看 Source 水位线加 idlenessCheckpoint 失败状态大或 Sink 不支持看 Checkpoint 大小检查 Sink背压下游处理慢看 Web UI 背压优化 SinkOOM状态无限增长加 TTL检查 key 数量7.6 我踩过的几个坑第一个坑keyBy之后用map以为状态还在结果状态丢了。keyBy产生KeyedStreammap之后变回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 上一眼就能看出的问题日志里要翻半天。