AIS数据集清洗实战：从原始数据到高质量训练集的完整流程

1次阅读

没有评论

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

AIS（Automatic Identification System）数据是船舶轨迹分析的重要来源，但在实际应用中，我们常常会遇到各种脏数据问题。这些问题会严重影响后续的分析和建模效果，所以数据清洗是 AIS 数据处理中至关重要的一环。

常见的 AIS 数据问题包括：

信号丢失导致的轨迹断裂 ：由于通信中断或设备故障，AIS 信号可能会突然消失，导致船舶轨迹不连续。
基站误差引起的坐标漂移 ：AIS 基站的位置误差可能导致船舶坐标出现异常漂移，尤其是在多基站覆盖区域。
速度异常（SOG）和航向异常（COG）：由于设备故障或信号干扰，AIS 消息中的速度和航向可能会出现不合理值。
时间戳错误 ：不同基站或设备可能使用不同的时区或时间同步机制，导致时间戳不一致。
MMSI 跳变问题 ：船舶的 MMSI（Maritime Mobile Service Identity）可能会因设备更换或其他原因发生变化，导致同一船舶被误认为多艘不同的船舶。

传统规则过滤 ：基于预定义的规则（如速度范围、航向变化率等）过滤异常数据。优点是实现简单、计算开销小；缺点是规则难以覆盖所有异常情况，容易出现误判。
基于机器学习的异常检测 ：利用聚类、孤立森林等算法自动识别异常数据。优点是适应性更强，能发现复杂异常模式；缺点是需要标注数据训练模型，计算开销较大。

批处理清洗 ：适用于历史数据的离线处理，可以充分利用分布式计算资源进行大规模清洗。缺点是延迟较高，不适合实时应用。
流式处理方案 ：适用于实时数据流，能够快速响应新数据。缺点是实现复杂，对计算资源要求较高。

以下是一个基于 PySpark 的轨迹修复代码示例，用于填补信号丢失导致的轨迹断裂：

from pyspark.sql import SparkSession
from pyspark.sql.functions import lag, lead, col, when
from pyspark.sql.window import Window

def repair_trajectory(spark_df, ship_id_col="mmsi", timestamp_col="timestamp", 
                     lon_col="lon", lat_col="lat", max_gap_minutes=30):
    """
    修复 AIS 轨迹中的断裂点
    :param spark_df: 输入的 Spark DataFrame
    :param ship_id_col: 船舶 ID 列名
    :param timestamp_col: 时间戳列名
    :param lon_col: 经度列名
    :param lat_col: 纬度列名
    :param max_gap_minutes: 最大允许填补的时间间隔（分钟）:return: 修复后的 DataFrame
    """
    # 按船舶 ID 和时间戳排序
    window_spec = Window.partitionBy(ship_id_col).orderBy(timestamp_col)

    # 计算相邻点的时间差和距离
    df_with_gaps = spark_df.withColumn("prev_time", lag(timestamp_col).over(window_spec)) \
                          .withColumn("next_time", lead(timestamp_col).over(window_spec)) \
                          .withColumn("prev_lon", lag(lon_col).over(window_spec)) \
                          .withColumn("prev_lat", lag(lat_col).over(window_spec))

    # 标记需要填补的轨迹断裂点
    df_with_gaps = df_with_gaps.withColumn("gap_flag", 
        when((col("prev_time").isNotNull()) & 
            (col("next_time").isNotNull()) & 
            ((col(timestamp_col) - col("prev_time")).cast("long") > max_gap_minutes * 60) & 
            ((col("next_time") - col(timestamp_col)).cast("long") > max_gap_minutes * 60),
            True
        ).otherwise(False)
    )

    # 填补轨迹断裂点（这里简化为线性插值，实际应用中可能需要更复杂的算法）repaired_df = df_with_gaps.withColumn(lon_col, 
        when(col("gap_flag"), (col("prev_lon") + col(lon_col)) / 2).otherwise(col(lon_col))
    ).withColumn(lat_col, 
        when(col("gap_flag"), (col("prev_lat") + col(lat_col)) / 2).otherwise(col(lat_col))
    )

    return repaired_df.drop("prev_time", "next_time", "prev_lon", "prev_lat", "gap_flag")

Haversine 公式可以计算地球上两点之间的大圆距离。我们可以利用它来检测 AIS 消息中的速度异常：

