如何将SAS自内连接代码转换为PySpark语法（日期处理问题）

你贴的这段SAS SQL没有特殊的日期处理黑魔法，核心逻辑非常明确：按var1分组取每组最大日期值，再和主表a内连接，筛出主表中每个var1对应最新日期的记录。

逻辑拆解 #

• 子查询部分：对源表table按var1分组，计算每组date字段的最大值，命名为max_date，结果集别名recent
• 关联部分：主表a和recent做内连接，匹配条件为var1相等、主表date等于分组算出的max_date

等效PySpark实现 #

写法1：Spark SQL 语法（和原SAS写法完全对齐） #

Spark SQL原生兼容这类标准聚合关联语法，几乎可以直接平移原代码运行：

result = spark.sql("""
SELECT a.*
FROM a
INNER JOIN (
    SELECT var1, max(date) as max_date
    FROM table
    GROUP BY var1
) AS recent
ON a.var1 = recent.var1 AND a.date = recent.max_date
""")

写法2：DataFrame API 写法（PySpark日常开发常用） #

from pyspark.sql import functions as F

# 第一步：计算每个var1对应的最大日期
recent_df = table.groupBy("var1").agg(F.max("date").alias("max_date"))

# 第二步：和主表做内连接
result_df = a.alias("a").join(
    recent_df.alias("recent"),
    on=(
        (F.col("a.var1") == F.col("recent.var1")) &
        (F.col("a.date") == F.col("recent.max_date"))
    ),
    how="inner"
).drop("max_date") # 冗余的关联字段可按需删除

写法3：窗口函数写法（同表场景下性能更优） #

如果子查询里的table和主表a是同一张表，可以直接用窗口函数省掉join步骤，执行效率更高：

from pyspark.sql import functions as F
from pyspark.sql.window import Window

# 按var1分区，按日期倒序打行号，行号为1的就是每组最新日期的记录
win_spec = Window.partitionBy("var1").orderBy(F.col("date").desc())
result_df = a.withColumn("row_num", F.row_number().over(win_spec)) \
             .filter(F.col("row_num") == 1) \
             .drop("row_num")

注意事项 #

• 只要保证关联的两个date字段类型一致（同为DateType/TimestampType，不要一个是日期类型一个是字符串），就不需要额外做日期格式转换，原逻辑没有对日期做截断、偏移等特殊处理
• 如果存在同一个var1下多条记录date都等于最大日期的情况，上述写法会把所有匹配的记录都返回，和原SAS逻辑行为完全一致

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