PySpark中将DataFrame按时间戳分组扁平化输出单组单行的最高效方法

最优实现方案：按时间戳分组+固定维度Pivot转宽表 #

这个方案是你当前场景下效率最高、逻辑最简洁的实现，全程走Spark原生JVM运算，无额外冗余扫描和重复计算：

第一步：单时间戳转单行宽表 #

因为你的飞机编号固定为1-5，直接用groupBy+指定取值的pivot即可直接生成每行对应一个时间戳、所有飞机参数分列存储的宽表：

import pyspark.sql.functions as F

# *显式指定pivot的取值为[1,2,3,4,5]是性能优化核心，避免Spark额外扫描全表探测维度值，大幅降低开销*
df_wide = dfAirplaneData.groupBy("utc_time") \
    .pivot("plane_num", values=[1,2,3,4,5]) \
    .agg(
        F.first("lat").alias("lat"),
        F.first("lon").alias("lon"),
        F.first("height").alias("height")
    )

执行后生成的表结构为：utc_time、1_lat、1_lon、1_height、2_lat、2_lon、2_height……5_lat、5_lon、5_height，完全符合你单时间戳存所有飞机信息的需求。

第二步：自定义运算（含飞机间距离计算） #

直接在宽表基础上用Spark原生函数写运算逻辑即可，性能远高于Python UDF/Pandas UDF，以1号和2号飞机的球面距离计算为例（结果单位为公里）：

EARTH_RADIUS_KM = 6371.0

df_result = df_wide.withColumn(
    "distance_plane1_plane2",
    EARTH_RADIUS_KM * 2 * F.asin(
        F.sqrt(
            F.sin(F.radians(F.col("2_lat") - F.col("1_lat")) / 2) ** 2
            + F.cos(F.radians(F.col("1_lat"))) * F.cos(F.radians(F.col("2_lat")))
            * F.sin(F.radians(F.col("2_lon") - F.col("1_lon")) / 2) ** 2
        )
    )
)
# 其余两两飞机的距离计算可参照上述逻辑批量生成

其他方案的劣势对比 #

• 多次过滤自连接方案：需要多次扫描全表，5架飞机至少要扫描4次原表，超大规模数据下开销会成倍增长，完全不推荐。
• 开窗计算方案：会对同时间戳分组内的5行数据重复执行相同计算逻辑，最终还要做去重过滤，存在明显的算力浪费，逻辑实现也更繁琐。
• Pandas UDF分组映射方案：存在Python进程和JVM进程之间的序列化/反序列化额外开销，仅适合Spark原生函数无法实现的复杂自定义逻辑场景，你的需求用原生函数即可实现，不需要用该方案。

内容的提问来源于stack exchange，提问作者WIT