PySpark DataFrame计算本年与前两年销售额差值的实现方案

核心思路 #

要计算本年与上年、上年与前年的销售额差值，核心是用Window结合lag函数获取前序年份的销售额数据，再通过条件判断生成目标列（无对应前序年份时设为0）。

具体代码实现 #

假设你的DataFrame包含sdan8、sdan8_name、year（年份列，需为可排序的数值类型）、_prc_net_sales（净销售额列）这些字段：

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import col, lag, when, lit

# 初始化SparkSession（如果未初始化）
spark = SparkSession.builder.appName("SalesDiffCalc").getOrCreate()

# 示例原始数据
data = [
    ("A001", "产品A", 2021, 10000),
    ("A001", "产品A", 2022, 12000),
    ("A001", "产品A", 2023, 15000),
    ("B001", "产品B", 2022, 8000),
    ("B001", "产品B", 2023, 9500)
]
df = spark.createDataFrame(data, ["sdan8", "sdan8_name", "year", "_prc_net_sales"])

# 定义窗口：按sdan8分区，按年份升序排序
window_spec = Window.partitionBy("sdan8").orderBy("year")

# 计算目标列
result_df = df.withColumn(
    "prev_year_sales", lag(col("_prc_net_sales"), 1).over(window_spec)
).withColumn(
    "pre_prev_year_sales", lag(col("_prc_net_sales"), 2).over(window_spec)
).withColumn(
    "_dif_sales_cur_prev",
    when(col("prev_year_sales").isNotNull(), col("_prc_net_sales") - col("prev_year_sales")).otherwise(lit(0))
).withColumn(
    "_dif_sales_cur_prev_prec",
    when(col("pre_prev_year_sales").isNotNull(), col("prev_year_sales") - col("pre_prev_year_sales")).otherwise(lit(0))
).drop("prev_year_sales", "pre_prev_year_sales")  # 删除临时辅助列

# 展示结果
result_df.show()

示例输出 #

+-----+-----------+----+-------------+-------------------+-------------------------+
|sdan8|sdan8_name |year|_prc_net_sales|_dif_sales_cur_prev|_dif_sales_cur_prev_prec|
+-----+-----------+----+-------------+-------------------+-------------------------+
|A001 |产品A      |2021|10000        |0                  |0                        |
|A001 |产品A      |2022|12000        |2000               |0                        |
|A001 |产品A      |2023|15000        |3000               |2000                     |
|B001 |产品B      |2022|8000         |0                  |0                        |
|B001 |产品B      |2023|9500         |1500               |0                        |
+-----+-----------+----+-------------+-------------------+-------------------------+

关键说明 #

• 窗口必须按sdan8分区，确保每个产品的年份数据独立计算；按year排序保证lag能正确获取前序年份的值。
• 用when判断前序年份销售额是否存在（即是否有上一年/前年数据），不存在时设为0，符合需求中的“其余场景为0”。
• 如果你的年份列不是year，替换为实际列名即可；若年份是字符串类型，需先转换为数值类型再排序。

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