优化船舶目的地预测ML数据集生成代码

API 探路者

2026-7-29

船舶目的地预测代码优化求助

我是统计学家，正在开展船舶目的地预测任务，当前代码运行速度极慢，自身难以定位优化方向，希望得到帮助。我在Databricks环境运行，可考虑采用PySpark方案。

数据说明

待处理Pandas数据框表头

index   imo     fromporta toportb   trajectory                             trajectory_length    voyage_id
0       8005025 port1 port2 [(23, 59.8133333, 2022-02-11 23:53:...)]        9                   1
1       8005025 port1 port2 [(23.425, 59.81333, 2022-02-11 23:53:...)]      18                  1
2       8005025 port1 port2[(23.425, 59.813333, 2022-02-11 23:53:...)]      27                  1
3       4252666 port3 port2 [(17.63, 83.38, 2022-03-08 23:51:45), (1...)]   13                  2
4       4252666 port3 port2[(17.63, 83.38, 2022-03-08 23:51:45), (1...)]    26                  2

字段说明

imo：船舶标识（整数）
fromporta：出发港口（字符串）
toportb：到达港口（字符串）
trajectory：元组列表，每个元组包含(lat, lon, timestamp)
voyage_id：唯一航次的整数标识

预期输出

index   imo     fromporta toportb traj_length pred_toportb predicted_traj_length similarity(distance)   voyage_id
0       8005025 port1    port2       9               port75                    18           56.7     1
1       8005025 port1    port2       18              port80                    31           41.4    1
2       8005025 port1    port2       27              port2                     25           1.5     1
3       4252666 port3    port2       13              port5                     10           101.51    2 
4       4252666 port3    port2       26              port7                     18           65.6     2

现有代码逻辑及实现

核心逻辑：将每个航次与同出发港口、不同voyage_id的其他航次对比，通过traj-dist库计算欧几里得SSPD距离，找到最相似航次以预测目的地。

代码实现：

def calculate_fast_sspd_similarity(sampled_voyage, historical_voyage):
 traj1 = np.array([(lat, lon) for lat, lon, timestamp, draught in sampled_voyage])
 traj2 = np.array([(lat, lon) for lat, lon, timestamp, draught in historical_voyage])
 return tdist.sspd(traj1,traj2) #can add spherical here


def find_most_similar(sampled_voyage, historical_voyages, similarity_function):

 similarities = []
 for i, row in historical_voyages.iterrows():
   historical_voyage = row['trajectory']
   sim = similarity_function(sampled_voyage, historical_voyage)
   similarities.append(sim)
   #Choose lowest value
 
 most_similar_index = np.argmin(similarities)

 return most_similar_index, np.min(similarities)
        
def similarity_measure1(df, similarity_function):
 """Creates new df with similarity measure for each voyage"""
 result = []
 count=0
 for _, row in df.iterrows():

   count+=1
   sampled_voyage = row['trajectory']
   historical_voyages = df[(df['fromporta'] == row['fromporta']) & (df['voyage_id'] != row['voyage_id'])]
   
   if len(historical_voyages)==0:
     continue
     
   most_similar_index, similarity = find_most_similar(sampled_voyage, historical_voyages, similarity_function)
       
   most_similar = historical_voyages.iloc[most_similar_index]
       
   result.append({
           'imo': row['imo'],
           'fromporta': row['fromporta'], 
           'toportb': row['toportb'],
           'trajectory_length': row['trajectory_length'],
           'predicted_toportb': most_similar['toportb'],
           'predicted_trajectory_length': most_similar['trajectory_length'],
           'similarity': similarity
       })

 result_df = pd.DataFrame(result)
 return result_df

运行方式：

final_ml_df = similarity_measure1(ml_df, calculate_fast_sspd_similarity)

优化方案

一、Pandas代码优化（无需切换环境）

提前分组预处理，减少重复筛选
按fromporta分组，提前提取每个组内的轨迹、目的地、轨迹长度等数据，避免每次循环重复执行数据筛选操作：

import numpy as np
import pandas as pd
import traj_dist as tdist

# 预处理每个出发港口的历史数据
port_groups = df.groupby('fromporta').apply(
    lambda g: {
        'trajectories': np.array([np.array([(lat, lon) for lat, lon, ts, dr in traj]) for traj in g['trajectory']]),
        'toportbs': g['toportb'].values,
        'traj_lengths': g['trajectory_length'].values,
        'voyage_ids': g['voyage_id'].values
    }
).to_dict()

def calculate_sspd_batch(sampled_traj, historical_trajs):
    return np.array([tdist.sspd(sampled_traj, traj) for traj in historical_trajs])

