根据列值的变化对pyspark数据框进行分区

Posted

技术标签:

【中文标题】根据列值的变化对pyspark数据框进行分区【英文标题】:Partition pyspark dataframe based on the change in column value 【发布时间】:2017-10-30 11:50:24 【问题描述】:

我在 pyspark 中有一个数据框。 说有一些列a,b,c ... 我想随着列值的变化将数据分组。说

A  B
1  x
1  y
0  x
0  y
0  x
1  y
1  x
1  y

将有 3 个组为(1x,1y),(0x,0y,0x),(1y,1x,1y) 以及对应的行数据

【问题讨论】:

【参考方案1】:

如果我理解正确,您希望在每次 A 列更改值时创建一个不同的组。

首先,我们将创建一个单调递增的 id 以保持行顺序不变:

import pyspark.sql.functions as psf
df = sc.parallelize([[1,'x'],[1,'y'],[0,'x'],[0,'y'],[0,'x'],[1,'y'],[1,'x'],[1,'y']])\
    .toDF(['A', 'B'])\
    .withColumn("rn", psf.monotonically_increasing_id())
df.show()

    +---+---+----------+
    |  A|  B|        rn|
    +---+---+----------+
    |  1|  x|         0|
    |  1|  y|         1|
    |  0|  x|         2|
    |  0|  y|         3|
    |  0|  x|8589934592|
    |  1|  y|8589934593|
    |  1|  x|8589934594|
    |  1|  y|8589934595|
    +---+---+----------+

现在我们将使用窗口函数来创建一个包含 1 的列,每次 A 列发生变化:

from pyspark.sql import Window
w = Window.orderBy('rn')
df = df.withColumn("changed", (df.A != psf.lag('A', 1, 0).over(w)).cast('int'))

    +---+---+----------+-------+
    |  A|  B|        rn|changed|
    +---+---+----------+-------+
    |  1|  x|         0|      1|
    |  1|  y|         1|      0|
    |  0|  x|         2|      1|
    |  0|  y|         3|      0|
    |  0|  x|8589934592|      0|
    |  1|  y|8589934593|      1|
    |  1|  x|8589934594|      0|
    |  1|  y|8589934595|      0|
    +---+---+----------+-------+

最后我们将使用另一个窗口函数为每个组分配不同的数字:

df = df.withColumn("group_id", psf.sum("changed").over(w)).drop("rn").drop("changed")

    +---+---+--------+
    |  A|  B|group_id|
    +---+---+--------+
    |  1|  x|       1|
    |  1|  y|       1|
    |  0|  x|       2|
    |  0|  y|       2|
    |  0|  x|       2|
    |  1|  y|       3|
    |  1|  x|       3|
    |  1|  y|       3|
    +---+---+--------+

现在你可以建立你的群组了

【讨论】:

谢谢,很有帮助 没问题Jugraj,别忘了把你的问题标记为已解决:)

以上是关于根据列值的变化对pyspark数据框进行分区的主要内容,如果未能解决你的问题,请参考以下文章

根据列值对火花数据框进行分区?

PySpark - 根据另一列值的降序添加递增的整数排名值

python, pyspark : 获取 pyspark 数据框列值的总和

根据值或列表的变化对python数据框进行切片

从具有目标列值的列更新 pyspark 数据框

PySpark DataFrame 根据另一列中时间戳值的最小/最大条件更新列值