简介本资源是一个面向交通大数据分析、城市智能治理及GIS开发者的高德地图API实战项目聚焦全国主要城市自2019年10月8日起的实时与历史交通拥堵指数自动化采集与持久化存储。项目提供完整Java工程结构含4个核心Java类、MySQL建表SQL脚本city_crowd_data.sql、配置文件pom.xml、resources目录、说明文档docx及README.md技术指南共10个文件总大小仅42KB轻量易部署。代码已封装定时调度、API异常重试、数据清洗与批量入库逻辑支持长期稳定运行SQL脚本定义了含时间戳、城市编码、拥堵指数、数据来源等字段的规范化表结构便于后续趋势分析与可视化。目前已有44人学习下载适合具备Java基础与数据库操作能力的开发者快速复用采集框架、构建自有城市交通拥堵历史数据库并延伸应用于拥堵预测、交通态势看板或政策效果评估等场景。 你研究城市交通数据的时候十有八九会撞上同一堵墙打开高德地图开放平台实时路况接口、交通态势接口都在参数填好就能拿到当前全城路况但想往前翻历史数据没了。高德只给实时快照不给回放任何需要历史拥堵指数做分析的项目都得自己从启动那天开始一张张攒。这个项目就是这么来的——从2019年10月8日开始按固定频率定时调用高德地图API接口把全国主要城市的实时与历史交通拥堵指数数据采集下来持久化存储到数据库中逐步搭起一个城市交通拥堵指数历史数据库。本文会把整套方案的接口能力边界、采集设计、数据库建模、调度容灾和实测细节全部拆开讲适合正在做交通数据分析、通勤规律研究、物流调度优化或城市拥堵趋势观测的人直接参考复现。1. 为什么说高德API“给实时不给历史”自建指数库到底解决了什么1.1 高德交通API的真实能力边界能拿到什么、拿不到什么我先说结论高德的交通态势相关API本质上是“当前路况的切片接口”核心能力是给定一个地理矩形范围返回当前时刻这个范围内所有路段的通行状态、平均速度和道路名称。你把请求发出去拿到的是此时此刻的交通态势接口背后没有时间维度参数没有“某年某月某日某时刻的路况”这种查询能力。具体来说高德开放平台提供的主要路况能力是交通态势接口通过构造矩形区域的左下角和右上角经纬度就能拉取该范围内的实时道路交通态势。返回的字段里通常包括道路名称、路段方向、道路状态码畅通、缓行、拥堵、严重拥堵、通行速度、路段长度等。每条路段还有一个状态等级可以直接用来做拥堵评估。这些数据很丰富但也正因如此容易被误解。很多第一次接触的人以为可以一次性拉回整个城市的路况或者以为可以带上时间参数去查历史结果发现接口文档里根本没有“历史时间”这个输入项也没有“导出历史数据”的后台功能。这就是“给实时不给历史”的本质你想做历史分析必须自己在实时数据流经时把它截留下来存进数据库别无他法。1.2 一个“只进不出”的接口为什么还值得围绕它建库既然历史数据只能靠自己积累那第一个问题就是花几个月、几年时间持续采集到底值不值我的判断是只要你的使用场景里有一点“分析过去”的需求就非常值。城市拥堵指数如果只有今天的快照你只能回答“现在哪里堵”有了连续几个月或几年的历史库你能回答的问题就完全不一样了早高峰和晚高峰的拥堵峰值分别出现在几点工作日和周末的差异有多大。重点商圈周边道路的拥堵何时开始、何时结束是否与大型活动、天气变化相关。某条快速路改造前后周边路网的拥堵指数有没有明显变化。打车、外卖、物流配送的调度策略能不能根据一周内不同时段的拥堵规律做预判。我见过很多团队做城市交通分析时最头疼的不是建模、不是画图而是数据来源。政府开放的交通数据覆盖城市有限、字段粒度不一商业数据平台价格不低自己爬非官方来源又面临稳定性和合规风险。相比之下高德开放平台有官方API、结构清晰、字段稳定只要按照合理的频率和规范去调用个人和小团队完全能承担唯一需要付出的就是持续积累的时间成本。1.3 项目的时间窗口与城市范围从2019年10月8日开始的全国主要城市标题里“自2019年10月8日起至今”这个时间点我需要多说一句。很多人会误以为高德API能提供从2019年开始的历史数据实际完全不是。真实情况是API不提供任何历史回溯能力所以这个日期实际是项目的启动时间也就是数据采集的起点。从这天开始系统按固定频率持续运行才有后来覆盖多年的历史时间序列。如果你的项目还没有启动那就从今天开始攒越早开始未来的“历史”越长。城市范围我定义为全国主要城市包含直辖市、省会城市和部分计划单列市用高德的adcode城市编码作为唯一标识。比如北京110000、上海310000、广州440100、深圳440300、成都510100这些编码在后续的数据表设计和API调用中会反复用到。采集初期的城市列表我通过高德行政区划接口自动获取同时人工剔除掉一些偏远地区数据量过少的地级市最终保留了几十个核心城市覆盖全国主要人口和经济区域。2. 采集链路方案设计从“整城路况”到“城市拥堵指数”的核心算法2.1 为什么不能直接拉整个城市接口的矩形范围限制如果你拿着高德交通态势接口文档第一反应大概率是把城市外包矩形作为参数一次把全城路况拉回来。但实际操作中你会发现接口对矩形范围有大小限制一次性传一个覆盖整个城市的大矩形要么请求失败要么返回的路况数据被系统截断很多路段直接缺失。这不是高德故意刁难而是服务端在降低单次计算压力。一个覆盖全城的超大矩形包含的路段数量和坐标点数量巨大普通接口很难在稳定延迟内处理完。合理做法是“分片拉取”把城市拆成一个个小网格依次请求每个网格的路况最后把结果合并拼成整个城市的路况快照。这个策略和地图瓦片加载的逻辑很像。你刷手机地图的时候地图也不是一次性渲染整个屏幕而是按瓦片分层加载每个瓦片只绘制一个小区域拼起来就是一整幅地图。网格化采集路况本质上就是给路况数据做了一层“瓦片化”。2.2 网格粒度如何定一个中等城市需要多少个请求网格粒度是方案里最先要定的参数。网格划得太大局部路况细节会被吞掉而且容易触发接口的范围限制网格划得太小请求数量成倍增加整体扫描一轮的时间会拉长流量成本也飙升。我用的做法是按经度和纬度各0.1度划分网格。0.1度在赤道附近大概对应11公里左右在纬度40度左右大概对应8.5公里这个粒度对于观测城市级拥堵趋势已经足够既能保留主干道和快速路的路段细节又不会让网格数量失控。以北京为例北京城区的经度跨度大约1.6度纬度跨度大约1.2度按0.1度网格切分大概是16乘12也就是192个网格。每个网格请求一次扫完整个北京需要约192次API调用。一次请求加解析大约0.2秒单城一轮扫描大约40秒。这个耗时完全可以接受即便同时管几十个城市也能在几分钟内完成一轮全国扫描。如果你的分析需要更细的道路级路况可以将网格缩小到0.05度但要注意网格数量会变成原来的4倍扫描时间也相应拉长API配额消耗更快需要根据配额情况做权衡。2.3 从道路状态到城市拥堵指数聚合算法与指数映射每个网格返回的数据是“路段级”的有道路名称、状态码、速度、长度。但最终要入库和展示的核心指标是“城市拥堵指数”一个0到10之间的数值所以必须把路段级数据聚合成城市级指数。我采用的聚合逻辑分两步。第一步是加权平均速度把该城市所有路段按道路长度加权算出当前时刻全市路网的平均通行速度。之所以用长度加权是因为长路段对整体路网状态的影响自然比短路段大如果简单平均大量短小的支路会稀释主干路的拥堵信号。第二步是把平均速度映射到0到10的拥堵指数区间。我用的映射方式和大众认知中的“拥堵指数”对齐速度越高指数越低速度越低指数越高。具体映射关系大概是这样平均速度接近甚至超过自由流速时指数在0到2之间属于畅通速度明显下降但还能正常行驶时指数在3到5之间属于缓行速度很低、走走停停时指数在6到8之间属于拥堵几乎停滞时指数在9到10之间属于严重拥堵。为便于观察每个城市在同一时刻可能同时存在不同道路的拥堵状态聚合后得到的就是该城市整体拥堵程度的一个综合评分。除此之外我也会把每个网格内的状态码统计一并保存比如畅通路段占比、拥堵路段占比方便后续做更细粒度的分析。数据入库时同时保存原始JSON快照这样即使后续调整聚合算法也能用历史数据重新计算不会因为算法升级丢掉原始信息。3. 数据库建模与持久化千万条记录怎么存才能既完整又查得快3.1 表结构设计配置表、指标表、日志表、失败任务表数据库选型上我最终用了PostgreSQL。选择它不是因为高德API有什么特殊要求而是这组数据带有明显的时间序列特征PostgreSQL对这类数据的写入、分区、查询支持都很成熟而且相对轻量自己一台小服务器就能跑。如果你更熟MySQL整体方案也可以平移过去核心表结构和索引设计思路是一样的。整个项目里我建了四张表各司其职。第一张是城市配置表存所有采集城市的adcode、城市名、城市外包矩形边界、网格大小和启用状态。后续采集主程序会读取这张表动态决定要扫描哪些城市以及扫描范围新增城市时只改表不改代码。第二张是交通拥堵指数指标表这是核心数据表。每条记录对应一个城市在某个采集时刻的聚合指数字段包括adcode、城市名称、采集时间、拥堵指数值、拥堵等级、全市加权平均速度、参与聚合的路段数量、网格数量以及原始JSON快照。原始快照字段很占空间但对追溯和算法复算很有用我建议保留。第三张是采集日志表记录每一轮采集任务的城市、启动时间、成功网格数、失败网格数、总耗时和状态。这张表的价值在运维后期排查“某个时间段数据为什么缺失”时完全靠它。第四张是失败任务表。凡是请求失败的网格任务ID、城市、网格坐标、错误码、失败原因都会写进去供补采程序读取。3.2 时间分区与索引设计让查询不随数据量膨胀变慢交通指数数据是典型的时序数据。以5分钟一次、50个城市计算一天就是14400条记录一年超过500万条。如果所有数据堆在一张普通表里时间一久查询会明显变慢所以必须在表设计阶段就考虑分区。我的做法是按月份做范围分区每个分区存一个月的指标数据。这样查询某段时间的数据时数据库可以直接定位到对应分区不需要全表扫描。建表时声明分区键是采集时间月份变化时系统可以自动创建新分区也可以每天由调度任务检查并创建下月分区。索引方面最常用查询条件自然是“某个城市在某段时间内的指数序列”所以我在adcode和采集时间上建了联合索引。原始快照字段一般不参与查询条件不用建索引避免写入和存储负担。日志表和失败任务表索引相对简单主要按任务ID和城市编码查询建普通索引就够。3.3 批量写入与连接池别一条一条insert否则数据会越攒越落后刚开始做采集的时候我犯过一个典型错误每条指标数据都单独执行一次insert。看起来逻辑简单但数据量上来后写入速度立刻成为瓶颈采集速度跟不上调度频率后面就开始排队数据越攒越落后。后来改成批量写入用psycopg2的execute_values或者PostgreSQL原生的COPY语义一次性把一批几十条、几百条记录写进去。同样是插入5000条记录逐条插入可能需要几十秒批量插入往往只需要一两秒性能差距非常明显。连接管理也很关键。每轮采集任务开始前建立连接任务结束后统一提交关闭不要在每次insert时都新建连接。多城市并行采集时更要小心连接数量控制不好会把数据库连接池打满。我一般设置为每个采集进程一个长连接任务结束再归还实际运行起来非常稳定。4. 调度、异常与长期稳定性让采集任务连续跑几年不出问题的关键细节4.1 定时调度与采集频率5分钟一次是成本和时效的平衡点采集频率直接影响数据的时间分辨率也直接影响API配额消耗和存储量。我最终定的是每次更新频率为5分钟也就是每5分钟对全国主要城市各做一轮完整扫描。5分钟这个值不是拍脑袋定的。城市拥堵的演变周期通常在10分钟以上5分钟的采样间隔已经能还原拥堵形成和消散的完整过程再加密的话数据量成倍上涨但信息增益非常有限。从调度实现上说5分钟与crontab的分钟精度完美对齐本身就是天然的时间边界不容易累积漂移。调度器我用了系统级crontab定时执行采集主脚本。选择crontab而不是常驻后台进程是因为它足够简单可靠脚本崩溃了下一轮cron会重新拉起不会出现后台进程挂掉之后再也不会自动恢复的情况。采集任务跑几年稳定性一定优先于炫技crontab这种朴素方案恰恰是最稳的。4.2 限流、超时与重试高德API的“软性限制”必须提前应对长期调用高德API接口必然遇到的问题就是限流。高德开放平台对普通开发者有配额限制并发超过一定阈值或单位时间请求过于密集时接口会返回错误码。这个限制是“软性”的不是绝对禁止但如果你不注意调用节奏轻则单次请求失败重则一段时间内被系统限流所有请求都打不出去。我的处理方式有三个层次。第一层是控制并发同一时刻只有一个城市的扫描在跑城市之间串行或限制最大并发数避免瞬时请求量过大。第二层是请求间隔每个网格请求之间加一个小延迟比如0.1到0.2秒这样一轮扫描的时间能够接受也远在限流阈值之下。第三层是重试单次请求失败后不立即重试而是做指数退避第一次等1秒第二次等2秒第三次等4秒最多重试三次。三次仍然失败就记录到失败任务表留给补采程序处理。超时设置也必须显式指定。依赖库默认的超时时间往往太长网络异常时请求会卡住很久拖慢整个任务。我在所有HTTP请求里都设置了5秒连接超时和10秒读取超时超过就直接判失败走重试逻辑绝不傻等。4.3 数据完整性监控与补采机制历史是攒出来的中间缺了会很麻烦数据完整性是这个项目的生命线。实时数据错过一次未来分析的时候那5分钟就是永久空白无法通过API回补。所以必须有两个机制兜底监控和补采。监控的核心是检查采集日志表。在每天固定时间跑一个统计SQL校验每个城市当天应有的采集次数和实际入库次数差异超过阈值就报警。同时检查是否有连续多个时间点完全缺失的情况这种情况往往意味着调度器或网络整体故障光靠补采不一定能解决需要人工介入。补采针对的是“个别网格或个别城市请求失败”的场景。补采程序读取失败任务表按任务ID和城市编码重新发起请求把缺失的数据补写进指标表。补采一般安排在凌晨低峰时段避免和正常采集任务抢占资源和挤占配额。如果失败原因是全市性网络中断凌晨补采时也一并重试能补回多少算多少。5. 直接可复现的最小实现从配置到代码的完整示例5.1 环境准备与依赖整个采集服务就是一组Python脚本加一个PostgreSQL数据库环境要求很低。Python 3.8以上够用依赖库只需要requests、psycopg2-binary、retry这几个。数据库用PostgreSQL 12以上创建好数据库和对应表之后再建一个只读配置表用的普通账号尽量不要用超级管理员跑业务。高德开放平台需要先申请Web服务API Key申请地址在开放平台控制台创建应用后就能拿到key。Key建议放在环境变量或独立配置文件中不要写死在代码仓库里避免泄露后被其他人盗刷配额。5.2 核心采集代码网格扫描加聚合计算先看单城市扫描的核心代码。下面这个函数接收city配置按网格大小把城市矩形切成多个小网格逐格请求交通态势接口将各路段的长度、速度、状态解析出来最后聚合成该城市的拥堵指数。这一层是最核心的实现。import requests import time import math API_KEY your_amap_web_api_key TRAFFIC_URL https://restapi.amap.com/v3/traffic/status/rectangle STATUS_MAP { 1: 畅通, 2: 缓行, 3: 拥堵, 4: 严重拥堵 } def scan_city(city): adcode city[adcode] min_lon, min_lat city[min_lon], city[min_lat] max_lon, max_lat city[max_lon], city[max_lat] grid_size city.get(grid_size, 0.1) roads [] grid_count 0 failed_grids [] lon min_lon while lon max_lon: lat min_lat while lat max_lat: rect f{lon},{lat};{min(lon grid_size, max_lon)},{min(lat grid_size, max_lat)} try: resp requests.get( TRAFFIC_URL, params{key: API_KEY, rectangle: rect, level: 6}, timeout(5, 10) ) data resp.json() if data.get(status) 1: traffic_info data.get(trafficinfo, {}) road_list traffic_info.get(roads, []) for road in road_list: roads.append({ name: road.get(name, ), status: road.get(status, ), speed: float(road.get(speed, 0) or 0), length: float(road.get(length, 0) or 0) }) grid_count 1 else: failed_grids.append({rect: rect, info: data.get(info)}) except Exception as e: failed_grids.append({rect: rect, error: str(e)}) lat grid_size time.sleep(0.15) lon grid_size return aggregate_city_index(adcode, roads), grid_count, failed_grids这里有两个细节需要注意。第一个是rectangle参数格式左下角经度、纬度用逗号连接然后分号分隔右上角坐标顺序不能反。第二个是把网格边界做了min限制确保最后一个网格不会越过城市边界。聚合函数负责把路段级数据折算成城市级指数。我用长度加权平均速度作为基础指标再映射到0到10区间。加权逻辑是总长度除以总耗时相当于整个路网的平均速度。def aggregate_city_index(adcode, roads): if not roads: return { adcode: adcode, index: None, level: 未知, avg_speed: 0, road_count: 0, total_length: 0 } total_length sum(r[length] for r in roads) total_time sum(r[length] / r[speed] for r in roads if r[speed] 0) avg_speed total_length / total_time if total_time 0 else 0 if avg_speed 45: index 1.0 level 畅通 elif avg_speed 35: index 3.0 level 缓行 elif avg_speed 25: index 6.0 level 拥堵 elif avg_speed 15: index 8.0 level 严重拥堵 else: index 10.0 level 严重拥堵 return { adcode: adcode, index: round(index, 2), level: level, avg_speed: round(avg_speed, 2), road_count: len(roads), total_length: round(total_length, 2) }5.3 写库与调度入口采集完成后的落库逻辑采用批量插入的方式。指标表、日志表和失败任务表分别写入。插指标表时不仅写聚合结果也把原始JSON快照一并存入方便将来切换聚合算法后重算。import psycopg2 from psycopg2 import extras import json def save_index_batch(conn, records): sql INSERT INTO traffic_index (adcode, city_name, collect_time, index_value, congestion_level, avg_speed, road_count, grid_count, raw_json) VALUES %s ON CONFLICT (adcode, collect_time) DO UPDATE SET index_value EXCLUDED.index_value, congestion_level EXCLUDED.congestion_level, avg_speed EXCLUDED.avg_speed, road_count EXCLUDED.road_count, grid_count EXCLUDED.grid_count, raw_json EXCLUDED.raw_json values [ ( r[adcode], r[city_name], r[collect_time], r[index_value], r[congestion_level], r[avg_speed], r[road_count], r[grid_count], json.dumps(r[raw_json], ensure_asciiFalse) ) for r in records ] extras.execute_values(conn.cursor(), sql, values, page_size1000) conn.commit()调度入口比较简单。主脚本读取城市配置表遍历所有启用城市调用scan_city函数将结果汇总写入数据库。crontab里配置一行即可*/5 * * * * cd /opt/traffic-collector python3 main.py logs/collect.log 21这个入口还有一个关键点同一时刻只能有一个任务的单实例锁。我用了简单的文件锁避免上一轮任务还没跑完下一轮cron又拉起新进程导致同一时段重复采集和配额浪费。6. 我在实际运行中总结的几条实用提醒6.1 坐标系必须保持统一否则网格边界会错位高德地图使用GCJ-02坐标系和常见的WGS-84坐标系并不完全一致两者在城市尺度上会存在几十米到几百米的偏移。如果你从其他来源拿到的是WGS-84坐标的城市边界直接拿来切网格会导致网格和实际道路位置错位边缘区域的路况可能会漏采或者重复计算。我的做法是统一使用高德系坐标系。城市边界通过高德行政区划接口获取切网格也用同一套坐标这样从源头上就避免了混用问题。如果一定要从WGS-84转过来记得先做坐标转换再切网格不要嫌麻烦。这个问题我一开始没注意后来发现某个城市边缘路段的采集数量明显偏少排查半天才发现是两种坐标系混用的结果。6.2 空路况区域和边界城市的数据预处理城市范围内的网格不是每个都有道路数据。偏远郊区、大型水域、山区这些区域请求返回的道路列表可能为空这是正常现象聚合时直接跳过即可不要当成失败任务去补采。边界城市还要特别注意比如城市矩形范围边缘正好切到隔壁城市的区域会导致隔壁城市的部分道路被算进当前城市的指标里。处理办法是用高德行政区划接口拿到城市边界的多边形坐标做一次网格中心点与多边形的位置判断中心点不在本城市的网格直接跳过。这个判断会增加一点计算量但对城市指数的准确性影响很大尤其是在长三角、珠三角这类城市密集连片的区域。6.3 API Key安全管理与合规使用边界长期运行的项目API Key的安全管理必须当成正式工程问题对待。我的做法是Key统一放在独立的环境变量文件里采集进程启动时从环境变量加载代码仓库里只保留模板不保留真实Key。同时在高德控制台配置IP白名单只允许采集服务器IP调用接口即使Key泄漏别人也无法从其他网络环境盗用。合规方面也要多说一句。高德开放平台的接口调用前提是遵守其开发者服务条款和数据使用规范所以在做数据采集和持久化存储时要控制调用频率和数据规模合理使用避免对服务造成压力使用者应当仔细阅读并遵守高德开放平台的相关条款尤其注意数据的存储范围、用途限制和展示规范。这个项目的设计方案适合个人学习研究和内部数据分析如果要做商业化或对外提供数据服务务必先确认是否符合平台要求。这套采集系统跑到现在我最深的体会是写采集代码只花了两三天真正让数据连续几年不断档的功夫全在运维和细节上。网格参数、重试策略、失败补采、坐标系一致性、Key安全管理每一项单独拿出来都不复杂但连在一起就是一个稳定运行的自动化数据生产系统。如果你也想攒一份属于自己的城市交通历史数据库别指望有捷径先跑起来用半年后的视野回看今天你一定会庆幸自己从今天就开始采了。本文还有配套的精品资源点击获取
自建城市交通拥堵指数历史数据库：高德API实时数据采集与持久化方案
简介本资源是一个面向交通大数据分析、城市智能治理及GIS开发者的高德地图API实战项目聚焦全国主要城市自2019年10月8日起的实时与历史交通拥堵指数自动化采集与持久化存储。项目提供完整Java工程结构含4个核心Java类、MySQL建表SQL脚本city_crowd_data.sql、配置文件pom.xml、resources目录、说明文档docx及README.md技术指南共10个文件总大小仅42KB轻量易部署。代码已封装定时调度、API异常重试、数据清洗与批量入库逻辑支持长期稳定运行SQL脚本定义了含时间戳、城市编码、拥堵指数、数据来源等字段的规范化表结构便于后续趋势分析与可视化。目前已有44人学习下载适合具备Java基础与数据库操作能力的开发者快速复用采集框架、构建自有城市交通拥堵历史数据库并延伸应用于拥堵预测、交通态势看板或政策效果评估等场景。 你研究城市交通数据的时候十有八九会撞上同一堵墙打开高德地图开放平台实时路况接口、交通态势接口都在参数填好就能拿到当前全城路况但想往前翻历史数据没了。高德只给实时快照不给回放任何需要历史拥堵指数做分析的项目都得自己从启动那天开始一张张攒。这个项目就是这么来的——从2019年10月8日开始按固定频率定时调用高德地图API接口把全国主要城市的实时与历史交通拥堵指数数据采集下来持久化存储到数据库中逐步搭起一个城市交通拥堵指数历史数据库。本文会把整套方案的接口能力边界、采集设计、数据库建模、调度容灾和实测细节全部拆开讲适合正在做交通数据分析、通勤规律研究、物流调度优化或城市拥堵趋势观测的人直接参考复现。1. 为什么说高德API“给实时不给历史”自建指数库到底解决了什么1.1 高德交通API的真实能力边界能拿到什么、拿不到什么我先说结论高德的交通态势相关API本质上是“当前路况的切片接口”核心能力是给定一个地理矩形范围返回当前时刻这个范围内所有路段的通行状态、平均速度和道路名称。你把请求发出去拿到的是此时此刻的交通态势接口背后没有时间维度参数没有“某年某月某日某时刻的路况”这种查询能力。具体来说高德开放平台提供的主要路况能力是交通态势接口通过构造矩形区域的左下角和右上角经纬度就能拉取该范围内的实时道路交通态势。返回的字段里通常包括道路名称、路段方向、道路状态码畅通、缓行、拥堵、严重拥堵、通行速度、路段长度等。每条路段还有一个状态等级可以直接用来做拥堵评估。这些数据很丰富但也正因如此容易被误解。很多第一次接触的人以为可以一次性拉回整个城市的路况或者以为可以带上时间参数去查历史结果发现接口文档里根本没有“历史时间”这个输入项也没有“导出历史数据”的后台功能。这就是“给实时不给历史”的本质你想做历史分析必须自己在实时数据流经时把它截留下来存进数据库别无他法。1.2 一个“只进不出”的接口为什么还值得围绕它建库既然历史数据只能靠自己积累那第一个问题就是花几个月、几年时间持续采集到底值不值我的判断是只要你的使用场景里有一点“分析过去”的需求就非常值。城市拥堵指数如果只有今天的快照你只能回答“现在哪里堵”有了连续几个月或几年的历史库你能回答的问题就完全不一样了早高峰和晚高峰的拥堵峰值分别出现在几点工作日和周末的差异有多大。重点商圈周边道路的拥堵何时开始、何时结束是否与大型活动、天气变化相关。某条快速路改造前后周边路网的拥堵指数有没有明显变化。打车、外卖、物流配送的调度策略能不能根据一周内不同时段的拥堵规律做预判。我见过很多团队做城市交通分析时最头疼的不是建模、不是画图而是数据来源。政府开放的交通数据覆盖城市有限、字段粒度不一商业数据平台价格不低自己爬非官方来源又面临稳定性和合规风险。相比之下高德开放平台有官方API、结构清晰、字段稳定只要按照合理的频率和规范去调用个人和小团队完全能承担唯一需要付出的就是持续积累的时间成本。1.3 项目的时间窗口与城市范围从2019年10月8日开始的全国主要城市标题里“自2019年10月8日起至今”这个时间点我需要多说一句。很多人会误以为高德API能提供从2019年开始的历史数据实际完全不是。真实情况是API不提供任何历史回溯能力所以这个日期实际是项目的启动时间也就是数据采集的起点。从这天开始系统按固定频率持续运行才有后来覆盖多年的历史时间序列。如果你的项目还没有启动那就从今天开始攒越早开始未来的“历史”越长。城市范围我定义为全国主要城市包含直辖市、省会城市和部分计划单列市用高德的adcode城市编码作为唯一标识。比如北京110000、上海310000、广州440100、深圳440300、成都510100这些编码在后续的数据表设计和API调用中会反复用到。采集初期的城市列表我通过高德行政区划接口自动获取同时人工剔除掉一些偏远地区数据量过少的地级市最终保留了几十个核心城市覆盖全国主要人口和经济区域。2. 采集链路方案设计从“整城路况”到“城市拥堵指数”的核心算法2.1 为什么不能直接拉整个城市接口的矩形范围限制如果你拿着高德交通态势接口文档第一反应大概率是把城市外包矩形作为参数一次把全城路况拉回来。但实际操作中你会发现接口对矩形范围有大小限制一次性传一个覆盖整个城市的大矩形要么请求失败要么返回的路况数据被系统截断很多路段直接缺失。这不是高德故意刁难而是服务端在降低单次计算压力。一个覆盖全城的超大矩形包含的路段数量和坐标点数量巨大普通接口很难在稳定延迟内处理完。合理做法是“分片拉取”把城市拆成一个个小网格依次请求每个网格的路况最后把结果合并拼成整个城市的路况快照。这个策略和地图瓦片加载的逻辑很像。你刷手机地图的时候地图也不是一次性渲染整个屏幕而是按瓦片分层加载每个瓦片只绘制一个小区域拼起来就是一整幅地图。网格化采集路况本质上就是给路况数据做了一层“瓦片化”。2.2 网格粒度如何定一个中等城市需要多少个请求网格粒度是方案里最先要定的参数。网格划得太大局部路况细节会被吞掉而且容易触发接口的范围限制网格划得太小请求数量成倍增加整体扫描一轮的时间会拉长流量成本也飙升。我用的做法是按经度和纬度各0.1度划分网格。0.1度在赤道附近大概对应11公里左右在纬度40度左右大概对应8.5公里这个粒度对于观测城市级拥堵趋势已经足够既能保留主干道和快速路的路段细节又不会让网格数量失控。以北京为例北京城区的经度跨度大约1.6度纬度跨度大约1.2度按0.1度网格切分大概是16乘12也就是192个网格。每个网格请求一次扫完整个北京需要约192次API调用。一次请求加解析大约0.2秒单城一轮扫描大约40秒。这个耗时完全可以接受即便同时管几十个城市也能在几分钟内完成一轮全国扫描。如果你的分析需要更细的道路级路况可以将网格缩小到0.05度但要注意网格数量会变成原来的4倍扫描时间也相应拉长API配额消耗更快需要根据配额情况做权衡。2.3 从道路状态到城市拥堵指数聚合算法与指数映射每个网格返回的数据是“路段级”的有道路名称、状态码、速度、长度。但最终要入库和展示的核心指标是“城市拥堵指数”一个0到10之间的数值所以必须把路段级数据聚合成城市级指数。我采用的聚合逻辑分两步。第一步是加权平均速度把该城市所有路段按道路长度加权算出当前时刻全市路网的平均通行速度。之所以用长度加权是因为长路段对整体路网状态的影响自然比短路段大如果简单平均大量短小的支路会稀释主干路的拥堵信号。第二步是把平均速度映射到0到10的拥堵指数区间。我用的映射方式和大众认知中的“拥堵指数”对齐速度越高指数越低速度越低指数越高。具体映射关系大概是这样平均速度接近甚至超过自由流速时指数在0到2之间属于畅通速度明显下降但还能正常行驶时指数在3到5之间属于缓行速度很低、走走停停时指数在6到8之间属于拥堵几乎停滞时指数在9到10之间属于严重拥堵。为便于观察每个城市在同一时刻可能同时存在不同道路的拥堵状态聚合后得到的就是该城市整体拥堵程度的一个综合评分。除此之外我也会把每个网格内的状态码统计一并保存比如畅通路段占比、拥堵路段占比方便后续做更细粒度的分析。数据入库时同时保存原始JSON快照这样即使后续调整聚合算法也能用历史数据重新计算不会因为算法升级丢掉原始信息。3. 数据库建模与持久化千万条记录怎么存才能既完整又查得快3.1 表结构设计配置表、指标表、日志表、失败任务表数据库选型上我最终用了PostgreSQL。选择它不是因为高德API有什么特殊要求而是这组数据带有明显的时间序列特征PostgreSQL对这类数据的写入、分区、查询支持都很成熟而且相对轻量自己一台小服务器就能跑。如果你更熟MySQL整体方案也可以平移过去核心表结构和索引设计思路是一样的。整个项目里我建了四张表各司其职。第一张是城市配置表存所有采集城市的adcode、城市名、城市外包矩形边界、网格大小和启用状态。后续采集主程序会读取这张表动态决定要扫描哪些城市以及扫描范围新增城市时只改表不改代码。第二张是交通拥堵指数指标表这是核心数据表。每条记录对应一个城市在某个采集时刻的聚合指数字段包括adcode、城市名称、采集时间、拥堵指数值、拥堵等级、全市加权平均速度、参与聚合的路段数量、网格数量以及原始JSON快照。原始快照字段很占空间但对追溯和算法复算很有用我建议保留。第三张是采集日志表记录每一轮采集任务的城市、启动时间、成功网格数、失败网格数、总耗时和状态。这张表的价值在运维后期排查“某个时间段数据为什么缺失”时完全靠它。第四张是失败任务表。凡是请求失败的网格任务ID、城市、网格坐标、错误码、失败原因都会写进去供补采程序读取。3.2 时间分区与索引设计让查询不随数据量膨胀变慢交通指数数据是典型的时序数据。以5分钟一次、50个城市计算一天就是14400条记录一年超过500万条。如果所有数据堆在一张普通表里时间一久查询会明显变慢所以必须在表设计阶段就考虑分区。我的做法是按月份做范围分区每个分区存一个月的指标数据。这样查询某段时间的数据时数据库可以直接定位到对应分区不需要全表扫描。建表时声明分区键是采集时间月份变化时系统可以自动创建新分区也可以每天由调度任务检查并创建下月分区。索引方面最常用查询条件自然是“某个城市在某段时间内的指数序列”所以我在adcode和采集时间上建了联合索引。原始快照字段一般不参与查询条件不用建索引避免写入和存储负担。日志表和失败任务表索引相对简单主要按任务ID和城市编码查询建普通索引就够。3.3 批量写入与连接池别一条一条insert否则数据会越攒越落后刚开始做采集的时候我犯过一个典型错误每条指标数据都单独执行一次insert。看起来逻辑简单但数据量上来后写入速度立刻成为瓶颈采集速度跟不上调度频率后面就开始排队数据越攒越落后。后来改成批量写入用psycopg2的execute_values或者PostgreSQL原生的COPY语义一次性把一批几十条、几百条记录写进去。同样是插入5000条记录逐条插入可能需要几十秒批量插入往往只需要一两秒性能差距非常明显。连接管理也很关键。每轮采集任务开始前建立连接任务结束后统一提交关闭不要在每次insert时都新建连接。多城市并行采集时更要小心连接数量控制不好会把数据库连接池打满。我一般设置为每个采集进程一个长连接任务结束再归还实际运行起来非常稳定。4. 调度、异常与长期稳定性让采集任务连续跑几年不出问题的关键细节4.1 定时调度与采集频率5分钟一次是成本和时效的平衡点采集频率直接影响数据的时间分辨率也直接影响API配额消耗和存储量。我最终定的是每次更新频率为5分钟也就是每5分钟对全国主要城市各做一轮完整扫描。5分钟这个值不是拍脑袋定的。城市拥堵的演变周期通常在10分钟以上5分钟的采样间隔已经能还原拥堵形成和消散的完整过程再加密的话数据量成倍上涨但信息增益非常有限。从调度实现上说5分钟与crontab的分钟精度完美对齐本身就是天然的时间边界不容易累积漂移。调度器我用了系统级crontab定时执行采集主脚本。选择crontab而不是常驻后台进程是因为它足够简单可靠脚本崩溃了下一轮cron会重新拉起不会出现后台进程挂掉之后再也不会自动恢复的情况。采集任务跑几年稳定性一定优先于炫技crontab这种朴素方案恰恰是最稳的。4.2 限流、超时与重试高德API的“软性限制”必须提前应对长期调用高德API接口必然遇到的问题就是限流。高德开放平台对普通开发者有配额限制并发超过一定阈值或单位时间请求过于密集时接口会返回错误码。这个限制是“软性”的不是绝对禁止但如果你不注意调用节奏轻则单次请求失败重则一段时间内被系统限流所有请求都打不出去。我的处理方式有三个层次。第一层是控制并发同一时刻只有一个城市的扫描在跑城市之间串行或限制最大并发数避免瞬时请求量过大。第二层是请求间隔每个网格请求之间加一个小延迟比如0.1到0.2秒这样一轮扫描的时间能够接受也远在限流阈值之下。第三层是重试单次请求失败后不立即重试而是做指数退避第一次等1秒第二次等2秒第三次等4秒最多重试三次。三次仍然失败就记录到失败任务表留给补采程序处理。超时设置也必须显式指定。依赖库默认的超时时间往往太长网络异常时请求会卡住很久拖慢整个任务。我在所有HTTP请求里都设置了5秒连接超时和10秒读取超时超过就直接判失败走重试逻辑绝不傻等。4.3 数据完整性监控与补采机制历史是攒出来的中间缺了会很麻烦数据完整性是这个项目的生命线。实时数据错过一次未来分析的时候那5分钟就是永久空白无法通过API回补。所以必须有两个机制兜底监控和补采。监控的核心是检查采集日志表。在每天固定时间跑一个统计SQL校验每个城市当天应有的采集次数和实际入库次数差异超过阈值就报警。同时检查是否有连续多个时间点完全缺失的情况这种情况往往意味着调度器或网络整体故障光靠补采不一定能解决需要人工介入。补采针对的是“个别网格或个别城市请求失败”的场景。补采程序读取失败任务表按任务ID和城市编码重新发起请求把缺失的数据补写进指标表。补采一般安排在凌晨低峰时段避免和正常采集任务抢占资源和挤占配额。如果失败原因是全市性网络中断凌晨补采时也一并重试能补回多少算多少。5. 直接可复现的最小实现从配置到代码的完整示例5.1 环境准备与依赖整个采集服务就是一组Python脚本加一个PostgreSQL数据库环境要求很低。Python 3.8以上够用依赖库只需要requests、psycopg2-binary、retry这几个。数据库用PostgreSQL 12以上创建好数据库和对应表之后再建一个只读配置表用的普通账号尽量不要用超级管理员跑业务。高德开放平台需要先申请Web服务API Key申请地址在开放平台控制台创建应用后就能拿到key。Key建议放在环境变量或独立配置文件中不要写死在代码仓库里避免泄露后被其他人盗刷配额。5.2 核心采集代码网格扫描加聚合计算先看单城市扫描的核心代码。下面这个函数接收city配置按网格大小把城市矩形切成多个小网格逐格请求交通态势接口将各路段的长度、速度、状态解析出来最后聚合成该城市的拥堵指数。这一层是最核心的实现。import requests import time import math API_KEY your_amap_web_api_key TRAFFIC_URL https://restapi.amap.com/v3/traffic/status/rectangle STATUS_MAP { 1: 畅通, 2: 缓行, 3: 拥堵, 4: 严重拥堵 } def scan_city(city): adcode city[adcode] min_lon, min_lat city[min_lon], city[min_lat] max_lon, max_lat city[max_lon], city[max_lat] grid_size city.get(grid_size, 0.1) roads [] grid_count 0 failed_grids [] lon min_lon while lon max_lon: lat min_lat while lat max_lat: rect f{lon},{lat};{min(lon grid_size, max_lon)},{min(lat grid_size, max_lat)} try: resp requests.get( TRAFFIC_URL, params{key: API_KEY, rectangle: rect, level: 6}, timeout(5, 10) ) data resp.json() if data.get(status) 1: traffic_info data.get(trafficinfo, {}) road_list traffic_info.get(roads, []) for road in road_list: roads.append({ name: road.get(name, ), status: road.get(status, ), speed: float(road.get(speed, 0) or 0), length: float(road.get(length, 0) or 0) }) grid_count 1 else: failed_grids.append({rect: rect, info: data.get(info)}) except Exception as e: failed_grids.append({rect: rect, error: str(e)}) lat grid_size time.sleep(0.15) lon grid_size return aggregate_city_index(adcode, roads), grid_count, failed_grids这里有两个细节需要注意。第一个是rectangle参数格式左下角经度、纬度用逗号连接然后分号分隔右上角坐标顺序不能反。第二个是把网格边界做了min限制确保最后一个网格不会越过城市边界。聚合函数负责把路段级数据折算成城市级指数。我用长度加权平均速度作为基础指标再映射到0到10区间。加权逻辑是总长度除以总耗时相当于整个路网的平均速度。def aggregate_city_index(adcode, roads): if not roads: return { adcode: adcode, index: None, level: 未知, avg_speed: 0, road_count: 0, total_length: 0 } total_length sum(r[length] for r in roads) total_time sum(r[length] / r[speed] for r in roads if r[speed] 0) avg_speed total_length / total_time if total_time 0 else 0 if avg_speed 45: index 1.0 level 畅通 elif avg_speed 35: index 3.0 level 缓行 elif avg_speed 25: index 6.0 level 拥堵 elif avg_speed 15: index 8.0 level 严重拥堵 else: index 10.0 level 严重拥堵 return { adcode: adcode, index: round(index, 2), level: level, avg_speed: round(avg_speed, 2), road_count: len(roads), total_length: round(total_length, 2) }5.3 写库与调度入口采集完成后的落库逻辑采用批量插入的方式。指标表、日志表和失败任务表分别写入。插指标表时不仅写聚合结果也把原始JSON快照一并存入方便将来切换聚合算法后重算。import psycopg2 from psycopg2 import extras import json def save_index_batch(conn, records): sql INSERT INTO traffic_index (adcode, city_name, collect_time, index_value, congestion_level, avg_speed, road_count, grid_count, raw_json) VALUES %s ON CONFLICT (adcode, collect_time) DO UPDATE SET index_value EXCLUDED.index_value, congestion_level EXCLUDED.congestion_level, avg_speed EXCLUDED.avg_speed, road_count EXCLUDED.road_count, grid_count EXCLUDED.grid_count, raw_json EXCLUDED.raw_json values [ ( r[adcode], r[city_name], r[collect_time], r[index_value], r[congestion_level], r[avg_speed], r[road_count], r[grid_count], json.dumps(r[raw_json], ensure_asciiFalse) ) for r in records ] extras.execute_values(conn.cursor(), sql, values, page_size1000) conn.commit()调度入口比较简单。主脚本读取城市配置表遍历所有启用城市调用scan_city函数将结果汇总写入数据库。crontab里配置一行即可*/5 * * * * cd /opt/traffic-collector python3 main.py logs/collect.log 21这个入口还有一个关键点同一时刻只能有一个任务的单实例锁。我用了简单的文件锁避免上一轮任务还没跑完下一轮cron又拉起新进程导致同一时段重复采集和配额浪费。6. 我在实际运行中总结的几条实用提醒6.1 坐标系必须保持统一否则网格边界会错位高德地图使用GCJ-02坐标系和常见的WGS-84坐标系并不完全一致两者在城市尺度上会存在几十米到几百米的偏移。如果你从其他来源拿到的是WGS-84坐标的城市边界直接拿来切网格会导致网格和实际道路位置错位边缘区域的路况可能会漏采或者重复计算。我的做法是统一使用高德系坐标系。城市边界通过高德行政区划接口获取切网格也用同一套坐标这样从源头上就避免了混用问题。如果一定要从WGS-84转过来记得先做坐标转换再切网格不要嫌麻烦。这个问题我一开始没注意后来发现某个城市边缘路段的采集数量明显偏少排查半天才发现是两种坐标系混用的结果。6.2 空路况区域和边界城市的数据预处理城市范围内的网格不是每个都有道路数据。偏远郊区、大型水域、山区这些区域请求返回的道路列表可能为空这是正常现象聚合时直接跳过即可不要当成失败任务去补采。边界城市还要特别注意比如城市矩形范围边缘正好切到隔壁城市的区域会导致隔壁城市的部分道路被算进当前城市的指标里。处理办法是用高德行政区划接口拿到城市边界的多边形坐标做一次网格中心点与多边形的位置判断中心点不在本城市的网格直接跳过。这个判断会增加一点计算量但对城市指数的准确性影响很大尤其是在长三角、珠三角这类城市密集连片的区域。6.3 API Key安全管理与合规使用边界长期运行的项目API Key的安全管理必须当成正式工程问题对待。我的做法是Key统一放在独立的环境变量文件里采集进程启动时从环境变量加载代码仓库里只保留模板不保留真实Key。同时在高德控制台配置IP白名单只允许采集服务器IP调用接口即使Key泄漏别人也无法从其他网络环境盗用。合规方面也要多说一句。高德开放平台的接口调用前提是遵守其开发者服务条款和数据使用规范所以在做数据采集和持久化存储时要控制调用频率和数据规模合理使用避免对服务造成压力使用者应当仔细阅读并遵守高德开放平台的相关条款尤其注意数据的存储范围、用途限制和展示规范。这个项目的设计方案适合个人学习研究和内部数据分析如果要做商业化或对外提供数据服务务必先确认是否符合平台要求。这套采集系统跑到现在我最深的体会是写采集代码只花了两三天真正让数据连续几年不断档的功夫全在运维和细节上。网格参数、重试策略、失败补采、坐标系一致性、Key安全管理每一项单独拿出来都不复杂但连在一起就是一个稳定运行的自动化数据生产系统。如果你也想攒一份属于自己的城市交通历史数据库别指望有捷径先跑起来用半年后的视野回看今天你一定会庆幸自己从今天就开始采了。本文还有配套的精品资源点击获取
