如何使用 PySpark 逐年获得百分比变化答案

【问题标题】：How to get percent change year over year by group with PySpark如何使用 PySpark 逐年获得百分比变化
【发布时间】：2021-11-25 15:55:32
【问题描述】：

Pyspark 初学者并尝试通过对产品进行分组来计算产品数量的同比百分比变化。

我有这个数据框

    prod_desc  year  prod_count
0  product_a  2019          53
1  product_b  2019          44
2  product_c  2019          36
3  product_a  2020          52
4  product_b  2020          43
5  product_c  2020          42

需要的输出：

     prod_desc  year  prod_count  Percentage change
0  product_a  2019          53                NaN
1  product_b  2019          44                NaN
2  product_c  2019          36                NaN
3  product_a  2020          52              -1.88
4  product_b  2020          43              -2.27
5  product_c  2020          42              16.60

我可以在 python 中使用以下逻辑来做到这一点 pandas 需要使用 pyspark 来实现同样的目标

ds['percentage change'] = ds.sort_values('year').groupby(['prod_desc']).agg({'prod_count':'pct_change'})

任何帮助将不胜感激

【问题讨论】：

标签： python-3.x pyspark

【解决方案1】：

你的 DF：

df = spark.createDataFrame(
  [
     (0, 'product_a', 2019, 53)
    ,(1, 'product_b', 2019, 44)
    ,(2, 'product_c', 2019, 36)
    ,(3, 'product_a', 2020, 52)
    ,(4, 'product_b', 2020, 43)
    ,(5, 'product_c', 2020, 42)
  ], ['id', 'prod_desc', 'year', 'prod_count']
)

您可以使用带有滞后函数的窗口函数：

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

W = Window.partitionBy("prod_desc").orderBy('year')

df.withColumn('prod_count_y-1',F.lag(df['prod_count']).over(W))\
  .withColumn('var %', F.round((F.col('prod_count')/F.col('prod_count_y-1') -1)*100,2))\
  .show()

+---+---------+----+----------+--------------+-----+
| id|prod_desc|year|prod_count|prod_count_y-1|var %|
+---+---------+----+----------+--------------+-----+
|  0|product_a|2019|        53|          null| null|
|  3|product_a|2020|        52|            53|-1.89|
|  1|product_b|2019|        44|          null| null|
|  4|product_b|2020|        43|            44|-2.27|
|  2|product_c|2019|        36|          null| null|
|  5|product_c|2020|        42|            36|16.67|
+---+---------+----+----------+--------------+-----+

【讨论】：

【解决方案2】：

这是一个使用Window 和lag 的示例来跟踪先前的值并计算百分比变化：

from pyspark.sql import SparkSession, Window
from pyspark.sql import functions as F

spark = SparkSession.builder.getOrCreate()
data = [
    {"prod_desc": "product_a", "year": 2019, "prod_count": 53},
    {"prod_desc": "product_b", "year": 2019, "prod_count": 44},
    {"prod_desc": "product_c", "year": 2019, "prod_count": 36},
    {"prod_desc": "product_a", "year": 2020, "prod_count": 52},
    {"prod_desc": "product_b", "year": 2020, "prod_count": 43},
    {"prod_desc": "product_c", "year": 2020, "prod_count": 42},
]

df = spark.createDataFrame(data)
window = Window.partitionBy("prod_desc").orderBy("year")
df = df.withColumn("prev_value", F.lag(df.prod_count).over(window))
df = (
    df.withColumn(
        "Percentage change",
        F.when(F.isnull(df.prod_count - df.prev_value), None).otherwise(
            (df.prod_count - df.prev_value) * 100 / df.prev_value
        ),
    )
    .drop("prev_value")
    .orderBy("year", "prod_desc")
)

结果：

+----------+---------+----+-------------------+
|prod_count|prod_desc|year|Percentage change  |
+----------+---------+----+-------------------+
|53        |product_a|2019|null               |
|44        |product_b|2019|null               |
|36        |product_c|2019|null               |
|52        |product_a|2020|-1.8867924528301887|
|43        |product_b|2020|-2.272727272727273 |
|42        |product_c|2020|16.666666666666668 |
+----------+---------+----+-------------------+

【讨论】：