Spark中比较两个DataFrame日期触发TypeError的问题排查

错误原因 #

你直接拿两个Spark DataFrame对象做大小比较，这是不允许的。spark.sql()返回的是DataFrame（分布式数据集），不是单个日期值，Python无法直接对两个DataFrame执行>或<这类比较操作，因此抛出TypeError。

解决思路 #

必须从DataFrame中提取出单个日期标量值，再进行比较。另外，你用DISTINCT tm_update来获取表2的日期并不严谨——表2每周更新，可能存在多个时间戳，应该取最新的（即MAX(tm_update)），这样才能准确判断是否达到90天的触发条件。

修改后的代码 #

# 计算表1最大更新日期加90天，提取单个值
futr_date_df = spark.sql('SELECT date_add(MAX(tm_update), 90) AS future_date FROM tbl_one')
futr_date_val = futr_date_df.select("future_date").first()[0]

# 获取表2最新的更新时间戳，提取单个值
curr_date_df = spark.sql('SELECT MAX(tm_update) AS current_date FROM tbl_two')
curr_date_val = curr_date_df.select("current_date").first()[0]

# 处理表1为空的边界情况（首次运行）
if futr_date_val is None:
    print('首次运行，执行表2数据转换并追加到表1的代码块')
elif curr_date_val > futr_date_val:
    print('执行表2数据转换并追加到表1的代码块')
else:
    print('忽略，下周再检查')

关键说明 #

• first()[0]：从DataFrame的第一行中取出第一个字段的值，将DataFrame转换为单个日期对象
• 用MAX(tm_update)替代DISTINCT tm_update：确保拿到表2最新的更新时间，避免因存在多个历史时间戳导致判断错误
• 增加空值判断：处理表1初始为空的场景，保证首次运行能正常执行追加逻辑

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