pyspark - 使用 Pyspark 更改重复行的值，保持第一条记录不变

Question

如何更改包含特定列重复记录的行上的列状态值，并将第一个（具有较低 id）保持为 A，例如：

逻辑：

• 如果account_id和user_id已经存在状态为E，第一条记录（较低的id）为A
• 如果 user_id 存在且 account_id 不同则状态为 I，第一条记录（较低的 id）为 A

输入样本：

ID	帐户ID	用户身份
1	一个	1
2	一个	1
3	b	1
4	C	2
5	C	2
6	C	2
7	d	3
8	d	3
9	e	3

输出样本

ID	帐户ID	用户身份	地位
1	一个	1	一个
2	一个	1	乙
3	b	1	我
4	C	2	一个
5	C	2	乙
6	C	2	乙
7	d	3	一个
8	d	3	乙
9	e	3	我

我想我需要组合成多个数据集并将其加入，比较和更改值，我想我想多了，帮助？谢谢！！

谢谢

标签： pyspark

Answer 1

两个窗口函数将帮助您确定重复并对其进行排名。

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

(df
    # Distinguishes between "first occurrence" vs "2nd occurrence" and so on
    .withColumn('rank', F.rank().over(W.partitionBy('account_id', 'user_id').orderBy('id')))
    
    # Detecting if there is no duplication per pair of 'account_id' and 'user_id'
    .withColumn('count', F.count('*').over(W.partitionBy('account_id', 'user_id')))
    
    # building status based on conditions
    .withColumn('status', F
        .when(F.col('count') == 1, 'I') # if there is only one record, status is 'I'
        .when(F.col('rank') == 1, 'A')  # if there is more than one record, the first occurrence is 'A'
        .otherwise('E')                 # finally, the other occurrences are 'E'
    )
    .orderBy('id')
    .show()
)

# Output
# +---+----------+-------+----+-----+------+
# | id|account_id|user_id|rank|count|status|
# +---+----------+-------+----+-----+------+
# |  1|         a|      1|   1|    2|     A|
# |  2|         a|      1|   2|    2|     E|
# |  3|         b|      1|   1|    1|     I|
# |  4|         c|      2|   1|    3|     A|
# |  5|         c|      2|   2|    3|     E|
# |  6|         c|      2|   3|    3|     E|
# |  7|         d|      3|   1|    2|     A|
# |  8|         d|      3|   2|    2|     E|
# |  9|         e|      3|   1|    1|     I|
# +---+----------+-------+----+-----+------+