凌晨一点半，我被一条告警电话叫醒。实时大屏上的成交金额和订单数量，跟凌晨跑批出来的离线报表差了将近两个数量级，数据组同事在群里反复贴截图：“到底以哪个为准？”这种事在数据岗待过的人应该都懂：业务方既要秒级更新的实时指标，又要凌晨全量的对账结果，而团队手里往往不止一套处理系统。当时我负责的数据任务就是典型的Lambda架构：离线用MR/Spark跑批，实时用另一套流计算引擎消费Kafka，两套代码、两套调度、两套口径，每次对不上账都要猜半天。
后来我把批量处理和实时流处理的数据任务统一迁到了Flink上。Flink的核心思路是“批流一体”：一个DataStream模型同时描述有界数据流（批量数据）和无界数据流（实时流数据），同一套代码可以按两种执行模式运行，只是边界语义不同。这篇文章就是基于这个“8.1.1 Flink快速应用”章节整理出来的实操笔记，适合刚开始接触Flink、或者正在被批流两套系统折磨的朋友。跟着下文把批量处理和实时流处理的例子跑通，你就能理解为什么Flink能成为实时计算的常用底座。
1. 批与流的两难：为什么我最终把离线任务和实时任务搬到了同一套引擎上
先说一个很容易被新手忽略的点：Flink最厉害的地方不是“流处理性能比别的框架快多少”，而是它把批和流两套编程模型统一了。用一句话概括就是：批量数据本质上是一个“有界的数据流”，数据总会有终点的；实时流数据是“无界的数据流”，永远等不到终点。这两者在底层处理逻辑上有大量共通之处，完全可以抽象成同一个模型去操作。
1.1 从两张报表对不上说起
我之前维护过一个电商场景的数据任务。实时侧要统计当前小时的下单金额，离线侧每天凌晨跑全量订单、算前一天的最终金额。两个数字经常对不上，原因倒不复杂：实时侧Kafka里存在延迟数据、迟到数据，窗口一关就丢掉了；离线侧有完整数据，却因为中途某个上游任务重跑，把事实表数据刷了一遍，结果两边口径不一样。为了对齐，我不得不把实时任务的逻辑改成“当天先算、第二天再覆盖”，说白了就是批流两套代码都保留，等于双倍维护。
这种场景太普遍了。业务方不会在意你底层用了几套引擎，他们要的只是一个口径清晰的数字。可工程师在Lambda架构下往往要写好几份逻辑：一份批处理、一份流处理、还有一份“用于对账的批处理”。代码里到处充斥着差不多的Map、Join、Aggregate，却因为API不同，谁也没法复用。
1.2 Lambda架构为什么让人疲惫
Lambda架构的核心思想是：实时层负责低延迟，批处理层负责高准确度，最终在服务层做合并。听上去很完美，实际维护起来非常痛苦。
- 两套引擎的部署运维、监控告警、资源管理是两套体系。
- 同一个指标往往在批和流里各有定义，语义漂移在所难免。
- 数据重跑时，实时结果已经对不上了，还得靠批处理去校准。
- 新人入职光熟悉两套API就需要不少时间。
尤其当业务一旦从“小时级别”进化到“秒级别”，比如实时大屏、风控识别、推荐特征更新，批处理根本跟不上，而旧的流处理框架要么是微批模式，要么缺少端到端一致性保证，做精确计算特别费劲。
1.3 Flink的解题思路：把“批”当成“流”来跑
Flink在底层把一切数据都抽象成流：有结尾的流就是批，没有结尾的流就是流。DataStream API可以同时处理两者。你在代码里写一个DataStream<X>，把运行模式设为BATCH就是批处理，设为STREAMING就是流处理，代码骨架几乎不用变。窗口、状态、水位线这些流处理概念，在批处理模式下也会有对应的简化语义。
换句话说，Flink不是简单地把Spark Streaming那种“微批”包装成“流”，而是真的在做事件级流处理，同时反过来在批模式下借用流式执行引擎的分布式能力和容错机制。这种设计带来的直接好处是：同一套代码，既能做每天凌晨的批计算，又能支撑实时大屏。维护成本从两套降为一套，数据口径也能收敛到同一种表达上。
2. 十分钟搭好本地环境：Flink快速上手的依赖与配置
接触Flink的第一步其实不是部署集群，而是先把本地开发环境跑通。一个最简单的Flink作业在IDE里就能直接跑，不需要本地先起一个Flink集群。这一点和写普通Java应用一样轻松，反而是版本选错会浪费不少时间。
2.1 版本选择比想象中更重要
我用过的Flink版本里，1.12是一个分水岭，它正式把DataSet API标记为冻结状态，全面推荐DataStream API做批处理。所以你现在看网上很多老教程还在用ExecutionEnvironment.fromElements、DataSet<T>，这些代码在1.12以上的版本里编译可能能过，但已经不建议用了；到了1.18之后，旧的DataSet API入口直接被移除。
我个人推荐直接从Flink 1.14或更高版本起步，比如我用的是1.17.2。原因很简单：新版内置了RuntimeExecutionMode，支持STREAMING和BATCH两种执行模式，同时Kafka连接器拆成了独立JAR管理，版本对应关系更清晰。跟实时计算相关的组件生态也都在往新版本迁移，网上踩坑资料相对好找。
2.2 Maven项目骨架与核心依赖
在IDE里新建一个Java工程（我用的是Maven），pom.xml里加上最核心的依赖即可。
<properties> <flink.version>1.17.2</flink.version> </properties> <dependencies> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients</artifactId> <version>${flink.version}</version> </dependency> </dependencies>如果只是本地学习，不加flink-clients其实也能跑，但提交到集群、指定执行模式时会用到它，建议加上。注意这些依赖在打包提交集群时建议把scope设为provided，否则打出来的JAR会和集群自带的Flink包冲突，不过本地IDE运行无所谓。
2.3 在本地直接运行的两种姿势
- 直接在IDE里写
main方法，调env.execute()，Flink会以本地进程的方式启动一个MiniCluster执行任务，日志直接打在控制台，最适合调试算子逻辑。 - 在本地起一个独立测试集群，通过命令行提交作业，这种方式更接近生产环境，能看到任务调度、并行度分配的完整过程。
我建议初学者先用第一种方式跑通逻辑，再用第二种方式感受一次“打包、提交、查看Web UI”的完整流程。第二种方式的Web UI默认地址是localhost:8081，可以直观看到DAG、并行度、checkpoint耗时等信息。
3. 批量处理实战：用DataStream API读取CSV做分组聚合
到了实战环节，第一步先跑批量处理。虽然Flink给批处理提供了Table/SQL API和旧DataSet API，但为了后面能和流处理代码共享逻辑，我还是推荐直接用DataStream API。
3.1 批处理模式怎么开启
在Flink全局配置里有execution.runtime-mode，有三种取值：STREAMING（默认）、BATCH、AUTOMATIC。用代码设置时建议直接通过StreamExecutionEnvironment设置运行模式。
Configuration conf = new Configuration(); conf.setString("execution.runtime-mode", "BATCH"); StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);也可以使用env.setRuntimeMode(RuntimeExecutionMode.BATCH)，效果一样。
开启BATCH模式后有几个明显变化：
- 输入扫描完才会继续往下游走，能做的优化更多，比如排序、分区裁剪。
- 状态和checkpoint的管理方式不同，批处理不会像流式那样持续做周期检查点。
- 水位线这种专门处理无界流的概念基本失效，因为数据是有限的，窗口会在输入结束后一次性触发。
3.2 示例：销售数据分类汇总
假设我们有一个sales.csv，每一行代表一条订单，字段是分类,商品,金额：
手机,iPhone15,6999 手机,华为Mate60,4999 家电,冰箱,3000 家电,洗衣机,2500 手机,耳机,999需求很简单：按分类统计销售额合计。用DataStream API写出来很直接。
package com.demo.flink; import org.apache.flink.api.common.functions.ReduceFunction; import org.apache.flink.api.common.typeinfo.Types; import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; public class BatchSalesAgg { public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); conf.setString("execution.runtime-mode", "BATCH"); StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf); DataStream<String> lines = env.readTextFile("file:///tmp/sales.csv"); DataStream<SalesRecord> records = lines .map(line -> { String[] fields = line.split(","); return new SalesRecord(fields[0].trim(), fields[1].trim(), Double.parseDouble(fields[2].trim())); }) .returns(Types.POJO(SalesRecord.class)); DataStream<SalesRecord> aggregated = records .keyBy(SalesRecord::getCategory) .reduce((ReduceFunction<SalesRecord>) (r1, r2) -> new SalesRecord(r1.getCategory(), "all", r1.getAmount() + r2.getAmount())); aggregated.print(); env.execute("batch-sales-agg"); } }用reduce做累加是最直观的批处理思路：keyBy负责把同分类的订单分到同一个逻辑分组，reduce两两叠加，结果就是该分类的总额。SalesRecord就是个简单POJO，字段要带public的getter/setter，这里做了省略。
3.3 批量结果输出：注意小文件问题
批任务最常见的输出是落文件或落表。生产环境我一般用FileSink写文件，但有一个坑：如果并行度设得很大，会产生大量小文件，后面再被分析引擎读取时效率很低。
解决办法是在写入前做一次rebalance或partitionBy，把输出分区数收敛到和下游分区一致；如果用的是FileSink，还可以设置滚动策略，比如按大小或时间滚动。
FileSink<String> sink = FileSink .forRowFormat(new Path("hdfs:///tmp/sales_out"), new SimpleStringEncoder<String>("UTF-8")) .withRollingPolicy( DefaultRollingPolicy.builder() .withRolloverInterval(TimeUnit.MINUTES.toMillis(10)) .withInactivityInterval(TimeUnit.MINUTES.toMillis(5)) .withMaxPartSize(128L * 1024 * 1024) .build() ) .build(); aggregated.map(SalesRecord::toString).sinkTo(sink);这个策略的意思是：单个文件超过128MB、写文件超过10分钟、或文件空闲5分钟时，就滚动生成新文件。对批量任务来说，滚动策略能有效避免大量小文件。
4. 实时流处理实战：从Kafka消费事件流做连续统计
批量处理跑通之后，再来看看实时流处理。这里我以最常见的实时计算场景为例：消费Kafka里的用户行为日志，统计每分钟每个用户的活跃次数，顺便把超时产生的迟到数据抛到侧输出流里。
4.1 连接器选择与参数配置
从Flink 1.13开始，像Kafka这种外部系统的连接器不再跟着Flink主包走，而是拆成独立JAR。你需要单独引入：
<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka</artifactId> <version>1.17.2</version> </dependency>需要注意版本对应关系：1.17新版的Kafka连接器类名通常带v2或按Kafka协议版本区分。如果在网上搜到旧博客用FlinkKafkaConsumer011、FlinkKafkaConsumer010，那基本是Flink 1.11之前的用法，不要再套用。
连接Kafka的配置项里，bootstrap.servers、group.id是必填。还有一个细节：在流任务里，消费偏移量的提交和checkpoint绑定在一起。如果没开启checkpoint，Flink默认不向Kafka提交offset，作业重启后会根据auto.offset.reset策略决定从哪开始读，这也是常见的数据“重了”或“丢了”的原因之一。
Properties props = new Properties(); props.setProperty("bootstrap.servers", "localhost:9092"); props.setProperty("group.id", "flink-user-events-group"); DataStream<String> sourceStream = env.addSource( new FlinkKafkaConsumer<String>( "user-events", new SimpleStringSchema(), props ).setStartFromLatest() );4.2 事件时间、水位线与窗口的配合
消费进来的日志是一条JSON字符串，先要做反序列化，变成一个UserEvent对象。这个对象至少有三个字段：userId、eventTime、eventType。
实时统计的关键是时间语义。默认ProcessingTime是指运行机器处理事件那一刻的时间，但业务通常关心的是事件的真实发生时间，也就是EventTime。我们给数据流分配时间戳和水位线：
DataStream<UserEvent> eventStream = sourceStream .map(json -> parseEvent(json)) .assignTimestampsAndWatermarks( WatermarkStrategy.<UserEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((event, timestamp) -> event.getEventTime()) );forBoundedOutOfOrderness(Duration.ofSeconds(5))表示允许最多5秒的乱序延迟，也就是说水位线等于“已收到最大事件时间戳减去5秒”。水位线的作用是告诉窗口：“到此刻为止，早于水位线的事件都应该到齐了，可以触发计算了”。
然后用窗口做每分钟的活跃统计：
DataStream<ActiveStat> activeStats = eventStream .keyBy(UserEvent::getUserId) .window(TumblingEventTimeWindows.of(Time.minutes(1))) .allowedLateness(Time.minutes(1)) .sideOutputLateData(lateTag) .process(new ProcessWindowFunction<>() { @Override public void process(String key, Context context, Iterable<UserEvent> elements, Collector<ActiveStat> out) { out.collect(new ActiveStat(key, context.window().getEnd(), countElements(elements))); } });这里有三点要注意：
allowedLateness(Time.minutes(1))：允许事件迟到1分钟。1分钟窗口结束后，只要水位线还在“窗口结束时间+1分钟”之前，迟到的数据仍能进入窗口并触发更新。sideOutputLateData(lateTag)：把超过允许迟到范围的数据放进侧输出流，不至于直接丢掉，方便事后观测。ProcessWindowFunction能拿到窗口开始/结束时间和完整元素集合，适合做统计和调试；如果追求性能，一般用“增量聚合（reduce/aggregate）+ ProcessWindowFunction”组合，避免把所有数据都攒在内存里。
4.3 状态与精确一次语义
实时任务比批量任务复杂的地方在于涉及状态。比如要判断“某个用户是否首次出现”或者“当前累计金额是否超过阈值”，如果不做状态存储，很难做到精确统计。Flink的状态默认保存在内存或RocksDB中，并且会定期做checkpoint。
推荐在生产环境开启checkpoint，间隔可以设成30秒到1分钟：
env.enableCheckpointing(60_000); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30_000); env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setExternalizedCheckpointCleanup( ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);这里checkpointingMode设为EXACTLY_ONCE后，配合Kafka连接器的两阶段提交，能做到端到端精确一次：每个事件只会影响最终结果一次。这个特性在生产中非常关键，比如交易金额、库存扣减这类场景，重复计算往往意味着资损。
不过要注意：精确一次不是免费的，它会带来更低吞吐和更高延迟。如果业务能容忍偶尔重复，其实用AT_LEAST_ONCE简单省事。
5. 同一套代码跑批与流：执行模式切换背后的差异
前面分别演示了批处理和流处理的例子，如果仔细看代码，你会发现它们的算子非常像：都是map、keyBy、reduce/window/process，最大的区别只是执行模式和输入源不同。这正是Flink“批流一体”的工程价值。
5.1 STREAMING与BATCH模式下发生了什么变化
我用一张表来总结两者在引擎层的主要差异：
| 维度 | STREAMING模式 | BATCH模式 |
|---|---|---|
| 数据源 | 无界流，持续到达 | 有界数据，读取完成后结束 |
| 窗口触发 | 时间/数量/会话窗口动态触发 | 输入全部结束后统一触发 |
| 水位线 | 必须处理，决定乱序事件何时进入窗口 | 基本不需要，有界数据天然知道边界 |
| checkpoint | 周期性执行，保证故障恢复 | 内部用排序和恢复机制，不依赖持续checkpoint |
| 状态后端 | 内存/RocksDB，需要持久化 | 常用于局部聚合，排序后可优化 |
| 容错恢复 | 从最近checkpoint恢复 | 任务失败后往往整体重跑 |
这个表格在面试中也常被问到，建议多看几遍。核心结论是：Flink不是“批处理抄流处理”，也不是“流处理抄批处理”，而是底层共用一套分布式执行引擎，在碰到有界输入时对优化器开绿灯。
5.2 窗口语义在两种模式下的区别
窗口这个概念很典型。在流模式下，每分钟窗口要靠水位线来决定什么时候关闭；在批模式下，因为输入是有限的，Flink可以直接把所有属于某分钟的数据都读进来，然后在“读取完输入”时一次性触发窗口计算，不需要水位线，也不会有“窗口等待”的问题。
换句话说，批模式天然解决了一个流计算头疼的问题：窗口边界不确定。比如你统计“每分钟订单量”，如果上游数据乱序严重，流式的窗口可能飘来飘去；但如果从HDFS上读一天的历史日志，批量统计每一分钟的数据，结果是完全确定的。这也就是为什么离线报表总是比实时大屏“更准”的原因——不是实时引擎不对，而是无界输入天然带有不确定性。
5.3 什么时候该用批，什么时候该用流
运行模式不是越流式越好。我的选型经验如下：
- 数据源是Kafka这类消息队列，且要求秒级/分钟级延迟的，用
STREAMING。 - 数据源是文件或数据库全量导出，要求结果精确、可重复跑的，用
BATCH。 - 数据源虽然是文件，但希望“文件一旦产生就能立刻计算，不等到全量结束”，这时候用
STREAMING读文件目录也能实现增量计算，不过边界和文件滚动逻辑要设计好。
更推荐的做法是：用DataStream API把业务逻辑封装成公共算子，再根据输入源和执行模式做参数切换。一份核心逻辑，既服务实时大屏，也服务离线对账，这是Flink统一批流后最值回票价的地方。
6. 运行Flink作业时最值得记录的排错清单
这部分是实战中的碎坑汇总。我把印象最深的几个问题按“现象、排查、解决”列出来，很多问题不是逻辑写错，而是对Flink运行机制的误判。
6.1 窗口迟迟不触发：先怀疑水位线
现象很常见：实时作业跑起来了，Kafka也有数据，但事件时间窗口的统计结果一直不出。排查逻辑是这样的：
第一步，在ProcessWindowFunction外用process算子打印每批数据的时间戳，看看最大事件时间是否在增长。第二步，打印水位线，观察数据流里的时间戳信息。如果水位线不动，说明要么没有事件时间字段，要么上游某个分区没有数据。
很多初学者会忽略：forBoundedOutOfOrderness生成的水位线是全局的，如果并行度里有1个子任务分区一直没数据，默认策略会等待这个分区的最大时间戳，整体水位线就可能卡死。解决办法是给WatermarkStrategy加withIdleness(Duration.ofSeconds(30))，让空闲分区不再阻塞水位线推进。
WatermarkStrategy.<UserEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withIdleness(Duration.ofSeconds(30)) .withTimestampAssigner(...)这个坑我自己踩过好几次，每次都是线上大屏不出数才被叫去查。
6.2 Kafka消费的group.id与checkpoint纠缠
另一个高频坑：Kafka消息被消费了，作业重启后却从最早开始重新消费一遍。原因有两个方向：
- 没有开启checkpoint，也没有手动提交offset，FlinkKafkaConsumer默认不提交，重启后Kafka的
auto.offset.reset生效，如果策略是earliest就会重读。 - 开启了checkpoint，但Flink里的offset和Kafka的offset记录是两回事，如果
group.id变了，也会从新的偏移开始。
解决思路：确认env.enableCheckpointing是否开启；确认group.id是否稳定；如果调试环境想从头或从最新读，用setStartFromEarliest()和setStartFromLatest()显式控制。
6.3 IDE里直接跑作业报ClassNotFound
本地IDE经常可以正常跑通，但一打包提交到集群，就报各种ClassNotFoundException，比如缺少Kafka连接器类、缺少JSON解析类。原因是打包时没有把依赖打进去。
解决办法是用maven-shade-plugin把依赖合并到同一个JAR里，或者至少确保运行时聚类包含所需依赖。如果使用providedscope的Flink主包，连接器这种额外依赖不要用provided。
<plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <version>3.2.4</version> <executions> <execution> <phase>package</phase> <goals><goal>shade</goal></goals> </execution> </executions> </plugin>另外，如果JSON反序列化只用SimpleStringSchema然后再手动解析，那会相对稳定；如果使用第三方序列化器，要注意版本兼容。Kafka连接器和某些序列化库之间偶尔有二进制兼容问题，尽量保持Flink小版本统一。
6.4 数据倾斜导致的背压和延迟
最后再提一个常见性能问题：某个key的数据量特别大，导致单个子任务处理不过来，表现为整个作业延迟升高、背压严重。排查方式是看Web UI里各Task的发送速率和接收速率，或者从任务管理日志看哪个线程处理最慢。
解决思路有三种：
- 加并行度，但要配合
keyBy的分组键设计，单纯“加并行度”对数据倾斜帮助有限。 - 对热点key做二次打散：比如先加随机后缀分摊，再进行一次全量聚合，这里会牺牲一点精确性。
- 用
map阶段加两层聚合（局部聚合 + 全局聚合），思路和MapReduce里的Combine类似。
我在实时成交金额统计里遇到过单个头部用户贡献了80%流量，当时就是靠“热点key加盐 + 两层聚合”解决的。
写到这里，Flink批量处理和实时流处理的应用路径已经基本清楚了：从环境搭建、DataStream API编程，到执行模式切换，再到线上排错，都是工程实践中绕不开的点。我个人最建议的做法是先别急着上集群和全套监控，把本节示例在本地跑通，换个数据源、改改并行度、试试水位线调大调小的影响，再谈特性应用。状态、checkpoint、精确一次这些概念，光看文档印象不深，亲手触发一次数据乱序、看一次水位线卡顿，理解才会到位。
最后补充一个调试小技巧：在本地IDE跑流作业时，把并行度设小一点，打开Web UI的算子配置加一行日志，能看到非常细的算子间数据流动情况。尤其是水位线和时间戳字段，一眼就能看出到底是谁拖慢了窗口触发。这个习惯帮我排查掉了大量看不见的隐性问题。
