基于Flink的实时城市交通监控平台设计与实现

简介：基于Flink的大数据实施城市交通监控平台是一份面向高校学生及大数据初学者的课程设计项目资源，核心使用Apache Flink流处理框架构建实时交通监控系统，覆盖数据采集、窗口计算、事件时间处理、状态管理等关键环节，能够帮助读者理解流计算在交通流量分析、拥堵预测等场景中的落地方式，并展示了从需求到编码的工程化实现路径。压缩包共81个文件，大小约18.64MB，内部以Java和Scala源码、XML配置、properties配置文件以及编译生成的class文件为主，同时包含Maven工程配置与分层模块结构，便于对照工程源码学习Flink项目的组织与依赖管理，也便于快速定位关键代码。目前已有1175人学习下载，适合作为Flink入门实践的参考资料，也可直接用于大学生课程设计。通过该项目，读者可以掌握Flink DataStream API的使用、事件时间与水印机制、窗口聚合操作，并学习如何将大数据技术应用于城市交通监控系统的整体架构设计，对提升实时数据处理能力有明显价值。

1. 基于Flink的大数据实施城市交通监控平台是什么：一个能放进简历的实时项目

当你拿到一个名为“基于Flink的大数据实施城市交通监控平台.zip”的项目包时，最忌讳的是直接解压去找代码。这个标题其实已经说明白了：它是一个用Flink做实时计算、覆盖数据采集到可视化大屏的完整工程，常见于大数据毕业设计、招投标方案或者个人实战项目。它的核心价值在于让你用一条真实业务流把Kafka、Flink、MySQL、Redis、ECharts串起来。解决什么问题？每5分钟统计一个路口的车流量，每1分钟刷新一次路段的平均车速，并给交通拥堵等级打分。适合谁？刚学完Flink基础却不知道怎么组装项目的学生，或者想拿实时计算项目去面试的开发者。把这条链路跑通，你对窗口、水位线、状态和背压的理解会扎实很多。

2. 拆解城市交通监控平台：四层架构与Flink选型理由

2.1 数据源层：卡口过车与GPS轨迹如何进Kafka

交通监控的数据源分两类：一类是固定卡口的过车记录，包含车牌、经过时间、路口ID、车道号；另一类是车载GPS上报的轨迹点，包含车辆ID、时间戳、经纬度、瞬时速度。真实的项目里还有地磁检测器和视频识别，但毕设或demo阶段用这两类就够。

常见做法是用一个模拟数据生成器每秒往Kafka写消息。Kafka在这里不是可有可无的中间件，而是整个平台的数据总线。为什么不让Flink直接读文件？因为实时监控拼的是“持续到达、乱序、延迟”的语义，Kafka天然支持分区并行、消息回放和削峰。卡口设备断网后重新连接时，积压的数据会被Flink从Kafka拉回继续计算，这是文件模拟做不到的。

Topic设计上，我一般分两个：traffic-gps存GPS轨迹，traffic-pass存卡口过车。分区数不要拍脑袋设1，因为Flink算子的并行度上限受制于Kafka分区数。比如你的并行度打算设为4，topic分区就至少4。副本数在单机开发环境设1，生产环境设3。消息体用JSON或简单CSV。

生成器代码不复杂，核心是循环发送。示例：

import json import time import random from kafka import KafkaProducer producer = KafkaProducer( bootstrap_servers='localhost:9092', value_serializer=lambda v: json.dumps(v).encode('utf-8') ) roads = ['RD001', 'RD002', 'RD003', 'RD004'] while True: msg = { 'road_id': random.choice(roads), 'event_time': int(time.time() * 1000), 'speed': round(random.uniform(5, 80), 1), 'plate': '川A' + str(random.randint(10000, 99999)) } producer.send('traffic-gps', value=msg) time.sleep(0.5) # 每500ms一条

这段代码把事件时间戳设为发送时的毫秒值，车速在5到80公里/小时之间随机。注意time.sleep(0.5)决定了数据注入速率，你可以改成用循环次数控制每秒多少条。调试时我会把速率压到每秒2条，方便观察窗口输出。

生产者参数里需要关注acks和linger_ms。默认acks可能是1，对于监控平台建议在测试环境保持默认即可，不需要等所有副本确认。linger_ms设成5ms可以攒批发送，吞吐更高，但会增加几十毫秒延迟。交通监控对延迟的容忍度在秒级，所以不用太过纠结。

