如何用PySpark基于前一日数据添加24、25小时动态列？

这个场景我之前也碰到过，你的join写法因为列名冲突和关联逻辑错误触发了笛卡尔积，咱们来一步步解决它：

方法一：正确的自关联（适合有明确Yesterday关联键的场景） #

你之前的核心问题是没给关联的第二个DataFrame设置别名，导致Spark无法区分两个表的列，关联条件也写错了。正确的写法应该给两个表分别起别名，用a.Yesterday关联b.Day：

from pyspark.sql.functions import col

# 给原DF起别名a，关联自身别名b，关联条件是a的Yesterday等于b的Day
df_joined = df.alias("a").join(
    df.alias("b"),
    on=col("a.Yesterday") == col("b.Day"),
    how="left"
)

# 保留原表所有列，把b表的0、1列重命名为24、25
df_final = df_joined.select(
    "a.*",
    col("b.0").alias("24"),
    col("b.1").alias("25")
)

这样每个行只会关联到对应前一天的那一行，不会触发笛卡尔积。比如2012-01-05的"24"列会取到2012-01-04的"0"列值3，"25"列取到2012-01-04的"1"列值null，完全符合你的需求。

方法二：窗口函数Lag（适合日期连续且按顺序排列的场景） #

如果你的Day列是连续按时间排序的，用窗口函数lag会更高效，不需要做关联操作，直接获取前一行的对应列值：

from pyspark.sql import Window
from pyspark.sql.functions import lag

# 按Day升序排列的窗口
window_spec = Window.orderBy("Day")

# 添加24列：取前一行的"0"列值；25列：取前一行的"1"列值
df_final = df.withColumn("24", lag("0", 1).over(window_spec)) \
             .withColumn("25", lag("1", 1).over(window_spec))

这种方法的优势是避免了join操作，性能更好，前提是Day列的顺序和Yesterday对应的前一天完全一致（没有日期缺失）。

为什么你之前的join会报错？ #

你之前的代码里，关联条件写的是df["Yesterday"] == df["Day"]，这是在同一个DataFrame内比较列，而非两个不同表之间的关联。Spark会认为你要把每个行和所有满足Yesterday=Day的行关联，这大概率会产生笛卡尔积，所以触发了AnalysisException。

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