路径规划的数据存储架构：实时路况、历史轨迹与AI预测的统一存储
一、当导航推荐的"最优路径"堵成了停车场
某配送平台的算法团队发现一个矛盾：系统推荐的"预计30分钟送达"路线，实际平均耗时48分钟。根本原因是路径规划引擎同时依赖于三种完全不同类型的数据：
这三种数据的更新频率、存储介质、查询模式完全不同。实时路况要求毫秒级读取，历史轨迹需要扫描海量数据，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 {}
四、路径规划数据架构的四个边界
边界一：实时数据的覆盖盲区。小路、新建道路可能没有实时路况数据。此时应完全依赖历史数据和AI预测，而非插值或外推。
边界二：历史轨迹的稀疏问题。凌晨3点的历史轨迹数据可能只有几条，统计不可靠。需要做"时段聚合"——将凌晨2-4点合并为一个时段统计。
边界三：AI模型的时效性衰减。道路施工、新开商场等物理变化会让AI预测在变化发生后的前几天严重失准。需要检测预测残差，当日均误差>30%时触发模型紧急重训。
边界四：路径规划的超时保护。如果Redis集群挂了、ClickHouse正在做Merge、AI模型容器重启，路径规划不能因为这些依赖的故障而完全不可用。降级策略：实时路况不可用→用历史数据×1.5倍安全系数；历史数据不可用→用地图的静态预估时间；AI不可用→仅用实时+历史。
五、总结
物流路径规划的数据存储架构是多源异构数据融合的典型案例。Redis提供毫秒级的实时路况访问，ClickHouse提供秒级的历史统计查询，AI模型提供分钟级的预测更新。三者在路径规划服务中做加权融合，通过动态权重（数据越新鲜越可靠，权重越高）实现最优路径推荐。
当前方案的核心不是"哪个数据源最准"，而是**"当一个数据源失效时，其他数据源能否兜底"**。
本文属于「行业场景与项目复盘」系列，探讨物流路径规划的多源数据融合存储与查询架构。
