凌晨一点半我被一条告警电话叫醒。实时大屏上的成交金额和订单数量跟凌晨跑批出来的离线报表差了将近两个数量级数据组同事在群里反复贴截图“到底以哪个为准”这种事在数据岗待过的人应该都懂业务方既要秒级更新的实时指标又要凌晨全量的对账结果而团队手里往往不止一套处理系统。当时我负责的数据任务就是典型的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可以同时处理两者。你在代码里写一个DataStreamX把运行模式设为BATCH就是批处理设为STREAMING就是流处理代码骨架几乎不用变。窗口、状态、水位线这些流处理概念在批处理模式下也会有对应的简化语义。换句话说Flink不是简单地把Spark Streaming那种“微批”包装成“流”而是真的在做事件级流处理同时反过来在批模式下借用流式执行引擎的分布式能力和容错机制。这种设计带来的直接好处是同一套代码既能做每天凌晨的批计算又能支撑实时大屏。维护成本从两套降为一套数据口径也能收敛到同一种表达上。2. 十分钟搭好本地环境Flink快速上手的依赖与配置接触Flink的第一步其实不是部署集群而是先把本地开发环境跑通。一个最简单的Flink作业在IDE里就能直接跑不需要本地先起一个Flink集群。这一点和写普通Java应用一样轻松反而是版本选错会浪费不少时间。2.1 版本选择比想象中更重要我用过的Flink版本里1.12是一个分水岭它正式把DataSet API标记为冻结状态全面推荐DataStream API做批处理。所以你现在看网上很多老教程还在用ExecutionEnvironment.fromElements、DataSetT这些代码在1.12以上的版本里编译可能能过但已经不建议用了到了1.18之后旧的DataSet API入口直接被移除。我个人推荐直接从Flink 1.14或更高版本起步比如我用的是1.17.2。原因很简单新版内置了RuntimeExecutionMode支持STREAMING和BATCH两种执行模式同时Kafka连接器拆成了独立JAR管理版本对应关系更清晰。跟实时计算相关的组件生态也都在往新版本迁移网上踩坑资料相对好找。2.2 Maven项目骨架与核心依赖在IDE里新建一个Java工程我用的是Mavenpom.xml里加上最核心的依赖即可。properties flink.version1.17.2/flink.version /properties dependencies dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version${flink.version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-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); DataStreamString lines env.readTextFile(file:///tmp/sales.csv); DataStreamSalesRecord 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)); DataStreamSalesRecord aggregated records .keyBy(SalesRecord::getCategory) .reduce((ReduceFunctionSalesRecord) (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还可以设置滚动策略比如按大小或时间滚动。FileSinkString sink FileSink .forRowFormat(new Path(hdfs:///tmp/sales_out), new SimpleStringEncoderString(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 groupIdorg.apache.flink/groupId artifactIdflink-connector-kafka/artifactId version1.17.2/version /dependency需要注意版本对应关系1.17新版的Kafka连接器类名通常带v2或按Kafka协议版本区分。如果在网上搜到旧博客用FlinkKafkaConsumer011、FlinkKafkaConsumer010那基本是Flink 1.11之前的用法不要再套用。连接Kafka的配置项里bootstrap.servers、group.id是必填。还有一个细节在流任务里消费偏移量的提交和checkpoint绑定在一起。如果没开启checkpointFlink默认不向Kafka提交offset作业重启后会根据auto.offset.reset策略决定从哪开始读这也是常见的数据“重了”或“丢了”的原因之一。Properties props new Properties(); props.setProperty(bootstrap.servers, localhost:9092); props.setProperty(group.id, flink-user-events-group); DataStreamString sourceStream env.addSource( new FlinkKafkaConsumerString( user-events, new SimpleStringSchema(), props ).setStartFromLatest() );4.2 事件时间、水位线与窗口的配合消费进来的日志是一条JSON字符串先要做反序列化变成一个UserEvent对象。这个对象至少有三个字段userId、eventTime、eventType。实时统计的关键是时间语义。默认ProcessingTime是指运行机器处理事件那一刻的时间但业务通常关心的是事件的真实发生时间也就是EventTime。我们给数据流分配时间戳和水位线DataStreamUserEvent eventStream sourceStream .map(json - parseEvent(json)) .assignTimestampsAndWatermarks( WatermarkStrategy.UserEventforBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((event, timestamp) - event.getEventTime()) );forBoundedOutOfOrderness(Duration.ofSeconds(5))表示允许最多5秒的乱序延迟也就是说水位线等于“已收到最大事件时间戳减去5秒”。水位线的作用是告诉窗口“到此刻为止早于水位线的事件都应该到齐了可以触发计算了”。然后用窗口做每分钟的活跃统计DataStreamActiveStat 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, IterableUserEvent elements, CollectorActiveStat 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.UserEventforBoundedOutOfOrderness(Duration.ofSeconds(5)) .withIdleness(Duration.ofSeconds(30)) .withTimestampAssigner(...)这个坑我自己踩过好几次每次都是线上大屏不出数才被叫去查。6.2 Kafka消费的group.id与checkpoint纠缠另一个高频坑Kafka消息被消费了作业重启后却从最早开始重新消费一遍。原因有两个方向没有开启checkpoint也没有手动提交offsetFlinkKafkaConsumer默认不提交重启后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 groupIdorg.apache.maven.plugins/groupId artifactIdmaven-shade-plugin/artifactId version3.2.4/version executions execution phasepackage/phase goalsgoalshade/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的算子配置加一行日志能看到非常细的算子间数据流动情况。尤其是水位线和时间戳字段一眼就能看出到底是谁拖慢了窗口触发。这个习惯帮我排查掉了大量看不见的隐性问题。