PySpark窗口函数标记满足特定条件的每个分区的第一行。

Question

考虑到这个数据

+--------+------+----------+--------+
|app_id  |order |entry_flag|operator|
+--------+------+----------+--------+
|AP-1    |1     |1         |S       |
|AP-1    |2     |0         |A       |
|AP-2    |3     |0         |S       |
|AP-2    |4     |0         |A       |
|AP-2    |5     |1         |S       |
|AP-2    |6     |0         |S       |
|AP-2    |7     |0         |A       |
|AP-2    |8     |0         |A       |
|AP-2    |9     |1         |A       |
|AP-2    |10    |0         |S       |
+--------+------+----------+--------+

我想添加一个新列flag_x，它是布尔型的，逻辑是：

按app_id分区/分组，按order排序，当我们遇到entry_flag = 1行时，向前移动，在它后面找到第一行，其中包含entry_flag = 0 and operator = A，标记flag_x = 1，否则是flag_x = 0

对于上面的样本，输出应该是：

+--------+------+----------+--------+------+
|app_id  |order |entry_flag|operator|flag_x|
+--------+------+----------+--------+------+
|AP-1    |1     |1         |S       |0     |
|AP-1    |2     |0         |A       |1     |
|AP-2    |3     |0         |S       |0     |
|AP-2    |4     |0         |A       |0     |
|AP-2    |5     |1         |S       |0     |
|AP-2    |6     |0         |S       |0     |
|AP-2    |7     |0         |A       |1     |
|AP-2    |8     |0         |A       |0     |
|AP-2    |9     |1         |A       |0     |
|AP-2    |10    |0         |S       |0     |
+--------+------+----------+--------+------+

我们如何使用PySpark数据访问操作来实现这一点？

Answer 1

您的问题不难解决，为了解释它，我将注释放在代码中：

from pyspark.sql import Row, Window
import pyspark.sql.functions as f


df = spark.createDataFrame([
  Row(app_id='AP-1', order=1, entry_flag=1, operator='S'),
  Row(app_id='AP-1', order=2, entry_flag=0, operator='A'),
  Row(app_id='AP-2', order=3, entry_flag=0, operator='S'),
  Row(app_id='AP-2', order=4, entry_flag=0, operator='A'),
  Row(app_id='AP-2', order=5, entry_flag=1, operator='S'),
  Row(app_id='AP-2', order=6, entry_flag=0, operator='S'),
  Row(app_id='AP-2', order=7, entry_flag=0, operator='A'),
  Row(app_id='AP-2', order=8, entry_flag=0, operator='A'),
  Row(app_id='AP-2', order=9, entry_flag=1, operator='A'),
  Row(app_id='AP-2', order=10, entry_flag=0, operator='S')
])

# Creating a column to group each entry where the value is 1
w_entry = Window.partitionBy('app_id').orderBy('order')
df = df.withColumn('group', f.sum('entry_flag').over(w_entry))

# Applying your boolean rule
df = df.withColumn('match', f.when(f.col('group') > f.lit(0), 
                                   (f.col('entry_flag') == f.lit(0)) & (f.col('operator')== f.lit('A')))
                             .otherwise(f.lit(False)))
# +------+-----+----------+--------+-----+-----+
# |app_id|order|entry_flag|operator|group|match|
# +------+-----+----------+--------+-----+-----+
# |AP-1  |1    |1         |S       |1    |false|
# |AP-1  |2    |0         |A       |1    |true |
# |AP-2  |3    |0         |S       |0    |false|
# |AP-2  |4    |0         |A       |0    |false|
# |AP-2  |5    |1         |S       |1    |false|
# |AP-2  |6    |0         |S       |1    |false|
# |AP-2  |7    |0         |A       |1    |true |
# |AP-2  |8    |0         |A       |1    |true |
# |AP-2  |9    |1         |A       |2    |false|
# |AP-2  |10   |0         |S       |2    |false|
# +------+-----+----------+--------+-----+-----+

# If a group has two or more matches like the example below
# |AP-2  |7    |0         |A       |1    |true |
# |AP-2  |8    |0         |A       |1    |true |
# identify which is the first occurrence and set `flag_x` with 1 to it.

w_flag = Window.partitionBy('app_id', 'group', 'match')
df = df.withColumn('flag_x', (f.col('match') & (f.col('order') == f.min('order').over(w_flag))).cast('int'))

# Drop temporary columns
df = df.drop('group', 'match')

df.show(truncate=False)
# +------+-----+----------+--------+------+
# |app_id|order|entry_flag|operator|flag_x|
# +------+-----+----------+--------+------+
# |AP-1  |1    |1         |S       |0     |
# |AP-1  |2    |0         |A       |1     |
# |AP-2  |3    |0         |S       |0     |
# |AP-2  |4    |0         |A       |0     |
# |AP-2  |5    |1         |S       |0     |
# |AP-2  |6    |0         |S       |0     |
# |AP-2  |7    |0         |A       |1     |
# |AP-2  |8    |0         |A       |0     |
# |AP-2  |9    |1         |A       |0     |
# |AP-2  |10   |0         |S       |0     |
# +------+-----+----------+--------+------+