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

发布时间：2026/7/25 7:41:19

路径规划的数据存储架构：实时路况、历史轨迹与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(f"traffic:segment:{seg_id}") results = pipeline.execute() traffic = {} for seg_id, data in zip(segment_ids, results): if data: # 拥堵系数 → 预计耗时（假设路段默认120秒） congestion = float(data.get(b'congestion', 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: 1=Monday '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 {}