3. 数据采集与预处理:GPS轨迹数据、订单数据、天气数据的采集与清洗实战
ETA预测模型的精度高度依赖输入数据的质量。本章聚焦于三类核心数据源——GPS轨迹数据、订单数据、天气数据的采集策略与清洗工程。我们将从数据源特征出发,设计采集管道,并针对每类数据特有的脏数据模式(漂移、缺失、异常、时区不一致等)给出可落地的清洗方案。
核心原则:数据清洗不是一次性任务,而是与特征工程、模型迭代并行的持续过程。本章所有清洗逻辑均需封装为可复用的ETL模块,支持离线批处理与实时流处理两种模式。
3.1 数据源特征与采集策略
三类数据在采集频率、数据量级、实时性要求上差异显著,需采用不同的采集管道。
| 数据源 | 采集频率 | 单条记录大小 | 日数据量(万级) | 实时性要求 | 推荐采集方式 |
|---|---|---|---|---|---|
| GPS轨迹 | 1-5秒/点 | ~200B | 500-2000 | 高(秒级) | Kafka + Flink 流处理 |
| 订单数据 | 事件驱动 | ~2KB | 50-200 | 中(分钟级) | CDC(Debezium)+ Kafka |
| 天气数据 | 1小时/次 | ~500B | 0.24-0.48 | 低(小时级) | 定时API拉取 + 对象存储 |
工程建议:GPS轨迹数据量最大且实时性最高,建议采用Apache Kafka作为消息缓冲层,Flink进行窗口聚合与初步清洗;订单数据通过数据库CDC(Change Data Capture)实时同步;天气数据使用Airflow调度定时任务,从气象局API或第三方服务(如OpenWeatherMap)拉取后存入Parquet格式。
3.2 GPS轨迹数据清洗实战
GPS轨迹数据是ETA预测的核心输入,但也是最容易产生脏数据的数据源。常见问题包括:定位漂移、信号丢失、时间戳异常、坐标重复等。
3.2.1 清洗流程
- 去重与排序:按设备ID + 时间戳排序,去除完全重复的记录。
- 时间戳校验:过滤时间戳早于1970-01-01或晚于当前时间+1小时的记录。
- 坐标范围过滤:剔除经纬度超出城市边界(如北京:39.4-41.1, 115.4-117.5)的点。
- 速度异常检测:计算相邻点间瞬时速度,剔除速度 > 120 km/h(城市道路)或 > 200 km/h(高速)的漂移点。
- 停留点合并:连续5分钟以上速度 < 1 km/h 且坐标变化 < 50m 的点,合并为停留事件。
- 轨迹分段:相邻点时间间隔 > 30分钟 或 距离 > 5km,自动切分为新轨迹段。
# GPS轨迹清洗核心逻辑(Python伪代码)
def clean_gps_trajectory(df: pd.DataFrame) -> pd.DataFrame:
# 1. 排序与去重
df = df.sort_values(['device_id', 'timestamp']).drop_duplicates(
subset=['device_id', 'timestamp', 'lat', 'lng']
)
# 2. 时间戳过滤
now = pd.Timestamp.now(tz='UTC')
df = df[(df['timestamp'] > '1970-01-01') & (df['timestamp'] <= now + pd.Timedelta(hours=1))]
# 3. 坐标范围过滤(以北京为例)
df = df[(df['lat'].between(39.4, 41.1)) & (df['lng'].between(115.4, 117.5))]
# 4. 速度异常检测(使用haversine计算距离)
df['dist_km'] = haversine_vectorized(df['lat'], df['lng'], df['lat'].shift(1), df['lng'].shift(1))
df['time_diff_h'] = (df['timestamp'] - df['timestamp'].shift(1)).dt.total_seconds() / 3600
df['speed_kmh'] = df['dist_km'] / df['time_diff_h'].replace(0, np.nan)
df = df[(df['speed_kmh'].isna()) | (df['speed_kmh'] <= 120)]
return df.drop(columns=['dist_km', 'time_diff_h', 'speed_kmh'])
避坑提示:速度异常检测时,务必处理首条记录(无前驱点)的NaN值,不要直接丢弃。建议使用分组(groupby device_id)后分别计算,避免跨设备计算速度。
3.3 订单数据清洗实战
订单数据通常来自业务数据库,脏数据模式相对固定:字段缺失、状态机异常、时间字段时区不一致、重复支付等。
3.3.1 清洗规则清单
| 字段 | 常见脏数据 | 清洗策略 |
|---|---|---|
| order_id | 空值、重复 | 丢弃空值;重复ID保留最新一条(按更新时间) |
| user_id / driver_id | 空值、格式异常 | 空值标记为-1;非数字字符过滤 |
| pickup_time / dropoff_time | 时区未标准化、时间倒挂 | 统一转为UTC+8;过滤 pickup_time >= dropoff_time 的记录 |
| order_status | 状态跳转异常(如未支付直接完成) | 按状态机校验:created→paid→picked_up→delivered→completed |
| payment_amount | 负数、超出阈值 | 过滤 < 0 或 > 10000 的记录;保留原始值用于异常分析 |
3.3.2 状态机校验示例
# 订单状态机校验
VALID_TRANSITIONS = {
'created': ['paid', 'cancelled'],
'paid': ['picked_up', 'cancelled'],
'picked_up': ['delivered', 'cancelled'],
'delivered': ['completed'],
'completed': [],
'cancelled': []
}
def validate_order_status(history: List[Dict]) -> bool:
"""校验单个订单的状态流转是否合法"""
sorted_history = sorted(history, key=lambda x: x['timestamp'])
for i in range(len(sorted_history) - 1):
curr_status = sorted_history[i]['status']
next_status = sorted_history[i+1]['status']
if next_status not in VALID_TRANSITIONS.get(curr_status, []):
return False
return True
3.4 天气数据清洗实战
天气数据通常来自外部API,脏数据主要表现为:缺失值(传感器故障)、异常值(极端天气记录错误)、空间粒度不匹配(气象站与配送区域不重合)。
3.4.1 清洗策略
- 缺失值处理:温度、湿度等连续变量使用前后2小时均值插补;降水量、风向等离散变量使用最近邻气象站数据填充。
- 异常值检测:使用3-sigma原则或IQR方法,过滤温度超出[-20°C, 50°C]、风速 > 50 m/s 等明显异常记录。
- 空间匹配:将气象站坐标与配送区域(网格或行政区划)进行空间连接,每个区域取最近3个气象站的加权平均(反距离权重法)。
# 反距离权重插值(IDW)示例
def idw_interpolation(station_df, target_lat, target_lng, power=2):
"""根据气象站数据插值目标点的天气值"""
distances = haversine_vectorized(
station_df['lat'], station_df['lng'],
target_lat, target_lng
)
weights = 1 / (distances ** power + 1e-10)
weights /= weights.sum()
interpolated_value = (station_df['value'] * weights).sum()
return interpolated_value
3.5 多源数据对齐与时间窗口划分
清洗后的三类数据需要按时间维度和空间维度对齐,才能用于ETA模型训练。
3.6 数据质量监控与告警
清洗管道上线后,必须建立数据质量监控体系,及时发现数据源异常或清洗逻辑失效。
| 监控指标 | 数据源 | 告警阈值 | 响应动作 |
|---|---|---|---|
| GPS点丢失率 | GPS轨迹 | > 10% 连续5分钟 | 切换备用数据源;标记该时段为低置信度 |
| 订单状态异常率 | 订单数据 | > 1% 每小时 | 暂停模型更新;人工核查业务系统 |
| 天气数据缺失率 | 天气数据 | > 20% 单日 | 启用历史均值填充;通知数据采购方 |
| 清洗后数据量波动 | 全部 | 日环比 > ±30% | 触发全链路数据审计 |
工程落地:建议使用Apache Griffin或自建基于Prometheus + Grafana的数据质量监控看板。每个清洗步骤输出数据质量报告(JSON格式),包含输入行数、输出行数、丢弃行数、丢弃原因分布,便于快速定位问题。