pyspark 检测分类变量的变化答案

【问题标题】：pyspark detect change of categorical variablepyspark 检测分类变量的变化
【发布时间】：2020-09-02 13:58:53
【问题描述】：

我有一个由两列组成的 spark 数据框。

+-----------------------+-----------+
|                 Metric|Recipe_name|
+-----------------------+-----------+
|                 100.  |       A   |
|                 200.  |       A   |
|                 300.  |       A   |
|                 10.   |       A   |
|                 20.   |       A   |
|                 10.   |       B   |
|                 20.   |       B   |
|                 10.   |       A   |
|                 20.   |       A   |
|                 ..    |      ..   |
|                 ..    |      ..   |
|                 10.   |       B   |

数据框是按时间排序的（您可以想象时间戳列在增加）。我需要添加一列“循环”。当我说一个新的周期开始时，有两种情况：

如果相同的配方正在运行，假设配方“A”，并且 Metric 的值减小（相对于最后一行），则新的循环开始。
假设我们从当前配方“A”切换到第二个配方“B”并切换回配方“A”，我们说配方“A”的新周期已经开始。

所以最后我希望有一个看起来像这样的列“循环”：

+-----------------------+-----------+-----------+
|                 Metric|Recipe_name|      Cycle|
+-----------------------+-----------+-----------+
|                 100.  |       A   |       0   |
|                 200.  |       A   |       0   |
|                 300.  |       A   |       0   |
|                 10.   |       A   |       1   |
|                 20.   |       A   |       1   |
|                 10.   |       B   |       0   |
|                 20.   |       B   |       0   |
|                 10.   |       A   |       2   |
|                 20.   |       A   |       2   |
|                 ..    |      ..   |       2   |
|                 ..    |      ..   |       2   |
|                 10.   |       B   |       1   |

所以这意味着配方A 具有循环0，然后公制减少并且循环更改为1。然后一个新的配方开始B，所以它有一个新的循环0。然后我们再次回到配方A，我们说配方A 开始一个新的循环，并且关于最后一个循环编号，它具有循环2（同样适用于配方B）。

总共有 200 种食谱。

感谢您的帮助。

【问题讨论】：

标签： python-3.x pyspark apache-spark-sql

【解决方案1】：

将我的订单栏替换为您的订单栏。在Recipe_name 列被分区时使用lag 函数比较您的情况。

w = Window.partitionBy('Recipe_name').orderBy('order')

df.withColumn('Cycle', when(col('Metric') < lag('Metric', 1, 0).over(w), 1).otherwise(0)) \
  .withColumn('Cycle', sum('Cycle').over(w)) \
  .orderBy('order') \
  .show()

+------+-----------+-----+
|Metric|Recipe_name|Cycle|
+------+-----------+-----+
|   100|          A|    0|
|   200|          A|    0|
|   300|          A|    0|
|    10|          A|    1|
|    20|          A|    1|
|    10|          B|    0|
|    20|          B|    0|
|    10|          A|    2|
|    20|          A|    2|
|    10|          B|    1|
+------+-----------+-----+

【讨论】：