Spark分区DataFrame获取全局时间戳最值错误问题排查

我正在开发一个用于时序分析的Python仓库，里面包含各类时序转换函数。近期需要处理一份近2TB的大数据集，该数据集是通过以下Spark代码写入的：

merged = spark.createDataFrame(df_data.rdd, union_schema) 
merged_output.write_dataframe(
    merged.repartitionByRange('data_id').sortWithinPartitions('data_id', 'timestamp'),
    output_format='soho',
    options={'noho': 'true'}
)

预览数据集时发现它被分割成了50多亿个文件。在将其作为转换函数输入时，我最初尝试用以下代码获取全局最小、最大时间戳：

start_time_input_df = input_df.select('timestamp').sort(F.col('timestamp')).head(1)[0][0]
end_time_input_df = input_df.select('timestamp').sort(F.col('timestamp').desc()).head(1)[0][0]

但运行后得到的结果错误——仅获取到单个分区/文件的最值，而通过SQL预览工具能得到正确的全局最值。

最终我通过Spark的聚合函数解决了这个问题，代码如下：

result = input_df.agg(F.min("timestamp").alias("min_timestamp"), F.max("timestamp").alias("max_timestamp")).collect()[0] 
min_timestamp = result["min_timestamp"] 
max_timestamp = result["max_timestamp"] 

问题原因说明 #

Spark的sort操作默认仅在单个分区内完成排序，head(1)只会返回第一个分区排序后的第一条数据，无法覆盖所有分区实现全局范围的最值查找。而agg配合min/max是全局聚合操作，会跨所有分区计算出真正的全局最值。

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