def similarity_measure_optimized(df):
    result = []
    for _, row in df.iterrows():
        port_data = port_groups.get(row['fromporta'])
        if not port_data:
            continue
        # 筛选不同voyage_id的历史数据
        mask = port_data['voyage_ids'] != row['voyage_id']
        if not mask.any():
            continue
        filtered_trajs = port_data['trajectories'][mask]
        filtered_toports = port_data['toportbs'][mask]
        filtered_lengths = port_data['traj_lengths'][mask]
        # 转换当前轨迹为numpy数组
        sampled_traj = np.array([(lat, lon) for lat, lon, ts, dr in row['trajectory']])
        # 批量计算相似度
        similarities = calculate_sspd_batch(sampled_traj, filtered_trajs)
        min_idx = np.argmin(similarities)
        result.append({
            'imo': row['imo'],
            'fromporta': row['fromporta'],
            'toportb': row['toportb'],
            'trajectory_length': row['trajectory_length'],
            'predicted_toportb': filtered_toports[min_idx],
            'predicted_trajectory_length': filtered_lengths[min_idx],
            'similarity': similarities[min_idx],
            'voyage_id': row['voyage_id']
        })
    return pd.DataFrame(result)

用Numba加速SSPD计算
若traj-dist的SSPD函数未优化，可手动实现并通过Numba装饰加速：

from numba import jit

@jit(nopython=True)
def sspd_numba(traj1, traj2):
    min_dist = np.inf
    for p1 in traj1:
        for p2 in traj2:
            dist = np.sqrt((p1[0]-p2[0])**2 + (p1[1]-p2[1])**2)
            if dist < min_dist:
                min_dist = dist
    return min_dist

替换原代码中的tdist.sspd调用即可提升单条轨迹对的计算速度。

改用分组Apply减少循环层级
将同港口的处理逻辑封装，通过groupby.apply批量处理：

def process_port_group(group):
    group_results = []
    trajs = np.array([np.array([(lat, lon) for lat, lon, ts, dr in traj]) for traj in group['trajectory']])
    toportbs = group['toportb'].values
    lengths = group['trajectory_length'].values
    voyage_ids = group['voyage_id'].values
    for i, row in group.iterrows():
        mask = voyage_ids != row['voyage_id']
        if not mask.any():
            continue
        filtered_trajs = trajs[mask]
        filtered_toports = toportbs[mask]
        filtered_lengths = lengths[mask]
        sampled_traj = trajs[i]
        similarities = np.array([tdist.sspd(sampled_traj, t) for t in filtered_trajs])
        min_idx = np.argmin(similarities)
        group_results.append({
            'imo': row['imo'],
            'fromporta': row['fromporta'],
            'toportb': row['toportb'],
            'trajectory_length': row['trajectory_length'],
            'predicted_toportb': filtered_toports[min_idx],
            'predicted_trajectory_length': filtered_lengths[min_idx],
            'similarity': similarities[min_idx],
            'voyage_id': row['voyage_id']
        })
    return pd.DataFrame(group_results)

final_ml_df = df.groupby('fromporta').apply(process_port_group).reset_index(drop=True)

二、PySpark方案（适配Databricks大数据场景）

数据格式转换
将Pandas数据转为Spark DataFrame，并提取轨迹中的经纬度：

from pyspark.sql import SparkSession
from pyspark.sql.functions import udf, col
from pyspark.sql.types import ArrayType, DoubleType, StructType, StructField

spark = SparkSession.builder.appName("VoyageSimilarity").getOrCreate()
spark_df = spark.createDataFrame(ml_df)

# 提取轨迹中的经纬度对
extract_lat_lon = udf(
    lambda traj: [(float(p[0]), float(p[1])) for p in traj],
    ArrayType(StructType([StructField("lat", DoubleType()), StructField("lon", DoubleType())]))
)
spark_df = spark_df.withColumn("traj_latlon", extract_lat_lon(col("trajectory")))

广播历史数据，减少Shuffle
按fromporta广播历史航次数据，避免全量笛卡尔积关联：

from pyspark.sql.functions import broadcast, collect_list, struct

# 收集每个港口的历史航次数据
port_history = spark_df.groupBy("fromporta").agg(
    collect_list(struct(
        col("traj_latlon").alias("hist_traj"),
        col("toportb").alias("hist_toport"),
        col("trajectory_length").alias("hist_length"),
        col("voyage_id").alias("hist_voyage_id")
    )).alias("history")
)

# 广播历史数据到所有节点
broadcast_port_history = broadcast(port_history)
joined_df = spark_df.join(broadcast_port_history, on="fromporta", how="left")

