Spark：如何编写函数实现多历史年份DataFrame关联合并

嗨，作为Spark新手，完全不用手动重复写10次SQL这么麻烦！我给你两种实用的方案，既能实现需求，又能让代码更简洁易维护：

这种方法会先把history表转换成宽表（每个key对应10个历史年份的value列），再和df1做一次关联，比多次join效率更高。

from pyspark.sql import functions as F

def get_10_year_history(df1, history_df, current_date):
    # 把history处理成宽表：每个key对应10个历史年份的value
    history_wide = history_df.groupBy("key").agg(
        *[
            # 筛选出对应年份的date，取对应的value，无数据时留空
            F.first(
                F.when(F.col("date") == F.add_months(F.lit(current_date), -12*i), F.col("value")),
                ignorenulls=True
            ).alias(f"previous_year_{i}") 
            for i in range(1, 11)
        ]
    )
    
    # 关联df1和宽表，保留df1所有数据（左关联）
    result_df = df1.join(history_wide, on="key", how="left")
    return result_df

# 调用示例：假设当前日期是'2024-01-01'，你也可以用F.current_date()取系统当前日期
current_date = '2024-01-01'
final_df = get_10_year_history(df1, history, current_date)
final_df.show()

代码说明： #

• 用groupBy("key")聚合每个key的历史数据
• 循环生成10个条件判断：检查history的date是否是current_date往前推12*i个月的日期，匹配的话取对应的value
• first(ignorenulls=True)确保如果同一个key对应多个相同日期的value，只取第一个（如果需要平均值/总和，换成avg()/sum()即可）
• 最后用左关联保留df1的所有行，即使某个key在某年份没有历史数据，对应字段会显示null

如果你更熟悉SQL语法，可以写个函数自动生成包含10次关联的SQL语句，不用手动复制粘贴：

def generate_history_sql(df1_view_name, history_view_name, current_date, year_count=10):
    # 动态生成join子句和查询字段
    join_clauses = []
    select_fields = []
    for i in range(1, year_count+1):
        alias = f"h{i}"
        # 每个年份对应一个左关联
        join_clauses.append(f"""
            LEFT JOIN {history_view_name} {alias} 
            ON df1.key = {alias}.key 
            AND {alias}.date = add_months('{current_date}', {-12*i})
        """)
        # 生成对应的字段别名
        select_fields.append(f"{alias}.value AS previous_year_{i}")
    
    # 拼接完整SQL
    sql_query = f"""
        SELECT df1.*, {', '.join(select_fields)}
        FROM {df1_view_name} df1
        {' '.join(join_clauses)}
    """
    return sql_query

# 先把DataFrame注册成临时视图
df1.createOrReplaceTempView("df1")
history.createOrReplaceTempView("history")

# 生成SQL并执行
current_date = '2024-01-01'
sql_query = generate_history_sql("df1", "history", current_date)
final_df = spark.sql(sql_query)
final_df.show()

代码说明： #

• 先把两个DataFrame注册成临时视图，这样Spark SQL能访问到它们
• 循环生成10个LEFT JOIN子句，每个对应一个历史年份的关联条件
• 最后拼接成完整的SQL语句，执行后就能得到包含10个历史值列的结果

1. 如果你想用系统当前日期，直接把F.lit(current_date)换成F.current_date()（DataFrame方案），或者把'{current_date}'换成current_date()（SQL方案）即可
2. 如果history中同一个key+日期有多个记录，根据你的业务需求替换聚合函数（比如avg()、sum()）
3. 用LEFT JOIN而非INNER JOIN，确保df1的所有数据都能保留，即使没有匹配的历史记录

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