根据列值的变化对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数据框进行分区的主要内容,如果未能解决你的问题,请参考以下文章