Kafka的auto.create.topics.enable默认是true，意味着你往一个不存在的topic发消息时会自动创建。这在调试阶段很省事，但生产环境必须关掉，否则拼错一个topic名，消息就进了无人消费的暗沟，事后排查很难。

2.2 实时计算层：窗口、水位、状态这三样必须调

Flink在这个平台里的角色是把无界的数据流按业务维度切成一段段有边界的统计。最核心的是窗口和水位线。为什么不用处理时间？卡口设备从抓拍到上传可能要经过几秒甚至几十秒的网络传输，处理时间会让数据落在错误的统计窗口里。比如8:00:30产生的数据，到8:00:31才被处理，处理时间窗口8:00:30之前就收不到它。所以必须用事件时间，即数据里自带的event_time。

水位线解决“等多久”的问题。假设你要统计8:00到8:05的车流量，但某个路口的数据晚了10秒才到，如果8:05一到立刻把窗口数据输出，那条迟到数据就丢了。常见做法是声明一个“最多允许5秒乱序”的水位线：

WatermarkStrategy.<CarEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((event, ts) -> event.eventTime)

这个5秒指的是：当Flink看到一条事件时间为T的数据时，它会触发所有事件时间小于等于T-5秒的窗口计算。换句话说，窗口会等5秒再出结果，换取对网络抖动和上游延时的容忍。如果你发现结果频繁丢失迟到数据，就把5改成10；如果觉得结果出得太慢，就调小。交通场景我一般设5到10秒。

状态方面，如果你要统计“当前路段车辆数”或“最近30分钟每辆车的轨迹”，就需要用到状态。监控平台里最常见的状态是去重和动态配置。比如车辆通过两个相邻卡口，判断它是直行还是拥堵，就得用KeyedState保存上一卡口的时间。注意给状态设置TTL，默认状态永远不清会把内存撑爆。用StateTtlConfig把保留时间设为窗口长度的2倍即可。

为什么这里选Flink而不是Spark Streaming或Storm？Spark Streaming是微批模型，最小批间隔在百毫秒到秒级，做“秒级刷新”的拥堵指数不够自然；Storm是逐条处理但延迟高，且没有内建的事件时间水位线和精确一次语义。Flink的优势是原生流处理、事件时间支持成熟、checkpoint机制可靠。对于城市交通监控这种“持续运行、数据乱序、结果要落库”的场景，Flink是最不折腾的选择。

窗口类型的选择直接决定结果粒度。我一般按口径区分：

2.3 存储与展示层：指标落到哪张表、刷新用什么粒度

计算完的车流量、平均速度不能只打在日志里，得给后端和可视化用。常见的是双写方案：实时性要求高的拥堵指数写Redis，供大屏秒级刷新；历史统计写MySQL，供报表查询。Redis中的key可以设计成congest:RD001:20260314:1500，value是计算出的等级。MySQL则按窗口粒度落表。

可视化部分一般用ECharts画大屏，后端通过WebSocket或轮询接口读Redis。要注意的是，不要把大屏直接查Flink的结果，因为Flink本身不是为查询设计的。中间一定要隔一层存储，否则每次刷新都会重复消费Kafka末端数据，浪费资源。还有一点：落库的字段不要只有数值，要带上窗口的起始和结束时间戳，否则前端不知道这个指标对应哪个时间段，图表的横轴就对不上。

Redis的key不要设永久TTL，建议设置成和窗口周期相同或稍长，比如滑动窗口1分钟刷新，TTL设90秒。否则历史key越来越多，大屏查到的永远是旧数据。到这里，架构已经立住了。下一节开始动手，先跑通一个最小链路。

3. 手把手跑通最小链路：Kafka到Flink到MySQL

3.1 环境准备：用Docker Compose起Kafka和MySQL

为了快速复现，我建议用Docker Compose把依赖一次性拉起。这里只需要Kafka和MySQL两个容器，ZooKeeper在Kafka 3版本之后可以不再单配，但很多镜像仍会带。一个精简的compose文件可以是这样：

version: '3.8' services: kafka: image: bitnami/kafka:3.6 ports: - "9092:9092" environment: - KAFKA_ENABLE_KRAFT=yes - KAFKA_CFG_NODE_ID=1 - KAFKA_CFG_PROCESS_ROLES=broker,controller - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@kafka:9093 mysql: image: mysql:8.0 environment: - MYSQL_ROOT_PASSWORD=root - MYSQL_DATABASE=traffic ports: - "3306:3306"

