PySpark优雅实现多次自连接改列名及获取站点上年PRCP值

你遇到的解析错误大概率是自连接SQL语句中表别名、字段引用的语法错误导致，以下是两种可行的实现方式：

方式一：使用DataFrame API实现（语法容错率更高） #

# 假设你已读取SQL Server数据得到初始DataFrame: df
from pyspark.sql.functions import col

# 分别给当前表和上一年数据表设置别名
df_current = df.alias("current")
df_prev = df.alias("prev")

# 执行左自连接，关联条件为站点ID相同，当前年 = 上年 + 1
df_result = df_current.join(
    df_prev,
    (col("current.StationId") == col("prev.StationId")) & (col("current.Year") == col("prev.Year") + 1),
    how="left"  # 左连接保留没有上年数据的站点记录
).select(
    "current.StationId",
    "current.Year",
    "current.Month",
    "current.PRCP",
    col("prev.PRCP").alias("prev_PRCP")
)

方式二：使用Spark SQL实现 #

# 先将DataFrame注册为临时视图
df.createOrReplaceTempView("station_prcp")

# 编写自连接SQL，注意必须给两个表设置不同别名
sql_str = """
SELECT 
    a.StationId,
    a.Year,
    a.Month,
    a.PRCP,
    b.PRCP as prev_PRCP
FROM station_prcp a
LEFT JOIN station_prcp b
ON a.StationId = b.StationId AND a.Year = b.Year + 1
"""
df_result = spark.sql(sql_str)

注意事项 #

• 如果同一站点同一年存在多条月度数据，要获取上年同月的PRCP，需将Month字段也加入关联条件：AND a.Month = b.Month
• 如果需要按年聚合PRCP（如年累计降水量），需要先做分组聚合再执行自连接，避免关联后生成多余重复数据

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