动态填充pyspark数据框中列中的行

Posted 2023-04-15

【中文标题】动态填充pyspark数据框中列中的行

【英文标题】：Dynamically fill up rows in a column in pyspark dataframe

【发布时间】：2020-09-29 08:06:03

【问题描述】：

我想要这样的东西：

Id  A  B  Flag  COL
 1  5  4   0     0
 1  5  8   1     1
 1  6  4   0     1
 1  4  7   1     2
 2  7  6   0     0
 2  8  9   1     1
 2  3  8   1     2

我有必须根据 id 进行分区的数据框，并且我有基于条件的标志（A

【问题讨论】：

你的逻辑有缺陷。您没有任何排序规则。如果我改组你的线路，我无法构造回数据框。您需要定义一个列来对您的行进行排序。

嗨史蒂文，逻辑是每当我得到标志为 1 时，我需要对“COL”列的前一行值做 +1 并且每当我得到标志为 0 时，我只需要与相同的值“COL”列的上一行值。

是的，但是如果你的行被打乱了怎么办？ Spark 确实会打乱您的数据，因此与ID=1 的行可能会一个接一个地出现……在纸面上，您的逻辑有效，但不适用于数据帧。请展示您的真实数据，而不是您认为更简单的示例，因为事实并非如此。

史蒂文，我编辑了我的问题。我已经使用了“ID”分区和排序来获得该特定格式。如果现在有任何意义，请告诉我。

更好但仍然无法正常工作。对于 ID=1，行没有任何顺序，因此列 COL 可以根据行的顺序而改变。它可以是0,0,1,2 或1,2,2,2 或任何其他组合。我们需要这个新列和上一个列。

【参考方案1】：

考虑到我制作的不同 cmets，这是我基于有效数据集的解决方案：

from pyspark.sql import functions as F, Window

df.show()  # Without columns parition and order, it is impossible to compute COL

+---------+-----+---+---+
|partition|order|  A|  B|
+---------+-----+---+---+
|        1|    1|  5|  4|
|        1|    2|  5|  8|
|        1|    3|  6|  4|
|        1|    4|  4|  7|
|        2|    1|  7|  6|
|        2|    2|  8|  9|
|        2|    3|  3|  8|
+---------+-----+---+---+

df.withColumn("flag", F.when(F.col("A") < F.col("B"), 1).otherwise(0)).withColumn(
    "COL",
    F.sum("flag").over(
        Window.partitionBy("partition").orderBy(
            "order"
        )  # Window is the reason why we need these two columns
    ),
).show()

+---------+-----+---+---+----+---+
|partition|order|  A|  B|flag|COL|
+---------+-----+---+---+----+---+
|        1|    1|  5|  4|   0|  0|
|        1|    2|  5|  8|   1|  1|
|        1|    3|  6|  4|   0|  1|
|        1|    4|  4|  7|   1|  2|
|        2|    1|  7|  6|   0|  0|
|        2|    2|  8|  9|   1|  1|
|        2|    3|  3|  8|   1|  2|
+---------+-----+---+---+----+---+