注意这里的KAFKA_CFG_LISTENERS和ADVERTISED_LISTENERS必须包含localhost，否则Flink容器或宿主机连不上。如果你在服务器上部署，要把localhost换成服务器IP。另外，MySQL 8.0默认的认证插件是caching_sha2_password，老版本的Flink JDBC驱动可能不兼容，稍后在5.5节会细说。

启动后用docker compose up -d，再执行docker compose ps确认两个容器都是Up状态。然后创建一个测试生产者，往traffic-gps里发几条消息。可以用Kafka自带命令行工具：

docker exec -it kafka-kafka-1 kafka-console-producer.sh --bootstrap-server localhost:9092 --topic traffic-gps

输入一行CSV，例如RD001,1700000000000,35。这里的三列分别是路口ID、毫秒时间戳、车速。这条消息会进入traffic-gps，后面Flink作业就能消费到。

3.2 车流量统计：事件时间窗口的完整Flink作业

开始写第一个作业。目标是每5分钟统计每个路口的过车数量。完整Java代码主体如下：

import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.serialization.SimpleStringSchema; import org.apache.flink.api.common.typeinfo.Types; import org.apache.flink.connector.kafka.source.KafkaSource; import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction; import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows; import org.apache.flink.streaming.api.windowing.time.Time; import org.apache.flink.streaming.api.windowing.windows.TimeWindow; import org.apache.flink.util.Collector; import java.time.Duration; public class RoadFlowJob { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); KafkaSource<String> source = KafkaSource.<String>builder() .setBootstrapServers("localhost:9092") .setTopics("traffic-gps") .setGroupId("road-flow-group") .setStartingOffsets(OffsetsInitializer.latest()) .setValueOnlyDeserializer(new SimpleStringSchema()) .build(); DataStream<String> raw = env.fromSource(source, WatermarkStrategy.noWatermarks(), "kafka-source"); raw.map(line -> { String[] f = line.split(","); return new Tuple3<>(f[0], Long.parseLong(f[1]), Double.parseDouble(f[2])); }) .returns(Types.TUPLE(Types.STRING, Types.LONG, Types.DOUBLE)) .assignTimestampsAndWatermarks( WatermarkStrategy.<Tuple3<String,Long,Double>>forBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((t, ts) -> t.f1) ) .keyBy(t -> t.f0) .window(TumblingEventTimeWindows.of(Time.minutes(5))) .process(new ProcessWindowFunction<Tuple3<String,Long,Double>, String, String, TimeWindow>() { @Override public void process(String key, Context context, Iterable<Tuple3<String,Long,Double>> elements, Collector<String> out) { long count = 0; for (Tuple3<String,Long,Double> ignored : elements) count++; out.collect(key + "," + context.window().getEnd() + "," + count); } }) .print(); env.execute("road-flow-job"); } }

这段代码的逻辑：从Kafka读CSV行，拆成三元组；用事件时间字段做水位线；按路口ID分组；开5分钟滚动窗口；窗口结束时输出“路口ID,窗口结束时间,计数”。其中TumblingEventTimeWindows.of(Time.minutes(5))是固定长度的滚动窗口，每个窗口互不重叠，适合做整点统计。如果你想做“最近5分钟”的实时看板，就要改滑动窗口：SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1))，表示窗口长度5分钟、每1分钟滑一次，这样大屏每60秒能看到一个新结果。

参数上最值得调的是Duration.ofSeconds(5)和Kafka的setStartingOffsets(OffsetsInitializer.latest())。第一次调试用latest()只消费启动后的消息，避免把历史数据全算一遍；如果要回放历史或应对作业重启，改成earliest()。

print()是开发期最简单的sink，它会输出到控制台。但正式环境不能这么干，要改写成落库或写Kafka，下一节会展示写MySQL的配置。

提示：开发阶段不要直接把env.setParallelism(1)挪到生产。并行度1会让KafkaSource只用一个分区消费，吞吐会受限。生产环境应当通过配置指定并行度，并让Kafka分区数大于等于并行度。

3.3 拥堵指数：从平均车速到道路等级的映射

车流量只能告诉你“车多不多”，不能告诉你“堵不堵”。所以第二个作业要算每个时间段内路段的平均车速，再映射成1到3级。1级畅通（≥40km/h）、2级缓行（20~40）、3级拥堵（<20）。

