路径规划的数据存储架构：实时路况、历史轨迹与AI预测的统一存储

路径规划的数据存储架构实时路况、历史轨迹与AI预测的统一存储一、当导航推荐的最优路径堵成了停车场某配送平台的算法团队发现一个矛盾系统推荐的预计30分钟送达路线实际平均耗时48分钟。根本原因是路径规划引擎同时依赖于三种完全不同类型的数据实时路况Redis每30秒更新一次来自地图API的交通拥堵数据历史轨迹HDFS千万条历史配送轨迹过去3个月同路段同时间的实际耗时AI预测模型服务基于天气/节假日/促销活动的拥堵预测这三种数据的更新频率、存储介质、查询模式完全不同。实时路况要求毫秒级读取历史轨迹需要扫描海量数据AI预测依赖模型的定时更新。把它们统一在一个查询中做路径规划就是要把三种存储范式融合到同一条SQL/API调用中。二、三种数据的统一存储与查询架构三、混合数据存储的实现ClickHouse轨迹表设计CREATE TABLE delivery_trajectories ( delivery_id String, courier_id UInt32, segment_id UInt64, -- 路段ID enter_time DateTime64(3), exit_time DateTime64(3), travel_time_sec Float32, distance_meters Float32, avg_speed_kmh Float32, is_congested UInt8, vehicle_type LowCardinality(String), weather LowCardinality(String), holiday_flag UInt8 ) ENGINE MergeTree() PARTITION BY toYYYYMMDD(enter_time) ORDER BY (segment_id, enter_time) TTL enter_time INTERVAL 90 DAY SETTINGS index_granularity 8192;统一路径规划服务from dataclasses import dataclass from typing import List, Tuple import numpy as np dataclass class RouteSegment: segment_id: int length_m: float current_traffic: float # 1.0畅通, 3.0严重拥堵 historical_avg_sec: float predicted_sec: float class RoutePlanner: def __init__(self, redis_client, clickhouse_client, ai_model, mysql_pool): self.redis redis_client self.ch clickhouse_client self.model ai_model self.mysql mysql_pool def plan_route(self, start: Tuple[float, float], end: Tuple[float, float], departure_time: datetime, context: dict) - dict: 混合数据融合的路径规划 # Step 1: 获取候选路径的路段序列从地图服务 candidate_routes self._get_candidate_routes(start, end) best_route None best_score float(inf) for route in candidate_routes: segment_ids route[segment_ids] # Step 2: 并行获取三种数据 realtime self._get_realtime_traffic(segment_ids) historical self._get_historical_speed( segment_ids, departure_time ) predicted self._get_ai_prediction( segment_ids, departure_time, context ) # Step 3: 加权融合计算每条路段的耗时 total_cost 0 for seg_id in segment_ids: rt_cost realtime.get(seg_id, float(inf)) hist_cost historical.get(seg_id, float(inf)) ai_cost predicted.get(seg_id, float(inf)) # 动态权重实时数据质量高时加大权重 rt_weight 0.5 if self._is_realtime_reliable(seg_id) else 0.1 hist_weight 0.3 ai_weight 1.0 - rt_weight - hist_weight segment_cost (rt_weight * rt_cost hist_weight * hist_cost ai_weight * ai_cost) total_cost segment_cost route[total_cost] total_cost if total_cost best_score: best_score total_cost best_route route return best_route def _get_realtime_traffic(self, segment_ids: list) - dict: 从Redis获取实时路况 try: pipeline self.redis.pipeline() for seg_id in segment_ids: pipeline.hgetall(ftraffic:segment:{seg_id}) results pipeline.execute() traffic {} for seg_id, data in zip(segment_ids, results): if data: # 拥堵系数 → 预计耗时假设路段默认120秒 congestion float(data.get(bcongestion, 1.0)) traffic[seg_id] 120.0 * congestion return traffic except Exception as e: raise TrafficDataException(实时路况查询失败, e) def _get_historical_speed(self, segment_ids: list, departure_time: datetime) - dict: 从ClickHouse获取历史平均耗时 try: # 查询同时段同星期、同小时的历史数据 dow departure_time.weekday() hour departure_time.hour query SELECT segment_id, avg(travel_time_sec) AS avg_time, quantile(0.9)(travel_time_sec) AS p90_time FROM delivery_trajectories WHERE segment_id IN %(seg_ids)s AND toDayOfWeek(enter_time) %(dow)s AND toHour(enter_time) %(hour)s AND enter_time now() - INTERVAL 30 DAY GROUP BY segment_id result self.ch.execute(query, { seg_ids: tuple(segment_ids), dow: dow 1, # ClickHouse: 1Monday hour: hour }) return {row[0]: row[1] for row in result} except Exception as e: raise TrafficDataException(历史轨迹查询失败, e) def _get_ai_prediction(self, segment_ids: list, departure_time: datetime, context: dict) - dict: 从AI模型获取耗时预测 try: # 构造特征向量 features self._build_features( segment_ids, departure_time, context ) predictions self.model.predict(features) return dict(zip(segment_ids, predictions)) except Exception as e: # AI预测不可用时降级到历史数据 return {}四、路径规划数据架构的四个边界边界一实时数据的覆盖盲区。小路、新建道路可能没有实时路况数据。此时应完全依赖历史数据和AI预测而非插值或外推。边界二历史轨迹的稀疏问题。凌晨3点的历史轨迹数据可能只有几条统计不可靠。需要做时段聚合——将凌晨2-4点合并为一个时段统计。边界三AI模型的时效性衰减。道路施工、新开商场等物理变化会让AI预测在变化发生后的前几天严重失准。需要检测预测残差当日均误差30%时触发模型紧急重训。边界四路径规划的超时保护。如果Redis集群挂了、ClickHouse正在做Merge、AI模型容器重启路径规划不能因为这些依赖的故障而完全不可用。降级策略实时路况不可用→用历史数据×1.5倍安全系数历史数据不可用→用地图的静态预估时间AI不可用→仅用实时历史。五、总结物流路径规划的数据存储架构是多源异构数据融合的典型案例。Redis提供毫秒级的实时路况访问ClickHouse提供秒级的历史统计查询AI模型提供分钟级的预测更新。三者在路径规划服务中做加权融合通过动态权重数据越新鲜越可靠权重越高实现最优路径推荐。当前方案的核心不是哪个数据源最准而是**当一个数据源失效时其他数据源能否兜底**。本文属于「行业场景与项目复盘」系列探讨物流路径规划的多源数据融合存储与查询架构。