PySpark DataFrame按create_date最大值过滤返回空列表问题

先来看你的场景：你有一个包含system_name、file_name、data_tablename和create_date（timestamp类型）的PySpark DataFrame，数据如下：

+-----------+----------------+------------------+--------------------------+
|system_name|file_name       |data_tablename    |create_date               |
+-----------+----------------+------------------+--------------------------+
|ABC        |abc_11132020    |dbo.refine_abc    |2020-11-13 19:34:01.448957|
|ABC        |abc_11132020    |dbo.refine_abc    |2020-11-13 20:44:26.315801|
|ABC        |abc_11162020_1  |dbo.refine_abc    |2020-11-16 20:07:12.354104|
+-----------+----------------+------------------+--------------------------+

你尝试获取最大的create_date然后过滤对应行，但执行后返回了空列表，明明看起来时间戳是匹配的。

问题原因 #

你遇到的核心问题是Spark的TimestampType列和Python原生datetime对象的类型不兼容。当你用collect()把最大时间戳取到本地变成Python datetime后，直接和Spark的Timestamp列用==比较时，Spark无法准确将Python datetime转换为和列中精度完全匹配的Timestamp类型，导致比较失败，最终返回空结果。

几种可行的解决方案 #

方案1：用Spark子查询直接过滤（推荐，避免数据拉到本地）

不要把最大时间戳拉到Python端，直接在Spark的DataFrame层面做关联过滤：

from pyspark.sql.functions import max

# 先获取最大时间戳的临时DataFrame
max_ts_df = df.select(max("create_date").alias("max_create_date"))

# 通过关联过滤出匹配的行
file_details = df.join(max_ts_df, df.create_date == max_ts_df.max_create_date) \
                 .drop("max_create_date") \
                 .collect()

方案2：用窗口函数获取最新行

如果后续需要扩展到按分组（比如按system_name）取最新行，窗口函数会更灵活：

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, desc

# 定义全局排序的窗口（如果要分组，就加partitionBy("system_name")）
window_spec = Window.orderBy(desc("create_date"))

# 给每行加行号，最新的行号为1
file_details = df.withColumn("row_num", row_number().over(window_spec)) \
                 .filter("row_num == 1") \
                 .drop("row_num") \
                 .collect()

方案3：把Python datetime转为Spark字面量（如果一定要用本地的时间戳）

如果你必须把时间戳拉到Python端处理，记得用lit()把Python datetime转为Spark能识别的Timestamp字面量：

from pyspark.sql.functions import lit, max

max_ts = df.select(max("create_date")).collect()[0][0]
# 用lit()包裹时间戳，让Spark正确识别类型
file_details = df.filter(df.create_date == lit(max_ts)).collect()

这样就能正确过滤出file_name为abc_11162020_1的行了。

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