1. 项目概述为什么从8.1.1开始学Flink做数据处理这些年我越来越觉得Flink是绕不开的一个家伙尤其是当你面对“既要批处理又要实时流处理”这种双重要求的时候Flink几乎是第一梯队的答案。8.1.1这个版本号看着像是小版本但它在API稳定性、窗口计算、状态管理以及和主流数据源的连接器方面已经打磨得非常顺手了。我写这篇内容就是想用一个实际项目的思路把Flink怎么玩批量数据、怎么玩实时流数据这件事讲透里面会包含我踩过的坑、看过的源码注释、以及一些网上文档里很少写到的小细节。如果你是刚接触Flink或者已经写过WordCount但面对真实业务场景还是有点懵这篇内容会很适合你。如果你已经用Flink跑过一些任务那这里关于状态清理、窗口触发、反压排查的部分也值得你重新扫一遍。我不会把每个API都罗列一遍而是按“先想清楚再动手”的方式带你把批量处理和实时流处理各做一遍最后再回头聊几个高频问题。这个项目叫“8.1.1 Flink快速应用”说直白点就是用一套代码能力把静态文件的批量统计和实时消息流的实时统计都做出来。它本身不复杂但把Flink的核心思维——批流一体、有状态计算、事件时间处理——都串起来了。后面我所有实操内容都是以我某次模拟业务的数据为例子把步骤和代码拆开讲。2. 批量处理数据实操先别急着写代码2.1 批量数据场景到底需要什么批量处理说白了就是“数据已经躺在那儿了不管它是文件、数据库表还是对象存储里的分区我们一次性把它们读出来算完”。这种场景最常见的就是日报、周报、离线指标计算、历史数据清洗。Flink做批处理的优势不在于它比Hive SQL快多少而在于同一套代码既能跑批又能跑流不用维护两套引擎基于DataStream API的批处理写起来比MapReduce简单太多生态连接器齐全读HDFS、读MySQL、读Parquet都有现成的东西。我见过不少团队明明只需要一个离线统计非要用Spark然后等到后面要做实时需求时又得再搭一套Flink集群。其实从一开始就选Flink后面会舒服很多。2.2 环境准备和项目骨架搭建我这次用的模拟环境是一台开发机装了Java 8和Maven 3.6Flink版本8.1.1对应的Scala版本是2.12。这里有个坑如果你用的Java版本太高比如Java 17某些Flink老版本会有模块访问限制跑本地任务时偶尔报IllegalAccessError。所以建议直接用Java 8或Java 11省心。项目依赖我只列最核心的几项dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version8.1.1/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-clients/artifactId version8.1.1/version /dependency如果是纯批处理不需要额外引入flink-batch因为Flink的批处理已经统一到DataStream API里了。这一点和老的Flink 1.x时代不一样那时候Batch API和Streaming API是分开的现在8.1.1这套统一模型是真的好用。2.3 批量读取与计算示例我先用一个最简单的场景读取本机一个包含订单记录的CSV文件统计每个品类的总销售额。文件长这样order_id, category, amount 1001,electronics,299.00 1002,books,59.00 1003,electronics,1299.00 1004,home,89.00代码核心如下StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setRuntimeMode(RuntimeExecutionMode.BATCH); DataStreamSourceString source env.readTextFile(/data/orders.csv); SingleOutputStreamOperatorTuple2String, Double categoryAmount source .map(line - { String[] fields line.split(,); if (fields.length 3) { return null; } return Tuple2.of(fields[1], Double.parseDouble(fields[2])); }) .filter(t - t ! null); categoryAmount .keyBy(t - t.f0) .reduce((t1, t2) - Tuple2.of(t1.f0, t1.f1 t2.f1)) .map(t - 品类: t.f0 销售额: t.f1) .print(); env.execute(batch-category-sales);注意三件事setRuntimeMode(RuntimeExecutionMode.BATCH)告诉Flink这是批任务底层会使用批式调度和数据交换策略更节省资源。readTextFile不支持从压缩包直接读如果文件是gz格式要改成readFile配合CompressedInputStream或者先解压再读。filter(t - t ! null)不能少因为我见过文件末尾有换行符导致解析出空数组的情况不加filter任务不会报错但结果会多出一条脏数据排查起来挺浪费时间的。跑完这个任务你会看到类似下面的输出品类: books 销售额: 59.0 品类: electronics 销售额: 1598.0 品类: home 销售额: 89.02.4 批量处理的资源调优心得很多人以为批处理就是把env.execute()跑完就没事了。其实8.1.1里批任务的并行度、网络缓冲、状态后端这些都要稍微想一下。我习惯这样配置并行度如果数据量在GB级别并行度设置为CPU核心数的2-3倍不要贪多。并行度太大容易导致shuffle时网络瓶颈太小又浪费资源。状态后端批任务用HashMapStateBackend就够不需要RocksDBStateBackend因为批处理状态生命周期短不涉及大状态增量检查点。网络缓冲如果任务出现“频繁Full GC但堆内存还有很多”这种诡异情况不妨调大taskmanager.memory.network.max给网络缓冲多留一点空间。这些经验都是我反复调优总结出来的不一定适合所有场景但至少能让你在刚开始跑批任务时少走弯路。3. 实时流处理数据实操从Kafka到实时指标3.1 实时流场景需求拆解实时流处理和批量处理的本质区别在于数据是持续不断到达的任务一旦启动就不会自己停结果要随着时间不断更新。这个项目里我模拟了一个电商实时订单流业务方要求实时统计每分钟内每个品类的订单数和GMV延迟要求是5秒以内。从技术选型上上游数据我用了Kafka因为这是目前最主流的接入方式。Flink作为消费者从Kafka的topic里读取订单数据然后做窗口计算、聚合、输出。3.2 创建Kafka数据源和Flink消费端先准备一个简单的Kafka生产者模拟不断产生订单消息。生产者的代码我用纯Java写每秒随机生成1-3条订单Properties props new Properties(); props.put(bootstrap.servers, localhost:9092); props.put(key.serializer, org.apache.kafka.common.serialization.StringSerializer); props.put(value.serializer, org.apache.kafka.common.serialization.StringSerializer); KafkaProducerString, String producer new KafkaProducer(props); Random random new Random(); String[] categories {electronics, books, home, clothing}; while (true) { String category categories[random.nextInt(categories.length)]; double amount 10 random.nextDouble() * 2000; String message category , String.format(%.2f, amount); producer.send(new ProducerRecord(order-topic, message)); Thread.sleep(random.nextInt(500) 200); }Flink端消费代码如下Properties kafkaProps new Properties(); kafkaProps.setProperty(bootstrap.servers, localhost:9092); kafkaProps.setProperty(group.id, flink-order-group); DataStreamSourceString stream env.addSource( new FlinkKafkaConsumer(order-topic, new SimpleStringSchema(), kafkaProps) );这里有个非常重要的点FlinkKafkaConsumer是带状态的它会记录消费到的Kafka offset。如果你任务重启并且没有启用检查点那它可能会重复消费或者从最新位置开始。所以生产环境一定要开启Checkpoint这个后面我专门讲。3.3 实时流处理窗口计算与代码过程实时流的核心是窗口。我用的是TumblingEventTimeWindows也就是滚动事件时间窗口。先定义订单实体类再设定事件时间水印然后做分类聚合。先看事件时间设置SingleOutputStreamOperatorOrder orderStream stream .map(line - { String[] fields line.split(,); return new Order(fields[0], Double.parseDouble(fields[1]), System.currentTimeMillis()); }) .assignTimestampsAndWatermarks( WatermarkStrategy.OrderforBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((order, ts) - order.getEventTime()) );forBoundedOutOfOrderness(Duration.ofSeconds(5))是允许事件时间乱序最多5秒。也就是说一个事件比窗口结束时间晚到但不超过5秒依然可以进入窗口。这里要解释下为什么是5秒而不是0秒真实业务中客户端上报时间往往和服务器接受时间有偏差移动端网络抖动也很常见。如果设置成0容忍你会发现大量数据被丢到迟到侧输出指标和业务方对不上。窗口计算代码orderStream .keyBy(Order::getCategory) .window(TumblingEventTimeWindows.of(Time.minutes(1))) .aggregate(new AggregateFunctionOrder, Tuple2Long, Double, String() { Override public Tuple2Long, Double createAccumulator() { return Tuple2.of(0L, 0.0D); } Override public Tuple2Long, Double add(Order value, Tuple2Long, Double acc) { return Tuple2.of(acc.f0 1, acc.f1 value.getAmount()); } Override public String getResult(Tuple2Long, Double acc) { return 订单数: acc.f0 , GMV: String.format(%.2f, acc.f1); } Override public Tuple2Long, Double merge(Tuple2Long, Double a, Tuple2Long, Double b) { return Tuple2.of(a.f0 b.f0, a.f1 b.f1); } }) .print();aggregate比reduce灵活的地方在于AggregateFunction可以维护更复杂的累加器状态比如同时累计订单数和金额。用reduce的话你得自己包一个自定义对象会麻烦不少。窗口设置成1分钟配合水印延迟5秒实际输出的窗口结果不会在窗口结束的那一瞬间立即打印而是会等到水印越过窗口结束时间后再触发。换句话说你看到结果的时间比业务上的窗口结束时间晚几秒。这个延迟是正常的不需要排查。3.4 状态管理和检查点配置实时流任务里状态管理和检查点是最重要的部分。我的习惯是任务一启动就开启Checkpoint配置如下CheckpointConfig checkpointConfig env.getCheckpointConfig(); checkpointConfig.setCheckpointStorage(hdfs:///flink-checkpoints); checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); checkpointConfig.setMinPauseBetweenCheckpoints(500); checkpointConfig.setCheckpointTimeout(60000); checkpointConfig.setMaxConcurrentCheckpoints(1); env.enableCheckpointing(5000);解释几个关键参数setCheckpointingMode(EXACTLY_ONCE)保证端到端精确一次语义。前提是下游也支持事务性写入比如Kafka sink、JDBC sink如果不支持那只能做到至少一次别被“精确一次”这个词骗了。setMinPauseBetweenCheckpoints(500)两个检查点之间至少间隔500毫秒防止频繁检查点影响吞吐。你想想如果一个窗口1分钟才输出一次结果检查点却每200毫秒打一次那除了浪费资源没别的意义。setCheckpointTimeout(60000)单个检查点如果60秒还没完成就取消它。这个参数主要用在状态特别大的任务上如果检查点总是超时你要么调大超时时间要么优化状态大小。状态后端的选型上如果订单流的状态量在几十GB以下用HashMapStateBackend就够了性能好、延迟低。如果状态到了百GB甚至TB级别那就必须上RocksDBStateBackend它利用磁盘存储状态虽然每轮访问有序列化开销但能撑住更大规模。注意RocksDBStateBackend需要额外引入依赖不是Flink核心自带。3.5 实时流项目中的反压问题反压是实时流任务里最常出现的性能杀手。症状就是Kafka消费速度越来越慢任务延迟越来越大但CPU和内存看起来都还有富余。很多经验不足的同学这时候第一反应是加并行度结果加了并行度反而更慢因为问题不在计算能力而在下游或网络。排查反压的基本方法打开Flink Web UI看每个算子的inPoolUsage和outPoolUsage数值接近1就说明有反压。从最下游算子往回找找到第一个出现高反压的算子通常就是瓶颈所在。如果瓶颈是聚合算子看是不是key分布倾斜导致某个子任务积压了很多数据。如果瓶颈是Sink看下外部系统写入延迟比如Kafka或ES是否打满。我在这个项目里就遇到过一个有意思的情况上游Kafka生产者每秒才发几条数据按理说Flink毫无压力但任务运行10分钟后延迟越来越大。后来看日志发现是因为我没有给SimpleStringSchema指定字符集默认用了平台默认编码在有些环境下处理特殊字符时耗了大量时间做编码转换。这个坑很小但排查起来相当费劲。4. 从批处理体系到实时流体系的融合实践4.1 批流一体统一API的价值我之所以把批量处理和实时流处理放在同一个标题里是因为Flink 8.1.1的批流一体做得确实不错。在8.1.1中DataStream API既能跑有界流也能跑无界流。有界流就是批无界流就是实时。同一个逻辑比如keyBy、window、aggregate两种模式下代码几乎一致区别只是setRuntimeMode和有界无界的数据源。这带来一个巨大的好处你可以先在本地用批量数据调试逻辑验证无误之后再把数据源换成Kafka秒变实时任务。不用担心批量模式能跑、流模式就挂了因为执行计划大部分是共享的。我实际开发中就是这么干的先造一批历史数据文件把窗口和水印逻辑调好再切成Kafka源。4.2 将批量逻辑改造成流处理逻辑的步骤从一个批量任务切换成实时任务通常涉及以下改动数据源从readTextFile换成FlinkKafkaConsumer。如果原来是用execute()之前的数据集直接聚合现在要加上窗口和水印。为keyBy之后的算子设计好状态策略避免无限增长。输出目标从print()改成写入数据库或Kafka。这里最容易漏的是水印。批量模式下没有水印的概念因为所有数据都在那哪有什么迟到不迟到。但流模式下如果没有水印窗口永远不会触发因为Flink不知道事件时间走到哪了。我见过有人从批切流结果窗口半天不出结果就是因为忘了加assignTimestampsAndWatermarks。另一个容易漏的是状态过期时间。实时任务无界如果对某个key聚合一次后状态就一直保留时间长了状态会无限膨胀。所以用keyBy做聚合时最好给状态配置TTLStateTtlConfig ttlConfig StateTtlConfig .newBuilder(Time.hours(2)) .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) .build();TTL设置要根据业务需求定比如热数据时效性只有2小时那状态超过2小时就可以让Flink异步清理掉。注意Flink不会精确到秒级清理TTL状态它只在状态被访问或者后台定期清理线程触发时才会真正删掉。4.3 双链路任务如何共用一套代码我是这样设计目录结构的一个Maven项目里面按batch和stream分两个包但核心的聚合逻辑封装成一个公共类这样batch和stream只是数据源和输出不同计算逻辑不变。比如我定义一个SalesAggregator里面有静态方法public static SingleOutputStreamOperatorString aggregate(KeyedStreamOrder, String keyedStream) { return keyedStream .window(TumblingEventTimeWindows.of(Time.minutes(1))) .aggregate(new SalesAggregateFunction()); }批量任务里我用批量数据源构造出KeyedStream调SalesAggregator.aggregate然后输出到文件实时任务里用Kafka源构造KeyedStream调同一个方法输出到Kafka或数据库。公共逻辑只写一遍后面维护起来真的很省心。还有一个细节批量任务里的Time.minutes(1)和实时任务里的Time.minutes(1)语义完全一样吗不完全一样。批量模式下窗口时间由数据里的eventTime字段确定实时模式下窗口时间同样由eventTime字段确定但因为有水印机制窗口的输出时间会晚于事件时间。如果业务方要求“必须准点出结果”那要考虑用ProcessingTime窗口但这会导致延迟虽然低结果准确性下降尤其乱序数据多的时候。所以时间语义的选择要提前和业务说清楚不能默认大家都懂。5. 常见问题排查与避坑技巧5.1 并行度设置的正确姿势并行度这个问题刚开始用Flink的人基本都会问。我的建议分三步走第一步看数据源分区数。比如Kafka topic有10个分区你的并行度就别低于10不然有分区空闲没人消费。第二步看窗口/聚合的计算压力。如果是简单过滤并行度设成Kafka分区数的1倍就够如果涉及keyBy后的热点key并行度适当增加到2倍让计算压力分摊开。第三步实测调整。看Web UI的每个子任务处理记录数如果长期不均衡说明key分布有问题或并行度不合理。不要盲目把并行度调得很大。并行度一高算子之间的网络连接数量成倍增加序列化和反序列化开销也随之上升。我有一个很强烈的个人体会并行度从4调到8可能提升50%从8调到16可能只提升5%从16调到32可能反而下降。原因就是网络shuffle的成本超过了并行计算带来的收益。5.2 Kafka offset提交异常Kafka消费者offset提交异常是我最常遇到的第二个问题。症状是任务不断重启每次重启都从最早的位置开始消费导致数据重复计算。原因通常是检查点开启后offset的提交方式变了。Flink在开启Checkpoint后默认把offset提交到检查点里而不是Kafka的__consumer_offsets。这样如果检查点失败offset也不会提交下次重启时恢复检查点里的旧offset这本身是合理的。但如果你在Flink之外再用Kafka自带的Consumer Group去读同一个topic你会发现group的offset一直不动于是产生“Flink没提交offset”的错觉。解决办法在Flink的FlinkKafkaConsumer中设置setStartFromLatest()或者在Kafka工具里手动调整偏移。另外如果你的下游系统不支持事务消息那你开EXACTLY_ONCE也没用实际还是会重复这时业务幂等性就非常重要比如数据库里用唯一键去重。5.3 窗口结果延迟或不触发窗口不触发或延迟触发我总结了四种常见原因水印推进太慢forBoundedOutOfOrderness设得太长比如设了1小时那窗口要等很久才触发。解决办法是设一个业务可接受的乱序容忍度比如5-10秒。事件时间的字段解析错误如果withTimestampAssigner返回的时间戳不是毫秒而是秒那所有数据的时间都会变成1970年附近窗口永远不会触发。没有数据进来窗口只有在收到第一条数据后才会创建。如果某类目一小时内没有订单那这个类目在这一小时里就不会有窗口结果。这是正常的不是bug。checkpint失败导致恢复任务Flink从最近一次成功的检查点恢复如果那段时间的窗口状态已经清掉了可能导致部分结果丢失或延迟。我建议大家在写窗口逻辑时务必加allowedLateness和侧输出。allowedLateness允许窗口结束后的一段时间内迟到的数据还能更新结果侧输出则是把那些实在太晚、进不了窗口的数据单独收集起来方便排查和补偿。OutputTagOrder lateTag new OutputTagOrder(late-orders) {}; SingleOutputStreamOperatorString result orderStream .keyBy(...) .window(...) .allowedLateness(Time.seconds(30)) .sideOutputLateData(lateTag) .aggregate(...); DataStreamOrder lateStream result.getSideOutput(lateTag);迟到的数据从侧输出里读出来以后可以打印成日志也可以直接丢弃但至少要让人知道到底有多少数据迟到了。这样你和业务方对指标时就不会被一句“数据不对啊”搞得措手不及。5.4 序列化性能问题Flink默认的序列化器对Java元组、POJO都很友好但对一些自行定义的类型如果没有无参构造和getter/setter可能无法直接序列化导致作业提交时直接报错。这个时候先把POJO类写好别想着用TypeInformation.of强行绕过。另外如果某个字段是MapString, ListString这种复杂嵌套结构序列化开销会比较高。我在这个项目里把一个事件拆成了多个简单字段并在map阶段就把金额从String转成double避免后续每处理一条都做一次字符串转数字。这种微小的优化在数据量大的时候差别很明显。6. 最后的几条经验总结做Flink资料整理这段时间我自己最大的变化是遇到一个新的数据处理需求不再急着选Spark还是Flink而是先看数据是“已经躺好”还是“源源不断来”。如果两者都有那就直接选Flink把API熟悉一遍并行度、状态、检查点、窗口这几个概念搞清楚后面基本能应对大部分场景。另外我真的建议你在本地把Demo跑通而不是只读文档。就拿我在这个项目里用的ThrowingKafkaProducer来说没有真正跑过一遍你永远不知道它会怎么影响你的任务。遇见有人问“为什么我加了这个没生效”一查基本都是缺少配置或依赖没有引入。这类问题看文档十遍不如自己踩一次坑来得实在。最后说一个实操小技巧如果你在本地用env.execute()跑流任务记得在IDE的配置里加上“-Dlog4j.configurationFilelog4j2.xml”不然控制台日志可能刷到崩溃真正关键的错误信息反而被淹没了。这个小细节我特意放在最后就是因为它看起来太小但实际体验差别真的很大。 
长河君 · 2026/10/22 10:07:18 · 阅读 1.2k Flink 8.1.1批流一体实战：从批量统计到实时流处理的核心技术解析
1. 项目概述为什么从8.1.1开始学Flink做数据处理这些年我越来越觉得Flink是绕不开的一个家伙尤其是当你面对“既要批处理又要实时流处理”这种双重要求的时候Flink几乎是第一梯队的答案。8.1.1这个版本号看着像是小版本但它在API稳定性、窗口计算、状态管理以及和主流数据源的连接器方面已经打磨得非常顺手了。我写这篇内容就是想用一个实际项目的思路把Flink怎么玩批量数据、怎么玩实时流数据这件事讲透里面会包含我踩过的坑、看过的源码注释、以及一些网上文档里很少写到的小细节。如果你是刚接触Flink或者已经写过WordCount但面对真实业务场景还是有点懵这篇内容会很适合你。如果你已经用Flink跑过一些任务那这里关于状态清理、窗口触发、反压排查的部分也值得你重新扫一遍。我不会把每个API都罗列一遍而是按“先想清楚再动手”的方式带你把批量处理和实时流处理各做一遍最后再回头聊几个高频问题。这个项目叫“8.1.1 Flink快速应用”说直白点就是用一套代码能力把静态文件的批量统计和实时消息流的实时统计都做出来。它本身不复杂但把Flink的核心思维——批流一体、有状态计算、事件时间处理——都串起来了。后面我所有实操内容都是以我某次模拟业务的数据为例子把步骤和代码拆开讲。2. 批量处理数据实操先别急着写代码2.1 批量数据场景到底需要什么批量处理说白了就是“数据已经躺在那儿了不管它是文件、数据库表还是对象存储里的分区我们一次性把它们读出来算完”。这种场景最常见的就是日报、周报、离线指标计算、历史数据清洗。Flink做批处理的优势不在于它比Hive SQL快多少而在于同一套代码既能跑批又能跑流不用维护两套引擎基于DataStream API的批处理写起来比MapReduce简单太多生态连接器齐全读HDFS、读MySQL、读Parquet都有现成的东西。我见过不少团队明明只需要一个离线统计非要用Spark然后等到后面要做实时需求时又得再搭一套Flink集群。其实从一开始就选Flink后面会舒服很多。2.2 环境准备和项目骨架搭建我这次用的模拟环境是一台开发机装了Java 8和Maven 3.6Flink版本8.1.1对应的Scala版本是2.12。这里有个坑如果你用的Java版本太高比如Java 17某些Flink老版本会有模块访问限制跑本地任务时偶尔报IllegalAccessError。所以建议直接用Java 8或Java 11省心。项目依赖我只列最核心的几项dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version8.1.1/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-clients/artifactId version8.1.1/version /dependency如果是纯批处理不需要额外引入flink-batch因为Flink的批处理已经统一到DataStream API里了。这一点和老的Flink 1.x时代不一样那时候Batch API和Streaming API是分开的现在8.1.1这套统一模型是真的好用。2.3 批量读取与计算示例我先用一个最简单的场景读取本机一个包含订单记录的CSV文件统计每个品类的总销售额。文件长这样order_id, category, amount 1001,electronics,299.00 1002,books,59.00 1003,electronics,1299.00 1004,home,89.00代码核心如下StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setRuntimeMode(RuntimeExecutionMode.BATCH); DataStreamSourceString source env.readTextFile(/data/orders.csv); SingleOutputStreamOperatorTuple2String, Double categoryAmount source .map(line - { String[] fields line.split(,); if (fields.length 3) { return null; } return Tuple2.of(fields[1], Double.parseDouble(fields[2])); }) .filter(t - t ! null); categoryAmount .keyBy(t - t.f0) .reduce((t1, t2) - Tuple2.of(t1.f0, t1.f1 t2.f1)) .map(t - 品类: t.f0 销售额: t.f1) .print(); env.execute(batch-category-sales);注意三件事setRuntimeMode(RuntimeExecutionMode.BATCH)告诉Flink这是批任务底层会使用批式调度和数据交换策略更节省资源。readTextFile不支持从压缩包直接读如果文件是gz格式要改成readFile配合CompressedInputStream或者先解压再读。filter(t - t ! null)不能少因为我见过文件末尾有换行符导致解析出空数组的情况不加filter任务不会报错但结果会多出一条脏数据排查起来挺浪费时间的。跑完这个任务你会看到类似下面的输出品类: books 销售额: 59.0 品类: electronics 销售额: 1598.0 品类: home 销售额: 89.02.4 批量处理的资源调优心得很多人以为批处理就是把env.execute()跑完就没事了。其实8.1.1里批任务的并行度、网络缓冲、状态后端这些都要稍微想一下。我习惯这样配置并行度如果数据量在GB级别并行度设置为CPU核心数的2-3倍不要贪多。并行度太大容易导致shuffle时网络瓶颈太小又浪费资源。状态后端批任务用HashMapStateBackend就够不需要RocksDBStateBackend因为批处理状态生命周期短不涉及大状态增量检查点。网络缓冲如果任务出现“频繁Full GC但堆内存还有很多”这种诡异情况不妨调大taskmanager.memory.network.max给网络缓冲多留一点空间。这些经验都是我反复调优总结出来的不一定适合所有场景但至少能让你在刚开始跑批任务时少走弯路。3. 实时流处理数据实操从Kafka到实时指标3.1 实时流场景需求拆解实时流处理和批量处理的本质区别在于数据是持续不断到达的任务一旦启动就不会自己停结果要随着时间不断更新。这个项目里我模拟了一个电商实时订单流业务方要求实时统计每分钟内每个品类的订单数和GMV延迟要求是5秒以内。从技术选型上上游数据我用了Kafka因为这是目前最主流的接入方式。Flink作为消费者从Kafka的topic里读取订单数据然后做窗口计算、聚合、输出。3.2 创建Kafka数据源和Flink消费端先准备一个简单的Kafka生产者模拟不断产生订单消息。生产者的代码我用纯Java写每秒随机生成1-3条订单Properties props new Properties(); props.put(bootstrap.servers, localhost:9092); props.put(key.serializer, org.apache.kafka.common.serialization.StringSerializer); props.put(value.serializer, org.apache.kafka.common.serialization.StringSerializer); KafkaProducerString, String producer new KafkaProducer(props); Random random new Random(); String[] categories {electronics, books, home, clothing}; while (true) { String category categories[random.nextInt(categories.length)]; double amount 10 random.nextDouble() * 2000; String message category , String.format(%.2f, amount); producer.send(new ProducerRecord(order-topic, message)); Thread.sleep(random.nextInt(500) 200); }Flink端消费代码如下Properties kafkaProps new Properties(); kafkaProps.setProperty(bootstrap.servers, localhost:9092); kafkaProps.setProperty(group.id, flink-order-group); DataStreamSourceString stream env.addSource( new FlinkKafkaConsumer(order-topic, new SimpleStringSchema(), kafkaProps) );这里有个非常重要的点FlinkKafkaConsumer是带状态的它会记录消费到的Kafka offset。如果你任务重启并且没有启用检查点那它可能会重复消费或者从最新位置开始。所以生产环境一定要开启Checkpoint这个后面我专门讲。3.3 实时流处理窗口计算与代码过程实时流的核心是窗口。我用的是TumblingEventTimeWindows也就是滚动事件时间窗口。先定义订单实体类再设定事件时间水印然后做分类聚合。先看事件时间设置SingleOutputStreamOperatorOrder orderStream stream .map(line - { String[] fields line.split(,); return new Order(fields[0], Double.parseDouble(fields[1]), System.currentTimeMillis()); }) .assignTimestampsAndWatermarks( WatermarkStrategy.OrderforBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((order, ts) - order.getEventTime()) );forBoundedOutOfOrderness(Duration.ofSeconds(5))是允许事件时间乱序最多5秒。也就是说一个事件比窗口结束时间晚到但不超过5秒依然可以进入窗口。这里要解释下为什么是5秒而不是0秒真实业务中客户端上报时间往往和服务器接受时间有偏差移动端网络抖动也很常见。如果设置成0容忍你会发现大量数据被丢到迟到侧输出指标和业务方对不上。窗口计算代码orderStream .keyBy(Order::getCategory) .window(TumblingEventTimeWindows.of(Time.minutes(1))) .aggregate(new AggregateFunctionOrder, Tuple2Long, Double, String() { Override public Tuple2Long, Double createAccumulator() { return Tuple2.of(0L, 0.0D); } Override public Tuple2Long, Double add(Order value, Tuple2Long, Double acc) { return Tuple2.of(acc.f0 1, acc.f1 value.getAmount()); } Override public String getResult(Tuple2Long, Double acc) { return 订单数: acc.f0 , GMV: String.format(%.2f, acc.f1); } Override public Tuple2Long, Double merge(Tuple2Long, Double a, Tuple2Long, Double b) { return Tuple2.of(a.f0 b.f0, a.f1 b.f1); } }) .print();aggregate比reduce灵活的地方在于AggregateFunction可以维护更复杂的累加器状态比如同时累计订单数和金额。用reduce的话你得自己包一个自定义对象会麻烦不少。窗口设置成1分钟配合水印延迟5秒实际输出的窗口结果不会在窗口结束的那一瞬间立即打印而是会等到水印越过窗口结束时间后再触发。换句话说你看到结果的时间比业务上的窗口结束时间晚几秒。这个延迟是正常的不需要排查。3.4 状态管理和检查点配置实时流任务里状态管理和检查点是最重要的部分。我的习惯是任务一启动就开启Checkpoint配置如下CheckpointConfig checkpointConfig env.getCheckpointConfig(); checkpointConfig.setCheckpointStorage(hdfs:///flink-checkpoints); checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); checkpointConfig.setMinPauseBetweenCheckpoints(500); checkpointConfig.setCheckpointTimeout(60000); checkpointConfig.setMaxConcurrentCheckpoints(1); env.enableCheckpointing(5000);解释几个关键参数setCheckpointingMode(EXACTLY_ONCE)保证端到端精确一次语义。前提是下游也支持事务性写入比如Kafka sink、JDBC sink如果不支持那只能做到至少一次别被“精确一次”这个词骗了。setMinPauseBetweenCheckpoints(500)两个检查点之间至少间隔500毫秒防止频繁检查点影响吞吐。你想想如果一个窗口1分钟才输出一次结果检查点却每200毫秒打一次那除了浪费资源没别的意义。setCheckpointTimeout(60000)单个检查点如果60秒还没完成就取消它。这个参数主要用在状态特别大的任务上如果检查点总是超时你要么调大超时时间要么优化状态大小。状态后端的选型上如果订单流的状态量在几十GB以下用HashMapStateBackend就够了性能好、延迟低。如果状态到了百GB甚至TB级别那就必须上RocksDBStateBackend它利用磁盘存储状态虽然每轮访问有序列化开销但能撑住更大规模。注意RocksDBStateBackend需要额外引入依赖不是Flink核心自带。3.5 实时流项目中的反压问题反压是实时流任务里最常出现的性能杀手。症状就是Kafka消费速度越来越慢任务延迟越来越大但CPU和内存看起来都还有富余。很多经验不足的同学这时候第一反应是加并行度结果加了并行度反而更慢因为问题不在计算能力而在下游或网络。排查反压的基本方法打开Flink Web UI看每个算子的inPoolUsage和outPoolUsage数值接近1就说明有反压。从最下游算子往回找找到第一个出现高反压的算子通常就是瓶颈所在。如果瓶颈是聚合算子看是不是key分布倾斜导致某个子任务积压了很多数据。如果瓶颈是Sink看下外部系统写入延迟比如Kafka或ES是否打满。我在这个项目里就遇到过一个有意思的情况上游Kafka生产者每秒才发几条数据按理说Flink毫无压力但任务运行10分钟后延迟越来越大。后来看日志发现是因为我没有给SimpleStringSchema指定字符集默认用了平台默认编码在有些环境下处理特殊字符时耗了大量时间做编码转换。这个坑很小但排查起来相当费劲。4. 从批处理体系到实时流体系的融合实践4.1 批流一体统一API的价值我之所以把批量处理和实时流处理放在同一个标题里是因为Flink 8.1.1的批流一体做得确实不错。在8.1.1中DataStream API既能跑有界流也能跑无界流。有界流就是批无界流就是实时。同一个逻辑比如keyBy、window、aggregate两种模式下代码几乎一致区别只是setRuntimeMode和有界无界的数据源。这带来一个巨大的好处你可以先在本地用批量数据调试逻辑验证无误之后再把数据源换成Kafka秒变实时任务。不用担心批量模式能跑、流模式就挂了因为执行计划大部分是共享的。我实际开发中就是这么干的先造一批历史数据文件把窗口和水印逻辑调好再切成Kafka源。4.2 将批量逻辑改造成流处理逻辑的步骤从一个批量任务切换成实时任务通常涉及以下改动数据源从readTextFile换成FlinkKafkaConsumer。如果原来是用execute()之前的数据集直接聚合现在要加上窗口和水印。为keyBy之后的算子设计好状态策略避免无限增长。输出目标从print()改成写入数据库或Kafka。这里最容易漏的是水印。批量模式下没有水印的概念因为所有数据都在那哪有什么迟到不迟到。但流模式下如果没有水印窗口永远不会触发因为Flink不知道事件时间走到哪了。我见过有人从批切流结果窗口半天不出结果就是因为忘了加assignTimestampsAndWatermarks。另一个容易漏的是状态过期时间。实时任务无界如果对某个key聚合一次后状态就一直保留时间长了状态会无限膨胀。所以用keyBy做聚合时最好给状态配置TTLStateTtlConfig ttlConfig StateTtlConfig .newBuilder(Time.hours(2)) .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) .build();TTL设置要根据业务需求定比如热数据时效性只有2小时那状态超过2小时就可以让Flink异步清理掉。注意Flink不会精确到秒级清理TTL状态它只在状态被访问或者后台定期清理线程触发时才会真正删掉。4.3 双链路任务如何共用一套代码我是这样设计目录结构的一个Maven项目里面按batch和stream分两个包但核心的聚合逻辑封装成一个公共类这样batch和stream只是数据源和输出不同计算逻辑不变。比如我定义一个SalesAggregator里面有静态方法public static SingleOutputStreamOperatorString aggregate(KeyedStreamOrder, String keyedStream) { return keyedStream .window(TumblingEventTimeWindows.of(Time.minutes(1))) .aggregate(new SalesAggregateFunction()); }批量任务里我用批量数据源构造出KeyedStream调SalesAggregator.aggregate然后输出到文件实时任务里用Kafka源构造KeyedStream调同一个方法输出到Kafka或数据库。公共逻辑只写一遍后面维护起来真的很省心。还有一个细节批量任务里的Time.minutes(1)和实时任务里的Time.minutes(1)语义完全一样吗不完全一样。批量模式下窗口时间由数据里的eventTime字段确定实时模式下窗口时间同样由eventTime字段确定但因为有水印机制窗口的输出时间会晚于事件时间。如果业务方要求“必须准点出结果”那要考虑用ProcessingTime窗口但这会导致延迟虽然低结果准确性下降尤其乱序数据多的时候。所以时间语义的选择要提前和业务说清楚不能默认大家都懂。5. 常见问题排查与避坑技巧5.1 并行度设置的正确姿势并行度这个问题刚开始用Flink的人基本都会问。我的建议分三步走第一步看数据源分区数。比如Kafka topic有10个分区你的并行度就别低于10不然有分区空闲没人消费。第二步看窗口/聚合的计算压力。如果是简单过滤并行度设成Kafka分区数的1倍就够如果涉及keyBy后的热点key并行度适当增加到2倍让计算压力分摊开。第三步实测调整。看Web UI的每个子任务处理记录数如果长期不均衡说明key分布有问题或并行度不合理。不要盲目把并行度调得很大。并行度一高算子之间的网络连接数量成倍增加序列化和反序列化开销也随之上升。我有一个很强烈的个人体会并行度从4调到8可能提升50%从8调到16可能只提升5%从16调到32可能反而下降。原因就是网络shuffle的成本超过了并行计算带来的收益。5.2 Kafka offset提交异常Kafka消费者offset提交异常是我最常遇到的第二个问题。症状是任务不断重启每次重启都从最早的位置开始消费导致数据重复计算。原因通常是检查点开启后offset的提交方式变了。Flink在开启Checkpoint后默认把offset提交到检查点里而不是Kafka的__consumer_offsets。这样如果检查点失败offset也不会提交下次重启时恢复检查点里的旧offset这本身是合理的。但如果你在Flink之外再用Kafka自带的Consumer Group去读同一个topic你会发现group的offset一直不动于是产生“Flink没提交offset”的错觉。解决办法在Flink的FlinkKafkaConsumer中设置setStartFromLatest()或者在Kafka工具里手动调整偏移。另外如果你的下游系统不支持事务消息那你开EXACTLY_ONCE也没用实际还是会重复这时业务幂等性就非常重要比如数据库里用唯一键去重。5.3 窗口结果延迟或不触发窗口不触发或延迟触发我总结了四种常见原因水印推进太慢forBoundedOutOfOrderness设得太长比如设了1小时那窗口要等很久才触发。解决办法是设一个业务可接受的乱序容忍度比如5-10秒。事件时间的字段解析错误如果withTimestampAssigner返回的时间戳不是毫秒而是秒那所有数据的时间都会变成1970年附近窗口永远不会触发。没有数据进来窗口只有在收到第一条数据后才会创建。如果某类目一小时内没有订单那这个类目在这一小时里就不会有窗口结果。这是正常的不是bug。checkpint失败导致恢复任务Flink从最近一次成功的检查点恢复如果那段时间的窗口状态已经清掉了可能导致部分结果丢失或延迟。我建议大家在写窗口逻辑时务必加allowedLateness和侧输出。allowedLateness允许窗口结束后的一段时间内迟到的数据还能更新结果侧输出则是把那些实在太晚、进不了窗口的数据单独收集起来方便排查和补偿。OutputTagOrder lateTag new OutputTagOrder(late-orders) {}; SingleOutputStreamOperatorString result orderStream .keyBy(...) .window(...) .allowedLateness(Time.seconds(30)) .sideOutputLateData(lateTag) .aggregate(...); DataStreamOrder lateStream result.getSideOutput(lateTag);迟到的数据从侧输出里读出来以后可以打印成日志也可以直接丢弃但至少要让人知道到底有多少数据迟到了。这样你和业务方对指标时就不会被一句“数据不对啊”搞得措手不及。5.4 序列化性能问题Flink默认的序列化器对Java元组、POJO都很友好但对一些自行定义的类型如果没有无参构造和getter/setter可能无法直接序列化导致作业提交时直接报错。这个时候先把POJO类写好别想着用TypeInformation.of强行绕过。另外如果某个字段是MapString, ListString这种复杂嵌套结构序列化开销会比较高。我在这个项目里把一个事件拆成了多个简单字段并在map阶段就把金额从String转成double避免后续每处理一条都做一次字符串转数字。这种微小的优化在数据量大的时候差别很明显。6. 最后的几条经验总结做Flink资料整理这段时间我自己最大的变化是遇到一个新的数据处理需求不再急着选Spark还是Flink而是先看数据是“已经躺好”还是“源源不断来”。如果两者都有那就直接选Flink把API熟悉一遍并行度、状态、检查点、窗口这几个概念搞清楚后面基本能应对大部分场景。另外我真的建议你在本地把Demo跑通而不是只读文档。就拿我在这个项目里用的ThrowingKafkaProducer来说没有真正跑过一遍你永远不知道它会怎么影响你的任务。遇见有人问“为什么我加了这个没生效”一查基本都是缺少配置或依赖没有引入。这类问题看文档十遍不如自己踩一次坑来得实在。最后说一个实操小技巧如果你在本地用env.execute()跑流任务记得在IDE的配置里加上“-Dlog4j.configurationFilelog4j2.xml”不然控制台日志可能刷到崩溃真正关键的错误信息反而被淹没了。这个小细节我特意放在最后就是因为它看起来太小但实际体验差别真的很大。