from math import radians, sin, cos, sqrt, atan2

def haversine(lon1, lat1, lon2, lat2):
    """
    计算两点之间的大圆距离（单位：米）:param lon1: 点 1 经度
    :param lat1: 点 1 纬度
    :param lon2: 点 2 经度
    :param lat2: 点 2 纬度
    :return: 距离（米）"""
    # 将十进制度数转化为弧度
    lon1, lat1, lon2, lat2 = map(radians, [lon1, lat1, lon2, lat2])

    # haversine 公式
    dlon = lon2 - lon1
    dlat = lat2 - lat1
    a = sin(dlat/2)**2 + cos(lat1) * cos(lat2) * sin(dlon/2)**2
    c = 2 * atan2(sqrt(a), sqrt(1-a))
    r = 6371000 # 地球平均半径，单位米
    return c * r

def detect_speed_anomalies(df, max_speed_knots=50):
    """
    检测速度异常（SOG）的 AIS 消息
    :param df: 包含 AIS 数据的 DataFrame
    :param max_speed_knots: 最大允许速度（节）:return: 标记了异常的 DataFrame
    """
    # 按船舶 ID 和时间戳排序
    window_spec = Window.partitionBy("mmsi").orderBy("timestamp")

    # 计算相邻点之间的时间和距离
    df_with_speed = df.withColumn("prev_time", lag("timestamp").over(window_spec)) \
                     .withColumn("prev_lon", lag("lon").over(window_spec)) \
                     .withColumn("prev_lat", lag("lat").over(window_spec))

    # 计算实际速度（节）df_with_speed = df_with_speed.withColumn("time_diff", 
        (col("timestamp").cast("long") - col("prev_time").cast("long"))
    ).withColumn("distance", 
        when((col("prev_time").isNotNull()) & (col("time_diff") > 0),
            haversine_udf(col("prev_lon"), col("prev_lat"), col("lon"), col("lat"))
        ).otherwise(0)
    ).withColumn("calculated_speed", 
        when(col("time_diff") > 0, 
            (col("distance") / col("time_diff")) * 1.94384 # 转换为节
        ).otherwise(0)
    )

    # 标记速度异常
    df_with_speed = df_with_speed.withColumn("speed_anomaly", 
        (col("calculated_speed") > max_speed_knots) | 
        (abs(col("calculated_speed") - col("sog")) > max_speed_knots * 0.5)
    )

    return df_with_speed

在分布式处理 AIS 数据时，合理的分区策略可以显著提高处理效率。以下是几种常用的分区策略：

按时间分区 ：适用于按时间范围查询的场景，可以快速定位特定时间段的数据。
按空间分区 ：将数据按地理区域划分，适用于空间范围查询。
按船舶 ID 分区 ：确保同一船舶的数据落在同一节点上，便于轨迹分析。

实际应用中，可以根据查询模式和数据处理需求，选择合适的分区策略或组合策略。

对于大规模的 AIS 数据集，建立时空索引可以加速查询和清洗操作。R-Tree 是一种常用的空间索引结构，可以有效支持范围查询和最近邻查询。

以下是一个使用 GeoSpark（Apache Sedona）构建时空索引的示例：

from pyspark.sql import SparkSession
from sedona.register import SedonaRegistrator
from sedona.utils import SedonaKryoRegistrator

# 初始化 SparkSession
spark = SparkSession.builder \
    .appName("AIS Data Cleaning") \
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
    .config("spark.kryo.registrator", SedonaKryoRegistrator.getName) \
    .getOrCreate()

SedonaRegistrator.registerAll(spark)

# 创建空间索引
from sedona.sql.types import GeometryType
from pyspark.sql.functions import expr

# 将经纬度转换为空间几何对象
df_with_geom = df.withColumn("geom", expr("ST_Point(CAST(lon AS Decimal(24,20)), CAST(lat AS Decimal(24,20)))"))

# 构建 R -Tree 索引
indexed_df = df_with_geom.withColumn("index", expr("ST_GeoHash(geom, 8)"))

MMSI 跳变是指同一船舶在不同时间段使用不同 MMSI 的情况。这可能是由于设备更换、MMSI 重新分配或其他原因导致的。处理 MMSI 跳变问题的方法包括：