AIS数据集清洗实战：从原始数据到高质量输入的完整指南

1次阅读

共计 2665 个字符，预计需要花费 7 分钟才能阅读完成。

AIS（Automatic Identification System）数据是海事领域的重要数据源，记录了船舶的实时动态和静态信息。原始 AIS 数据通常存在以下典型问题：

MMSI 跳变：同一船舶的 MMSI（海上移动服务标识）在短时间内发生不合理变化
经纬度漂移：船舶位置突然跳跃到不合理坐标（如陆地中央或极远海域）
时间戳断裂：数据采集间隔不均匀，存在长时间数据缺失
速度 / 航向突变：不符合船舶物理运动规律的速度或方向突变
重复记录：相同时间点的多条重复数据

使用 Pandas 的 drop_duplicates() 方法，关键参数配置：

df_cleaned = df.sort_values('timestamp').drop_duplicates(subset=['mmsi', 'timestamp'],  # 根据船舶 ID 和时间戳去重
    keep='first'  # 保留第一条记录
)

采用 Hampel 滤波器识别异常轨迹点，原理是计算滑动窗口内的中位数绝对偏差：

from scipy.stats import median_abs_deviation

def hampel_filter(values: np.ndarray, window_size: int=5, n_sigma: float=3.0) -> np.ndarray:
    """
    values: 待检测的经纬度序列
    window_size: 滑动窗口大小
    n_sigma: 异常值判定阈值
    """
    indices = []
    for i in range(len(values)):
        window = values[max(0, i-window_size):min(len(values), i+window_size)]
        median = np.median(window)
        mad = median_abs_deviation(window, scale='normal')
        if abs(values[i] - median) > n_sigma * mad:
            indices.append(i)
    return np.array(indices)

对时空数据进行线性插值，保持轨迹连续性：

from scipy.interpolate import interp1d

# 时间戳转换为秒数
base_time = df['timestamp'].min()
time_sec = (df['timestamp'] - base_time).dt.total_seconds()

# 创建插值函数
lat_interp = interp1d(time_sec, df['latitude'], kind='linear', fill_value='extrapolate')
lon_interp = interp1d(time_sec, df['longitude'], kind='linear', fill_value='extrapolate')

# 生成均匀时间序列
new_time_sec = np.linspace(time_sec.min(), time_sec.max(), num=1000)
new_lat = lat_interp(new_time_sec)
new_lon = lon_interp(new_time_sec)

所有函数都应添加类型提示，并处理边界情况：

from typing import Optional, Tuple
import pandas as pd

def clean_ais_data(
    raw_df: pd.DataFrame,
    time_col: str = 'timestamp',
    mmsi_col: str = 'mmsi'
) -> Tuple[pd.DataFrame, Optional[Exception]]:
    """
    执行 AIS 数据清洗主流程

    Args:
        raw_df: 原始 DataFrame
        time_col: 时间列名
        mmsi_col: 船舶 ID 列名

    Returns:
        清洗后的 DataFrame 和可能的异常对象
    """
    try:
        # 去重处理
        df = raw_df.sort_values(time_col).drop_duplicates(subset=[mmsi_col, time_col],
            keep='first'
        )

        # 时区统一处理（示例）if not pd.api.types.is_datetime64_any_dtype(df[time_col]):
            df[time_col] = pd.to_datetime(df[time_col], utc=True)

        return df, None
    except Exception as e:
        return raw_df, e

急转弯误判：设置合理的速度 / 航向变化阈值，避免将正常操作识别为异常
时区处理：始终使用 UTC 时间存储，避免本地时区转换错误
大规模数据：使用 Dask 替代 Pandas 处理超大规模数据集

import dask.dataframe as dd

ddf = dd.read_parquet('s3://bucket/large_ais_dataset/*.parquet')
ddf = ddf.drop_duplicates(subset=['mmsi', 'timestamp'])

使用 Matplotlib 绘制清洗前后轨迹对比：

import matplotlib.pyplot as plt

fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6))

# 原始数据
ax1.scatter(raw_df['longitude'], raw_df['latitude'], 
           c='red', s=5, label='原始数据')
ax1.set_title('清洗前轨迹')

# 清洗后数据
ax2.scatter(clean_df['longitude'], clean_df['latitude'], 
           c='blue', s=5, label='清洗后数据')
ax2.set_title('清洗后轨迹')

plt.tight_layout()
plt.savefig('trajectory_comparison.png')

对于更高精度的清洗需求，可考虑：