核心代码和刚才类似，只是窗口改成滑动窗口，聚合函数换成求平均速度：

DataStream<Tuple4<String, Long, Double, Integer>> congestion = events .keyBy(t -> t.deviceId) .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1))) .aggregate(new AggregateFunction<Tuple3<String,Long,Double>, Accumulator, Tuple2<String, Double>>() { // 这里省略Accumulator内部实现 }) .map(speed -> { int level = speed.f1 >= 40 ? 1 : speed.f1 >= 20 ? 2 : 3; return Tuple4.of(speed.f0, windowEnd, speed.f1, level); });

上面代码里故意留了省略号，实际开发时不要这样写。你需要自己实现一个AggregateFunction，维护累加车速和数量，在createAccumulator里初始化，在add里累加，在getResult里返回平均值。这个聚合函数的好处是状态只在窗口内累加两个double，比把窗口内所有元素都存下来再遍历要省内存得多。

拥堵等级的阈值不是死的。我在真实项目里会把阈值放到Redis配置里，而不是写死在代码里，这样交警值班员改阈值不需要重启作业。做法是在Flink里用一个MapState保存配置，再通过配置流或广播流更新。新手先不要碰广播流，写死阈值跑通，后面有精力再改。

最后把结果以“路口ID,窗口结束时间,平均车速,等级”的格式发给下一个Sink。下一章我们就把结果真正写进MySQL。

4. 结果存储与后端接口：表设计和MySQL幂等写入

4.1 三张核心表：路口统计、路段明细、预警记录

监控平台通常要存三类数据。第一类是路口/路段的周期统计，这是大屏和报表的底表；第二类是车辆过车明细，用于追溯单个车牌轨迹；第三类是拥堵预警，当等级达到3且持续时间超过阈值时记录一条预警。

建表语句如下：

CREATE TABLE road_flow_stat ( road_id VARCHAR(32) NOT NULL, window_start BIGINT NOT NULL, window_end BIGINT NOT NULL, car_count INT NOT NULL, avg_speed DOUBLE NOT NULL, congestion_level TINYINT NOT NULL, update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (road_id, window_end) ); CREATE TABLE vehicle_pass_detail ( id BIGINT AUTO_INCREMENT PRIMARY KEY, plate VARCHAR(16) NOT NULL, road_id VARCHAR(32) NOT NULL, event_time BIGINT NOT NULL, speed DOUBLE NOT NULL, INDEX idx_plate_time (plate, event_time) ); CREATE TABLE congestion_alert ( alert_id BIGINT AUTO_INCREMENT PRIMARY KEY, road_id VARCHAR(32) NOT NULL, start_time BIGINT NOT NULL, end_time BIGINT, max_level TINYINT NOT NULL, status TINYINT DEFAULT 0 );

注意road_flow_stat的主键设计为(road_id, window_end)，不是自增ID。为什么？因为Flink窗口每次计算结果都要回写同一时间段的统计值。比如同一个窗口因迟到数据被再次触发，或者作业重启后重算调整了数值，你要用INSERT ... ON DUPLICATE KEY UPDATE覆盖旧值，而不是插入第二条。如果主键用自增ID，就做不到幂等。

明细表不需要主键复合，用自增ID就行，因为它是只追加的。但一定要给车牌和时间建索引，否则按车牌查轨迹就会变全表扫描。预警表则要记录起止时间，因为拥堵是持续的状态，不是单点。

4.2 Flink JDBC Sink参数与去重方案

把统计结果写入MySQL，最简单的方案是Flink官方的JDBC connector。先用JdbcSink写upsert语句：

JdbcSink.sink( "INSERT INTO road_flow_stat " + "(road_id, window_end, car_count, avg_speed, congestion_level) " + "VALUES (?, ?, ?, ?, ?) " + "ON DUPLICATE KEY UPDATE car_count=VALUES(car_count), " + "avg_speed=VALUES(avg_speed), congestion_level=VALUES(congestion_level)", (ps, tuple) -> { ps.setString(1, tuple.f0); ps.setLong(2, tuple.f1); ps.setInt(3, tuple.f2); ps.setDouble(4, tuple.f3); ps.setInt(5, tuple.f4); }, JdbcExecutionOptions.builder() .withBatchSize(200) .withBatchIntervalMs(5000) .build(), new JdbcConnectionOptions.JdbcConnectionOptionsBuilder() .withUrl("jdbc:mysql://localhost:3306/traffic") .withUsername("root") .withPassword("root") .withDriverName("com.mysql.cj.jdbc.Driver") .build() )

这段代码里最容易被忽略的是withBatchSize(200)。JDBC Sink默认是按批提交的，但BatchSize太大或间隔太长，一旦Flink作业突然崩溃，最后一批没提交的几百条数据会丢。交通监控的数据量不大，我通常把batchSize设在100到300之间，interval设3000到5000毫秒。如果你要求极致不丢，可以调成withBatchSize(1)，但那样每条都提交，性能会沦为只有2、3条/秒，不足以支撑大规模卡口。

还有一个常见问题是“重复数据”。现象：作业重启后，Kafka offset没有提交成功，于是Flink从最早的offset重新消费，导致同一个窗口被计算两次，MySQL里出现重复记录。解决办法有两个：一是把MySQL主键设为(road_id, window_end)并用ON DUPLICATE KEY UPDATE；二是开启Flink的checkpoint，让Kafka source的位点交由checkpoint管理，重启后从上次checkpoint恢复。俩配合用，才是完整方案。

如果你用的是Flink SQL而不是DataStream API，写法更简单：

CREATE TABLE mysql_sink ( road_id STRING, window_end BIGINT, car_count INT, PRIMARY KEY (road_id, window_end) NOT ENFORCED ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://localhost:3306/traffic', 'table-name' = 'road_flow_stat', 'sink.buffer-flush.max-rows' = '200', 'sink.buffer-flush.interval' = '5s' );

Flink SQL的JDBC连接器自动把主键生成upsert语义，比手写ON DUPLICATE KEY省事得多。唯一的要求是MySQL表必须有主键，且Flink SQL里必须声明PRIMARY KEY ... NOT ENFORCED。

后端接口方面，大屏要的不是复杂查询，而是几个聚合接口。比如：

SELECT road_id, window_end, congestion_level FROM road_flow_stat WHERE window_end = (SELECT MAX(window_end) FROM road_flow_stat) ORDER BY congestion_level DESC;

这个SQL能拿到最新一轮所有路口的拥堵排名。高并发下不要直接查MySQL，把这份结果缓存到Redis，TTL设成和窗口刷新周期一致，能省很多数据库压力。

5. 平台能跑还不够：5个高频避坑点与排障思路

5.1 窗口到期了却不输出任何结果

现象：启动Flink作业后控制台一直为空，Kafka明明有数据。

原因：最常见的是.keyBy(t -> t.f0)用的字段不对，数据里没有那条路径；或者时间戳单位错了，把秒当毫秒，导致水位线永远落后，窗口永远不触发。比如某条消息的event_time是1700000000，秒级的，但代码用Long.parseLong(f[1])当成毫秒，实际时间会变成1970年，水位线也停在1970年，自然永远不会输出当前窗口。

解决：排查很简单，在map之后先print()一条原始时间，看时间戳是不是13位。如果不是，乘以1000转换。另外，检查Kafka topic是否确实有数据，用console consumer看一眼。

5.2 数据总是晚到，窗口结果缺斤少两

现象：统计出来的车流量比实际路口监控少很多，尤其在高峰期。

原因：水位线设置的乱序等待时间小于网络实际延迟。比如你设的是5秒，但卡口系统从抓拍、上传到Kafka经历8秒，这些数据会被丢进迟到通道，默认不参与窗口计算。

解决：先把forBoundedOutOfOrderness从5秒改成10秒，看结果是否明显上升。如果还丢，就要用allowedLateness让窗口再等一段时间。但要注意，延长等待会让结果晚发布，大屏实时性变差。交通场景一般用“先出一个准实时结果，迟到数据来了再更新”的策略，MySQL幂等写入正好配合这个思路。

怎么判断是不是水位线问题？打开Flink UI，看Watermark列的时间戳。如果它和当前时间差超过了预期，说明乱序等待时间或上游延迟有问题。水位线并不是一直增长的，Kafka没有新消息时它也会停滞，这和水位线没推进是两个概念。

5.3 作业重启后MySQL里出现重复行

现象：手动重启Flink作业后，同一road_id和window_end出现多条记录。

原因：Kafka offset的提交时机和checkpoint没对应上。如果没开启checkpoint，作业重启后从latest()或earliest()重新消费，重复计算不可避免。

解决：开启checkpoint并配置为RETAIN_ON_CANCELLATION。然后在JDBC Sink中保留ON DUPLICATE KEY UPDATE。双保险下重复行基本根除。注意开启checkpoint后，KafkaSource的初始offset会被忽略，改从checkpoint保存的位点恢复，这是正常行为。

验证重复行的SQL：SELECT road_id, window_end, COUNT(*) FROM road_flow_stat GROUP BY road_id, window_end HAVING COUNT(*) > 1;如果结果非空，说明幂等没生效。

5.4 背压导致结果延迟越来越大

现象：Flink UI的Backpressure显示HIGH，事件时间处理和迟到数据越来越多。

原因：多数情况不是Flink本身慢，而是下游sink慢了。JDBC写入时如果每条都提交事务，或者batchSize太小，背压会从MySQL倒灌到Flink。也可能是MySQL连接数不够，Flink每个并行子任务建一个连接，并行度8就占8个连接，默认连接池不够就排队。

解决：先看MySQL的show processlist，确认是否有大量Sleep连接。然后把JdbcExecutionOptions的batchSize调到200以上，interval调到3秒以上。如果MySQL连接数受限，考虑换用JdbcCatalog配合连接池，或者在sink前面加一个map缓冲。

还有一种背压来自KafkaSource反序列化慢，比如使用SimpleStringSchema没问题，但如果是自定义解析JSON，建议用内置JSON解析器，而不是在map里做高开销的正则。我见过有人用正则解析CSV，数据量一大就背压告警，换成split后快了10倍。

5.5 JDBC驱动版本和认证插件不匹配

现象：启动sink时报Unable to load authentication plugin 'caching_sha2_password'或Communications link failure。

原因：MySQL 8.0默认使用caching_sha2_password，而项目里引用的MySQL Connector/J版本太老，不认这个插件。

解决：把驱动版本升级到8.0.x，并在pom.xml中把com.mysql.jdbc.Driver改成com.mysql.cj.jdbc.Driver。这是个人人都可能遇到的坑，Flink 1.13之后官方文档示例都是新驱动名。不要为了省事去改MySQL的认证插件，那会降低安全性。

如果你用的是Flink 1.15以下版本，要注意JDBC连接器是旧的flink-jdbc包，部分版本不支持ON DUPLICATE KEY语法。直接升级到新版本最省心。

6. 进阶：从能跑到跑稳，状态后端与Checkpoint的调优技巧

平台已经能出数据了，但要让它在生产环境长时间运行，就要认真对待状态后端和Checkpoint。

状态后端有两个选择：HashMap和RocksDB。默认是HashMap，它快但把状态全塞内存。交通监控平台如果保存了车辆轨迹状态，或者用滑动窗口统计，状态量会随路口数量线性增长。当单台TaskManager超过5GB状态时，HashMap很容易OOM，GC也会频繁发生。我一般推荐RocksDB，它把状态落盘，内存只留缓存，牺牲一点吞吐换稳定性。切换方式是：env.setStateBackend(new EmbeddedRocksDBStateBackend())，并在配置文件中开启增量checkpoint。

Checkpoint周期默认很大，生产上建议设10到30秒。周期太短会频繁快照，影响主链路；太长则故障恢复时丢失的数据多。同时配置minPauseBetweenCheckpoints为几秒，避免两个checkpoint“打架”。一套我常用的配置：

execution.checkpointing.interval: 30s execution.checkpointing.min-pause: 10s execution.checkpointing.tolerable-failed-checkpoints: 3 state.backend: rocksdb state.checkpoints.dir: hdfs:///flink-checkpoints execution.savepoint.dir: hdfs:///flink-savepoints

最后说验证。很多人跑通Flink作业就以为完了，其实要验证结果正确性。我常做的是把同样的原始数据离线算一遍，再跟Flink结果对比。方法：用Kafka的replay能力，把一段时间的数据发到另一个topic，Flink从earliest()消费，同时用一条SQL做批量统计，两边对账。误差在允许范围内才敢上线。

另外一个习惯是给每个窗口结果附带一个“数据时间范围”字段，而不是只给数值。这样就算某个窗口因为迟到数据被重新计算，前端也能通过主键更新覆盖，不会出现曲线突然掉下去。这也是为什么我坚持在road_flow_stat表里同时存window_start和window_end，不要只存一个。

这半年我经手过的交通类实时项目，最后翻车的都不是Flink逻辑，而是那些看似不起眼的配置：时区、时间戳单位、状态TTL、JDBC连接。希望帮到你。

本文还有配套的精品资源，点击获取