分布式匹配UDF实现
编写UDF计算当前轨迹与历史轨迹的SSPD，找到最相似航次：

import traj_dist as tdist
import numpy as np
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType

def find_best_match(traj, voyage_id, history):
    if not history:
        return (None, None, None)
    min_dist = float('inf')
    best_toport = None
    best_length = None
    for hist_entry in history:
        if hist_entry["hist_voyage_id"] == voyage_id:
            continue
        traj_np = np.array([(p["lat"], p["lon"]) for p in traj])
        hist_traj_np = np.array([(p["lat"], p["lon"]) for p in hist_entry["hist_traj"]])
        dist = tdist.sspd(traj_np, hist_traj_np)
        if dist < min_dist:
            min_dist = dist
            best_toport = hist_entry["hist_toport"]
            best_length = hist_entry["hist_length"]
    return (best_toport, best_length, min_dist)

match_schema = StructType([
    StructField("predicted_toportb", StringType()),
    StructField("predicted_trajectory_length", IntegerType()),
    StructField("similarity", DoubleType())
])
find_best_match_udf = udf(find_best_match, match_schema)

# 生成结果DataFrame
result_spark_df = joined_df.withColumn(
    "match_result",
    find_best_match_udf(col("traj_latlon"), col("voyage_id"), col("history"))
).select(
    col("imo"),
    col("fromporta"),
    col("toportb"),
    col("trajectory_length").alias("traj_length"),
    col("match_result.predicted_toportb"),
    col("match_result.predicted_trajectory_length"),
    col("match_result.similarity").alias("similarity(distance)"),
    col("voyage_id")
).filter(col("predicted_toportb").isNotNull())

批量处理优化：Pandas UDF
用Pandas UDF批量处理每组数据，减少UDF调用开销：

from pyspark.sql.functions import pandas_udf
from pyspark.sql.types import DataFrameType

output_schema = StructType([
    StructField("imo", IntegerType()),
    StructField("fromporta", StringType()),
    StructField("toportb", StringType()),
    StructField("traj_length", IntegerType()),
    StructField("predicted_toportb", StringType()),
    StructField("predicted_trajectory_length", IntegerType()),
    StructField("similarity(distance)", DoubleType()),
    StructField("voyage_id", IntegerType())
])

@pandas_udf(output_schema)
def process_port_batch(df):
    port = df['fromporta'].iloc[0]
    # 提取当前批次的轨迹和属性
    current_trajs = df['traj_latlon'].apply(lambda x: np.array([(p['lat'], p['lon']) for p in x]))
    current_voyage_ids = df['voyage_id'].values
    # 获取同港口的历史数据（排除当前批次的航次）
    history_df = spark_df.filter((col("fromporta") == port) & (~col("voyage_id").isin(current_voyage_ids))).toPandas()
    if history_df.empty:
        return df[[]]
    hist_trajs = history_df['traj_latlon'].apply(lambda x: np.array([(p['lat'], p['lon']) for p in x]))
    hist_toports = history_df['toportb'].values
    hist_lengths = history_df['trajectory_length'].values
    hist_voyage_ids = history_df['voyage_id'].values

    results = []
    for idx, row in df.iterrows():
        traj = current_trajs.iloc[idx]
        voyage_id = current_voyage_ids[idx]
        mask = hist_voyage_ids != voyage_id
        if not mask.any():
            continue
        filtered_trajs = hist_trajs[mask]
        filtered_toports = hist_toports[mask]
        filtered_lengths = hist_lengths[mask]
        similarities = np.array([tdist.sspd(traj, t) for t in filtered_trajs])
        min_idx = np.argmin(similarities)
        results.append({
            'imo': row['imo'],
            'fromporta': port,
            'toportb': row['toportb'],
            'traj_length': row['trajectory_length'],
            'predicted_toportb': filtered_toports[min_idx],
            'predicted_trajectory_length': filtered_lengths[min_idx],
            'similarity(distance)': similarities[min_idx],
            'voyage_id': voyage_id
        })
    return pd.DataFrame(results)

final_spark_df = spark_df.groupBy("fromporta").apply(process_port_batch)

三、通用优化建议

提前转换轨迹格式：将所有trajectory列预转为numpy数组，避免重复转换开销。
过滤无效数据：提前删除trajectory为空的行

优化船舶目的地预测ML数据集生成代码

数据说明 #

待处理Pandas数据框表头 #

字段说明 #

预期输出 #

现有代码逻辑及实现 #

优化方案 #

一、Pandas代码优化（无需切换环境） #

二、PySpark方案（适配Databricks大数据场景） #

三、通用优化建议 #