企业官网建设流程全解析

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.6，Flink版本8.1.1对应的Scala版本是2.12。这里有个坑：如果你用的Java版本太高，比如Java 17，某些Flink老版本会有模块访问限制，跑本地任务时偶尔报IllegalAccessError。所以建议直接用Java 8或Java 11，省心。

项目依赖我只列最核心的几项：

<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>8.1.1</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients</artifactId> <version>8.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); DataStreamSource<String> source = env.readTextFile("/data/orders.csv"); SingleOutputStreamOperator<Tuple2<String, 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.0

2.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"); KafkaProducer<String, 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"); DataStreamSource<String> stream = env.addSource( new FlinkKafkaConsumer<>("order-topic", new SimpleStringSchema(), kafkaProps) );

这里有个非常重要的点：FlinkKafkaConsumer是带状态的，它会记录消费到的Kafka offset。如果你任务重启，并且没有启用检查点，那它可能会重复消费或者从最新位置开始。所以生产环境一定要开启Checkpoint，这个后面我专门讲。

3.3 实时流处理窗口计算与代码过程

实时流的核心是窗口。我用的是TumblingEventTimeWindows，也就是滚动事件时间窗口。先定义订单实体类，再设定事件时间水印，然后做分类聚合。

先看事件时间设置：

SingleOutputStreamOperator<Order> orderStream = stream .map(line -> { String[] fields = line.split(","); return new Order(fields[0], Double.parseDouble(fields[1]), System.currentTimeMillis()); }) .assignTimestampsAndWatermarks( WatermarkStrategy.<Order>forBoundedOutOfOrderness(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 AggregateFunction<Order, Tuple2<Long, Double>, String>() { @Override public Tuple2<Long, Double> createAccumulator() { return Tuple2.of(0L, 0.0D); } @Override public Tuple2<Long, Double> add(Order value, Tuple2<Long, Double> acc) { return Tuple2.of(acc.f0 + 1, acc.f1 + value.getAmount()); } @Override public String getResult(Tuple2<Long, Double> acc) { return "订单数: " + acc.f0 + ", GMV: " + String.format("%.2f", acc.f1); } @Override public Tuple2<Long, Double> merge(Tuple2<Long, Double> a, Tuple2<Long, 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 将批量逻辑改造成流处理逻辑的步骤

从一个批量任务切换成实时任务，通常涉及以下改动：

1. 数据源从readTextFile换成FlinkKafkaConsumer。
2. 如果原来是用execute()之前的数据集直接聚合，现在要加上窗口和水印。
3. 为keyBy之后的算子设计好状态策略，避免无限增长。
4. 输出目标从print()改成写入数据库或Kafka。

这里最容易漏的是水印。批量模式下没有水印的概念，因为所有数据都在那，哪有什么迟到不迟到。但流模式下如果没有水印，窗口永远不会触发，因为Flink不知道事件时间走到哪了。我见过有人从批切流，结果窗口半天不出结果，就是因为忘了加assignTimestampsAndWatermarks。

另一个容易漏的是状态过期时间。实时任务无界，如果对某个key聚合一次后状态就一直保留，时间长了状态会无限膨胀。所以用keyBy做聚合时，最好给状态配置TTL：

StateTtlConfig 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 SingleOutputStreamOperator<String> aggregate(KeyedStream<Order, 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允许窗口结束后的一段时间内，迟到的数据还能更新结果；侧输出则是把那些实在太晚、进不了窗口的数据单独收集起来，方便排查和补偿。

OutputTag<Order> lateTag = new OutputTag<Order>("late-orders") {}; SingleOutputStreamOperator<String> result = orderStream .keyBy(...) .window(...) .allowedLateness(Time.seconds(30)) .sideOutputLateData(lateTag) .aggregate(...); DataStream<Order> lateStream = result.getSideOutput(lateTag);

迟到的数据从侧输出里读出来以后，可以打印成日志，也可以直接丢弃，但至少要让人知道到底有多少数据迟到了。这样你和业务方对指标时，就不会被一句“数据不对啊”搞得措手不及。

5.4 序列化性能问题

Flink默认的序列化器对Java元组、POJO都很友好，但对一些自行定义的类型，如果没有无参构造和getter/setter，可能无法直接序列化，导致作业提交时直接报错。这个时候先把POJO类写好，别想着用TypeInformation.of强行绕过。

另外，如果某个字段是Map<String, List<String>>这种复杂嵌套结构，序列化开销会比较高。我在这个项目里把一个事件拆成了多个简单字段，并在map阶段就把金额从String转成double，避免后续每处理一条都做一次字符串转数字。这种微小的优化在数据量大的时候差别很明显。

6. 最后的几条经验总结

做Flink资料整理这段时间，我自己最大的变化是：遇到一个新的数据处理需求，不再急着选Spark还是Flink，而是先看数据是“已经躺好”还是“源源不断来”。如果两者都有，那就直接选Flink，把API熟悉一遍，并行度、状态、检查点、窗口这几个概念搞清楚，后面基本能应对大部分场景。

另外，我真的建议你在本地把Demo跑通，而不是只读文档。就拿我在这个项目里用的ThrowingKafkaProducer来说，没有真正跑过一遍，你永远不知道它会怎么影响你的任务。遇见有人问“为什么我加了这个没生效”，一查基本都是缺少配置或依赖没有引入。这类问题，看文档十遍不如自己踩一次坑来得实在。

最后说一个实操小技巧：如果你在本地用env.execute()跑流任务，记得在IDE的配置里加上“-Dlog4j.configurationFile=log4j2.xml”，不然控制台日志可能刷到崩溃，真正关键的错误信息反而被淹没了。这个小细节我特意放在最后，就是因为它看起来太小，但实际体验差别真的很大。