今年手上正好做了一个城市交通实时监控平台，从需求梳理到上线维护前后折腾了快三个月。这个项目最大的感受是：它不是单纯把数据扔给Flink跑几个SQL就完事，而是从数据接入、窗口计算、状态管理、结果落地到可视化展示一整条链路都要打通，每一环都有坑等着你。
这个平台最终要做的事情，简单说就是三件：实时接入城市主要路口的卡口数据、浮动车GPS数据，计算各路段的平均车速和拥堵指数，再把结果推给大屏和管理后台。数据量不大，峰值也就每秒两三千条，但胜在计算逻辑有点意思——要处理迟到数据、要维护路段状态、要跟静态的GIS路网数据做关联，各种边界条件让人头大。如果你正在做类似的实时监控、实时大屏、实时数仓的项目，或者准备用Flink做点实战练手，这篇文章应该能帮你少走一些弯路。
1. 整体设计与技术选型思路
1.1 项目需求拆解：监控平台到底要监控什么
需求方一开始提得很笼统：领导要看城市交通的实时态势。这句话翻译成技术语言，需要拆出四块具体能力：
- 车辆实时位置与轨迹：通过浮动车GPS数据，在地图上实时描点，看车辆走位。
- 路段平均速度与拥堵指数：这是核心中的核心。按路段（link）维度，统计过去5分钟内通过车辆的平均速度，再换算成拥堵等级（畅通、缓行、拥堵、严重拥堵）。
- 异常事件告警：比如某路段速度骤降、某路口车流长时间停滞，需要秒级触发告警。
- 历史回放与趋势对比：虽然叫实时平台，但值班人员和领导免不了要回看过去一小时、昨天同一时段的路况。
拆完需求就清楚了：这是一个典型的实时流式计算场景，数据有GPS坐标、有卡口过车记录，还要叠加上路段静态信息做维表关联。Flink在这个场景里几乎是标准答案，后面细说。
1.2 技术栈选型：为什么是Flink而不是Spark Streaming或Kafka Streams
选型时也做过对比，主要纠结过Spark Streaming和Kafka Streams。
Kafka Streams上手轻，但问题在于它本质是一个库而不是计算引擎，复杂的状态管理、窗口计算、精确一次语义都要自己拼装，而且它强绑定Kafka，如果数据源或结果存储要多样化，写起来很别扭。这个项目要做窗口聚合、维表关联、异步IO，Kafka Streams不是不能用，但开发效率低不少。
Spark Streaming（包括后来主推的Structured Streaming）最大的痛点是微批。我们之前做过POC，在窗口切换时会有秒级的数据延迟，而且对于事件时间（event time）的处理，Spark Streaming在当时版本下远不如Flink成熟。监控大屏上要求刷新延迟在5秒以内，微批模式下很难稳定做到。
Flink这边，四个优势正好命中需求：
- 真流式计算，延迟在毫秒到秒级，配合窗口可以达到5秒内的大屏刷新要求。
- 原生的event time + watermark机制，天然处理乱序和迟到数据，这正是GPS数据最常见的窘境。
- 强一致性的状态管理，做去重、做路段状态维护都方便。
- Flink SQL的成熟度足够高，我们的核心计算逻辑80%用SQL完成。
最终架构定下来是：Kafka（数据接入） -> Flink（实时计算） -> Doris（结果存储） -> Web可视化。Doris的选择后面单独说，这里先卖个关子。
1.3 整体架构图与数据流向设计
我习惯在动手前把数据流画清楚，就算不画正式架构图，白板上也要理一遍。这个平台的数据流向分成六层：
- 数据源层：浮动车GPS终端每5秒上报一条定位记录，包含车辆ID、经纬度、时间戳、瞬时速度、方向角。卡口设备抓拍过车记录，包含车牌、卡口ID、通过时间。两类数据都进入Kafka。
- 接入层：Kafka两个topic，分别对应GPS原始数据和卡口过车数据，分区数按8和4设置。
- 计算层：Flink集群，核心作业有三个——位置明细作业（清洗+坐标转换+维表补全）、路段统计作业（窗口聚合+拥堵计算）、事件告警作业（速度骤降检测+停滞检测）。
- 存储层：Doris存聚合结果和明细，Redis跑热数据缓存给大屏直接读，MySQL存GIS路网静态数据。
- 服务层：SpringBoot服务提供HTTP接口，大屏和后台拉数据。
- 展示层：Web大屏（地图+图表）、管理后台、移动端。
这里有个重要的设计决策：GPS原始数据可能每秒上千条，如果全部落到明细表再查，Doris压力太大。所以明细只保留最近1小时，用于轨迹回放，过期定期清理；而聚合结果永久保留。冷热分离，各取所需。
2. 环境搭建与Flink部署细节
2.1 集群规划与资源估算
我们用的是Flink 1.17，部署模式选了Flink Standalone Cluster，在Kubernetes还没完全普及的团队里，Standalone + 高可用是最容易维护的组合。三台机器，16核32G，系统盘加数据盘分开，其中一台跑JobManager并配置HA，另外两台跑TaskManager。
资源估算倒不复杂，按吞吐量反向推。目标支撑峰值5000条/秒，单条JSON解析后不到1KB，加上窗口计算和维表关联，估计单并行度处理能力在800条/秒左右。预留2.5倍缓冲和故障恢复容量，6个并行度足够。实际分配给每个TaskManager 8G内存，其中托管内存（managed memory）留了2G给RocksDB状态后端。
注意：这里说的并行度是计算并行度，不是TaskManager slot数。我习惯给TaskManager配1个slot，减少同一进程内多任务互相干扰的问题，虽然会有额外的JVM开销，但排查问题时极度舒服。
Flink安装配置这一块网上教程一抓一大把，我只说过三点实战中容易翻车的：
- flink-conf.yaml里的taskmanager.memory.process.size要留足，默认值可能触发JVM Metaspace溢出。
- 如果跑窗口聚合，state.backend.type一定要配成rocksdb，别用内存HashMap存大状态，GC会让你哭。
- JobManager HA要配合ZooKeeper，1.17版本甚至可以直接用内置的Kubernetes HA，但传统运维团队还是习惯ZooKeeper，稳定不折腾。
2.2 踩坑实录：Flink是不是一定要依赖HDFS
热词里有一条“flink一定要hdfs”，这个我也纠结过。很多教程在部署Flink时都会顺带搭一套HDFS，原因是checkpoint默认要写到HDFS。但这不代表Flink强制依赖HDFS。
我们当时没有现成的HDFS集群，也不想为了这个项目单独搭一套Hadoop。解决方案是用文件系统做checkpoint，也就是把checkpoint存到本地磁盘或者NFS共享目录。
代价是什么？如果机器本地盘挂了，checkpoint会丢，任务能恢复到什么程度取决于最后一次成功的checkpoint在不在。我们评估后觉得可以接受：checkpoint每30秒一次，挂了最多丢30秒数据，而上游Kafka会留存数据，通过Kafka的offset重置机制可以把这30秒补回来。
如果你有条件上HDFS或S3，最好还是用好一点的存储。但要说Flink一定要HDFS，那是误解。除了checkpoint和高可用相关的存储，Flink运行时本身完全不需要HDFS。
2.3 开发环境与SQL Client调试技巧
开发阶段大部分时间在跟Flink SQL打交道。Flink 1.17自带的SQL Client虽然能用，但交互体验一般。我们组里实际开发用两种方式：关键作业用Java写DataStream API调SQL，快速验证逻辑用DBeaver连接Flink Gateway执行SQL。
这里额外分享一个技巧：在IDE里本地调试Flink SQL作业时，可以启动一个本地Flink MiniCluster，然后通过Flink REST API提交SQL任务。这样既能断点调试UDF，又能验证完整的SQL执行计划。当时就是靠这个方式，把窗口SQL的watermark节奏和join行为调明白的。
3. 核心功能与Flink SQL实现
3.1 车辆位置实时接入与预处理
GPS原始数据的JSON结构大概是这样的：
{ "vehicleId": "沪A12345", "lng": 121.4737, "lat": 31.2304, "speed": 42.5, "angle": 135, "timestamp": "2024-06-15 08:30:12", "sourceType": "gps" }直接把这个JSON交给Flink解析没问题，但要做两件预处理：
第一，坐标转换。原始GPS坐标是WGS84坐标系，地图底图用的是GCJ02（火星坐标系），如果直接叠加，地图上的点会偏移几十米到几百米不等。所以我在Flink里写了个UDF，用标准坐标偏移算法做了转换。
第二，数据清洗。GPS设备偶尔会上报乱码、经纬度为0、车速为负的脏数据。清洗逻辑不复杂：经纬度超出城市边界范围就丢弃，车速小于0或大于200km/h丢弃，时间戳偏离当前时间超过10分钟丢弃。这些判断用SQL的WHERE条件就能完成。
CREATE TEMPORARY VIEW cleaned_gps AS SELECT vehicle_id, convert_coord(lng, lat) AS (gcj_lng, gcj_lat), speed, ts FROM source_kafka_gps WHERE lng BETWEEN 120.8 AND 122.0 AND lat BETWEEN 30.7 AND 31.8 AND speed >= 0 AND speed <= 200 AND ts BETWEEN NOW() - INTERVAL '10' MINUTE AND NOW() + INTERVAL '1' MINUTE;迟到的数据这一层不直接扔掉，而是打上迟到标记再往下游传，给窗口聚合层自己决定要不要参与计算。这样保证统计口径可追溯。
3.2 路段速度统计：5分钟窗口与迟到数据处理
路段速度统计是最核心的作业。逻辑大概是：把车辆GPS点匹配到路段上，每个GPS点估算出“这个车在这条路段上的瞬时速度”，然后对同一路段同一时间窗口内的所有点求平均速度。
这里用到了两个关键的Flink SQL特性：
事件时间与水位线。GPS数据在网络上传输时天然会乱序，只能靠事件时间而不是处理时间来计算。我在建Kafka源表时显式声明watermark：
CREATE TABLE gps_source ( vehicle_id STRING, lng DOUBLE, lat DOUBLE, speed DOUBLE, ts TIMESTAMP(3), WATERMARK FOR ts AS ts - INTERVAL '5' SECOND ) WITH ( 'connector' = 'kafka', 'topic' = 'gps_raw', 'properties.bootstrap.servers' = 'kafka1:9092,kafka2:9092', 'properties.group.id' = 'gps-group', 'scan.startup.mode' = 'latest-offset', 'format' = 'json', 'json.timestamp-format.standard' = 'SQL' );水位线设置为5秒，意味着最多容忍5秒的乱序。实际运行下来，GPS数据的乱序程度基本在2秒以内，这个参数留了余量。
窗口聚合语句长这样：
INSERT INTO doris_sink_link_speed SELECT link_id, TUMBLE_START(ts, INTERVAL '5' MINUTE) AS window_start, TUMBLE_END(ts, INTERVAL '5' MINUTE) AS window_end, AVG(speed) AS avg_speed, COUNT(*) AS sample_count, PERCENTILE_APPROX(speed, 0.9) AS v90 FROM ( SELECT match_link(lng, lat) AS link_id, speed, ts FROM cleaned_gps ) GROUP BY link_id, TUMBLE(ts, INTERVAL '5' MINUTE);这里的match_link是一个自定义函数，作用是根据经纬度匹配到最近的路段。这个函数内部用了一个内存中的空间索引结构，把城市路网划分成网格，先定位到网格再精确匹配，单次匹配耗时在微秒级别。
提示：5分钟窗口和1分钟窗口的实际效果差异很大。5分钟窗口可以平滑掉瞬时波动，适合展示拥堵指数；1分钟窗口保留更多细节，但会有不少空窗和抖动。最终我们做了双链路：一条跑5分钟窗口用于大屏展示，一条跑1分钟窗口用于告警检测。
3.3 拥堵指数计算：从速度到等级的换算逻辑
有了平均速度，怎么转成拥堵等级？这个没有统一标准，我们结合需求方的经验定了一套阈值，同时考虑不同道路等级（快速路、主干路、次干路、支路）的差异：
| 道路类型 | 畅通（>=） | 缓行（>=） | 拥堵（>=） | 严重拥堵（<） |
|---|---|---|---|---|
| 快速路 | 50km/h | 35km/h | 20km/h | 20km/h |
| 主干路 | 30km/h | 20km/h | 10km/h | 10km/h |
| 次干路 | 25km/h | 15km/h | 8km/h | 8km/h |
| 支路 | 20km/h | 12km/h | 6km/h | 6km/h |
这个表不是写死在代码里的，而是从MySQL维表加载，用Flink的维表join实时关联。这样需求方想调阈值，直接在后台改表就行，不用重启作业。
维表关联用LOOKUP JOIN，缓存策略选的是ALL（全量缓存），因为路网数据总共就几十万条，全量加载到内存完全没问题，还能避免查MySQL的性能损耗。
CREATE TEMPORARY VIEW link_with_speed AS SELECT s.link_id, s.window_start, s.avg_speed, d.road_type, d.link_name FROM doris_source_link_speed s LEFT JOIN mysql_dim_link d ON s.link_id = d.link_id;3.4 卡口流量与异常事件告警
告警作业是另一条链路。我们监控两类异常：
第一类，某路段平均速度在连续两个1分钟窗口内下降超过40%。这代表发生了突发拥堵或事故。实现上并不复杂，用MATCH_RECOGNIZE做模式匹配，识别出“快速->更快->骤降”的趋势序列。
第二类，某卡口在5分钟内过车数为0，但历史上这个时段平均过车数大于50。这代表卡口可能故障或者路段已经完全堵死。需要先把历史同期均值数据缓存到Redis，再在Flink作业里通过异步IO访问Redis比对。
第二类涉及外部存储关联，Flink SQL的维表join可以搞定，但延迟会稍高。后来我们改成在数据进入窗口聚合前先做一次open的生效检查，把整个判断逻辑封装在一个UDF里，性能提升明显。
告警结果写入告警表，同时通过WebSocket推给大屏，大屏上弹红色气泡，值班人员可以秒级看到。
3.5 温故知新：数据血缘如何跟踪
热词里提到“flink数据血缘”，这个项目也做了点尝试。实时计算作业多了以后，最痛苦的不是写SQL，而是维护“这张Doris表的数据到底从哪来、经过了哪些计算、依赖哪些原始表”。
Flink 1.17有内置的Lineage机制，可以在Catalog中自动记录作业的输入输出血缘。我们在此基础上做了一层轻量的血缘管理：每个Flink SQL作业在提交时附带一个作业元数据JSON，包含来源topic、目标表、窗口策略、责任人。最终血缘关系通过Doris的元数据表统一查询。
别小看这个动作，后面排查数据问题时省了大量时间。区域A的数据对不上，打开血缘图，一眼定位是该区域的GPS清洗逻辑有问题还是窗口参数配错了。
4. 结果存储与数据服务设计
4.1 为什么选了Doris而不是MySQL或ClickHouse
聚合结果和明细数据需要支撑大屏的高并发点查，同时要支持一些多维度的即席分析。一开始考虑过ClickHouse，但在点查场景和并发更新上不如Doris顺手。MySQL则不太适合存这种按时间分区的、高频写入的流式结果数据。
Doris的优势在于：
- 支持批量导入和高并发点查，正好适配Flink写入+大屏查询的模型。
- 内置的Unique模型支持主键更新，对数据重算和迟到数据修正非常友好。
- 冷热分区和动态分区管理省心，一天一个分区，老数据自动过期。
Doris建表时我们重点考虑了两个点。第一，分区和分桶键要按查询模式来。常用的查询维度是“路段ID + 时间”，所以建表用时间做分区，路段ID做分桶键。第二，Unique模型要设置sequence列处理乱序写入，避免迟到数据把新数据覆盖掉。
CREATE TABLE IF NOT EXISTS link_speed_5min ( link_id BIGINT, window_start DATETIME, avg_speed DOUBLE, sample_count INT, v90 DOUBLE, congest_level INT ) ENGINE=OLAP UNIQUE KEY(`link_id`, `window_start`) DISTRIBUTED BY HASH(`link_id`) BUCKETS 8 PROPERTIES ( "dynamic_partition.enable" = "true", "dynamic_partition.time_unit" = "DAY", "dynamic_partition.end" = "3", "dynamic_partition.start" = "-30", "replication_num" = "1" );4.2 Flink Doris Connector的典型报错与解决办法
在集成Flink和Doris时，我们遇到了一个很典型的报错，也是热词里提到的：“Flink type is DATEV2, but arrow type is DateDay”。这个报错的原因是Flink的Doris Connector在读取Doris表时，列类型映射出现问题。
Doris的DATEV2类型在和Flink的Arrow格式交互时，被映射成了DateDay类型，而Flink内部认为自己读到的是DATEV2，两边对不上就报了异常。解决办法其实简单，两种：
- 升级Connector版本。Doris官方提供的flink-doris-connector新版已经修复了DATEV2的映射问题。
- 建表时不要用DATEV2类型，改用DATE或DATETIME，或确保Connector和Doris版本兼容。
这个报错折腾了我们半天，核心教训是：Doris Connector版本和Doris集群版本要严格匹配，最好用官方推荐配对的版本组合。不要拿个最新版Connector去连旧版Doris，也不要反过来。
4.3 TiDB Flink SQL的兼容性对比
热词里有“tidb flink sql”，因为最开始我们考虑过用TiDB当结果存储。TiDB的Flink SQL Connector也相当成熟，尤其在实时数仓场景下跟Flink配合得很好。它兼容MySQL协议，如果我们团队更熟悉MySQL生态，用TiDB确实上手快。
但对比下来有个关键差异：Doris的Unique模型天然适配“按主键覆盖更新”的写入方式，Flink直接把聚合结果写入即可；而TiDB要走INSERT ... ON DUPLICATE KEY UPDATE或REPLACE INTO来做更新，写入性能在某些场景下会略逊一筹，而且TiDB的批量导入链路没有Doris的stream load那么顺滑。
所以最终选了Doris。如果你的场景偏事务型查询，且需要MySQL高度兼容，TiDB是完全可行的选择；但纯OLAP分析大屏场景，Doris更合适。
4.4 JDBC连接器异常排查经验
项目里还遇到过Flink的JDBC连接器异常。一次作业运行几天后突然报错，错误信息大致是连接超时或连接池耗尽。排查过程分三步：
第一，看是不是连接泄漏。JDBC Connector每个并行实例会持有自己的连接池，如果连接未正常释放，运行久了必然出问题。检查SQL中是否有窗口JOIN或维表JOIN频繁占用连接，并确保lookup.cache配置正确。
第二，看MySQL的wait_timeout。MySQL默认8小时断掉空闲连接，如果Connector没做连接有效性检测，就会拿到一个坏连接。解决办法是调大MySQL超时时间，或在连接串中加上autoReconnect=true。
第三，看并发和连接池配置。Flink作业的并行度乘以每个并行度的连接数，不能超过MySQL的max_connections，否则就是雪崩。
一个更省心的做法：维表数据量不大时，优先用LOOKUP JOIN的ALL缓存，一次性加载全量数据，避免频繁访问MySQL。
5. 常见问题与故障排查实录
5.1 数据延迟越来越大，问题出在哪
上线第一天就遇到一个教训。大屏上的数据延迟由最初的3秒慢慢涨到30秒，最后直接卡住。排查思路：
先看Flink UI，发现某个作业的背压（backpressure）已经变成HIGH。再看CPU和内存，发现某个TaskManager的CPU已经打满，垃圾回收频繁。进一步检查，是维表JOIN的缓存策略配置出了问题——LOOKUP JOIN默认可能不开缓存，每条数据都要查一次MySQL，吞吐上不去。
解决办法：把维表缓存改成ALL，并设置lookup.cache.ttl为1小时。修改之后压力立即降下来。
注意：维表缓存不是越大越好。如果维表数据会更新（比如路网等级调整），TTL太大会导致Flink读到的还是旧数据。建议维度数据更新不频繁时用ALL缓存，频繁更新时用LRU缓存并配置合理的TTL。
5.2 窗口计算结果突然为空的诡异问题
有一次某条路的5分钟统计结果一直是空的，其他路正常。检查Kafka日志发现，那条路的数据其实一直在进。
最后定位到问题出在match_link函数：那条路的坐标落在我划分的空间网格的边缘，网格索引的容差设置太小，导致坐标匹配到相邻网格后找不到匹配路段。
修复方式很简单，把网格匹配的容差从50米调到100米，并且增加一个“找不到最近路段时尝试相邻网格”的兜底逻辑。这种坐标匹配类的边界问题，单测很难覆盖全，最好的办法是线上多留日志，把匹配不到的数据单独落到一个debug表，定期人工检查。
5.3 背压排查与性能调优实录
背压问题在实时计算里几乎躲不掉。我的排查顺序是：
- 看Flink UI的背压指标，确定背压发生在算子之间还是整个作业层面。
- 检查Sink算子的写入性能，Doris批量写入如果攒批参数不合理，很容易成为瓶颈。
sink.batch.size和sink.batch.interval要配合调整。 - 检查窗口算子的状态大小，RocksDB状态太大时，序列化和反序列化开销会明显上升。用
spillable和调整RocksDB的block cache可以缓解。 - 最后才是调整并行度和资源配置。
我们最终通过把Sink的攒批大小调整为每批2000条或每2秒刷一次，Doris写入吞吐提升了一倍，背压问题基本消失。
5.4 数据重复与精确一次语义
一开始用Kafka + Flink + Doris的标准链路时，Doris里偶尔会出现重复数据。原因是Flink的checkpoint开启后，作业重启恢复时会从checkpoint和Kafka的offset重新消费，如果Sink不是幂等的，就会重复写入。
Doris的Unique模型天然支持主键幂等，重复写入相同主键的数据会覆盖，所以这个问题被Doris部分掩盖了。但明细表有些场景没法用Unique模型，于是开了Flink的CheckpointingMode.EXACTLY_ONCE，配合Doris的stream load两阶段提交，确保端到端精确一次。
这里要强调：精确一次不是只开一个开关就完事，上下游必须配合。上游Kafka要支持从offset恢复，中间Flink要开checkpoint，下游Sink要支持事务或幂等写入，缺一个环节，精确一次都是空话。
5.5 一张问题排查速查表
| 现象 | 可能原因 | 排查方法 | 解决办法 |
|---|---|---|---|
| 大屏延迟持续增大 | 维表JOIN缓存未开或配置不当 | 看Flink UI背压指标 | LOOKUP缓存改为ALL或LRU+合理TTL |
| 窗口结果为空 | 数据匹配不到维度 | 查debug表/日志 | 调整匹配容差，增加兜底逻辑 |
| 结果数据偶尔重复 | Sink未做幂等 | 检查主键约束 | 使用Unique模型或开启精确一次 |
| 作业运行几天后报错 | JDBC连接池泄漏/超时 | 查连接数和wait_timeout | 调整连接配置或改用ALL缓存 |
| 数据类型映射报错 | Connector版本不匹配 | 查Doris/Connector版本 | 升级Connector或调整建表类型 |
| 迟到的数据把新数据覆盖 | 窗口未处理迟到数据 | 看watermark设置 | 延迟关闭窗口或设置主键sequence |
6. 项目上线后的几点经验沉淀
最后聊点跟技术无关但跟项目成功有关的体会。这类监控平台，业务方一开始说的需求永远是模糊的，但他们对延迟和准确性的要求却非常具体。建议在动手前，先跟业务方对齐两个核心指标：数据从产生到大屏可见的最长链路时间，以及可容忍的数据误差范围。
链路时间是技术问题，误差范围就有点微妙了。我们当时跟业务方明确过：GPS数据正常情况下的覆盖率在85%左右，如果某个路段样本量太少（比如一分钟内只有两三辆车经过），统计结果可能失真。所以系统对样本量小于5的路段做了特殊标记，在界面上显示为“样本不足”而不是给出一个看起来精确但并不可靠的拥堵指数。这种主动暴露不确定性的做法，反而让业务方更信任系统。
维护期最大的坑是版本兼容。Flink、Doris Connector、Kafka Client这些组件的版本升级一定要谨慎，升级前务必在测试环境完整回归一遍SQL作业，不要轻信“兼容”两个字的官方文档。
最后再分享一个小技巧：上线前把Kafka的topic数据备份一份到本地，用这份数据做回放测试。这样既能验证新版本作业的正确性，又不会影响线上链路。我们靠着这个习惯，两次大版本升级都平稳过渡了。
