AIS数据集入门指南：从数据采集到预处理的全流程实践

1次阅读

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

AIS（Automatic Identification System）数据是船舶自动识别系统生成的实时航行信息，包含船舶位置、航速、航向等动态数据，以及船名、尺寸等静态数据。这类数据在航运优化、渔业监管、港口调度等领域具有重要商业价值。例如，通过分析历史 AIS 数据，可以优化航线规划，减少燃油消耗；实时监控 AIS 数据则有助于及时发现异常航行行为，提升海上安全。

然而，处理 AIS 数据也面临诸多技术挑战。首先是数据稀疏性问题，由于通信中断或设备故障，AIS 报文可能存在缺失。其次是报文解析复杂度高，原始 AIS 报文采用特定编码格式，需要专业解析工具。此外，数据量庞大（尤其是繁忙海域）对存储和计算性能提出了较高要求。

在处理 AIS 数据时，选择合适的存储格式至关重要。以下是几种常见格式的对比：

CSV：易于阅读和编辑，但缺乏数据类型支持，解析性能较差，不适合大规模数据。
JSON：支持嵌套结构，适合存储复杂的 AIS 报文，但文件体积较大，解析速度慢。
Parquet：列式存储，压缩率高，查询性能优异，特别适合大规模 AIS 数据集。

对于生产环境，推荐使用 Parquet 格式，尤其是在需要频繁查询特定字段（如船舶 MMSI 或时间范围）的场景。

船舶 MMSI（Maritime Mobile Service Identity）是唯一标识符，但原始数据中可能存在重复或无效记录。以下是使用 pandas 进行去重的代码示例：

import pandas as pd

def clean_mmsi(df):
    """
    清理 MMSI 列：去重并过滤无效值
    :param df: 包含 MMSI 列的 DataFrame
    :return: 清理后的 DataFrame
    """
    # 去除 MMSI 为空的记录
    df = df[df['MMSI'].notna()]

    # 转换 MMSI 为整数（原始数据可能是字符串或浮点数）try:
        df['MMSI'] = df['MMSI'].astype(int)
    except ValueError as e:
        print(f"MMSI 转换错误: {e}")
        # 记录异常值以便后续分析
        invalid_mmsi = df[~df['MMSI'].apply(lambda x: str(x).isdigit())]
        invalid_mmsi.to_csv('invalid_mmsi_records.csv', index=False)

        # 仅保留有效的 MMSI
        df = df[df['MMSI'].apply(lambda x: str(x).isdigit())]
        df['MMSI'] = df['MMSI'].astype(int)

    # 按 MMSI 去重，保留最后一条记录
    df = df.drop_duplicates(subset=['MMSI'], keep='last')
    return df

AIS 数据常因通信中断导致轨迹点缺失。线性插值是一种简单有效的修复方法：

import numpy as np

def interpolate_trajectory(df, mmsi, time_col='timestamp', interval='5T'):
    """
    对指定船舶的轨迹进行线性插值
    :param df: 包含轨迹数据的 DataFrame
    :param mmsi: 船舶 MMSI
    :param time_col: 时间列名
    :param interval: 插值时间间隔
    :return: 插值后的轨迹 DataFrame
    """
    # 筛选指定船舶的数据
    vessel_data = df[df['MMSI'] == mmsi].copy()
    vessel_data[time_col] = pd.to_datetime(vessel_data[time_col])
    vessel_data = vessel_data.set_index(time_col).sort_index()

    # 生成完整时间序列
    full_idx = pd.date_range(start=vessel_data.index.min(),
        end=vessel_data.index.max(),
        freq=interval
    )

    # 插值经纬度、航速等数值列
    numeric_cols = ['latitude', 'longitude', 'speed']
    vessel_data = vessel_data.reindex(full_idx)
    vessel_data[numeric_cols] = vessel_data[numeric_cols].interpolate(method='linear')

    # 填充静态属性（插值可能导致这些列变为 NaN）static_cols = ['MMSI', 'vessel_name', 'ship_type']
    for col in static_cols:
        vessel_data[col] = vessel_data[col].ffill().bfill()

    return vessel_data.reset_index()

对于地理空间查询，PostGIS 提供了高效的空间索引支持：

-- 创建包含地理信息的表
CREATE TABLE ais_data (
    mmsi integer,
    timestamp timestamp,
    longitude float,
    latitude float,
    geom geometry(Point, 4326)
);

-- 导入数据后创建空间索引
CREATE INDEX ais_data_geom_idx ON ais_data USING GIST(geom);

-- 查询某区域内的船舶（利用空间索引加速）SELECT mmsi, timestamp 
FROM ais_data 
WHERE ST_Within(
    geom,
    ST_MakeEnvelope(121.0, 30.0, 122.0, 31.0, 4326)
);

当 AIS 数据达到亿级时，pandas 可能遇到内存不足问题。Dask 提供了分布式计算能力：

import dask.dataframe as dd

# 读取大型 Parquet 文件
ddf = dd.read_parquet('large_ais_dataset/*.parquet')

# 执行分组聚合（延迟计算）result = ddf.groupby('MMSI').size().compute()  # 触发实际计算

注意事项：

合理设置分区大小（通常 128MB-1GB 每个分区）
避免跨分区频繁 shuffle 操作
监控集群内存使用情况

对于频繁访问的静态数据（如船舶注册信息），可以使用内存映射减少 I / O 开销：

import numpy as np

# 将船舶静态数据存储为内存映射文件
static_data = pd.DataFrame({'mmsi': [123456789, 987654321],
    'vessel_name': ['VESSEL_A', 'VESSEL_B']
})
static_data.to_pickle('static_data.pkl')

# 后续访问时使用内存映射
mmap_data = pd.read_pickle('static_data.pkl', mmap_mode='r')

AIS 报文中的时间戳通常采用 UTC，但部分数据源可能混用本地时区。建议：

始终明确记录原始数据的时区信息
在 ETL 流程早期统一转换为 UTC
使用 pandas 的 tz_localize 和tz_convert方法而非手工计算

错误示例：

# 错误：直接加减时区偏移（不考虑夏令时）df['timestamp'] = df['timestamp'] + pd.Timedelta(hours=8)

正确做法：

# 正确：使用时区感知转换
if df['timestamp'].dt.tz is None:
    df['timestamp'] = df['timestamp'].dt.tz_localize('Asia/Shanghai')
df['timestamp'] = df['timestamp'].dt.tz_convert('UTC')

船舶静态信息（如船名、类型）可能随时间变化，需要谨慎关联：

避免简单 JOIN 导致信息过时
对静态数据变化建立历史版本记录
查询时使用时间范围约束

-- 正确的时态关联查询
SELECT d.*, s.vessel_name
FROM dynamic_data d
JOIN static_data s ON d.mmsi = s.mmsi 
    AND d.timestamp BETWEEN s.valid_from AND s.valid_to;