PySpark - 如何根据一列中两个值过滤行的连续块了数据帧的

Question

我有一个数据帧，我想创建基于使用pyspark一些列值的另一数据帧。例如：下面是我的主要数据框 -

Part1   Part2   Part3   Part4
aaa      up      24     k-123
bbb     down     45     i-98
ccc     down     54     k-89
fff     int      23     l-34
xyz      up      22     o-89
www      up      89     u-56

现在，我想创建另一个数据帧，将搜索“下”第一次发生，并会去钱柜的“向上”第一次发生。因此，预期的数据帧将是：

   Part1    Part2   Part3   Part4
    bbb     down     45     i-98
    ccc     down     54     k-89
    fff     int      23     l-34
    xyz      up      22     o-89

Answer 1

第1步：创建DataFrame。

from pyspark.sql.functions import when, col, lit
df = spark.createDataFrame(
    [('aaa','up',24,'k-123'),('bbb','down',45,'i-98'),('ccc','down',54,'k-89'),
     ('fff','int', 23,'l-34'),('xyz','up',22,'o-89'),('www','up',89,'u-56')], 
    schema = ['Part1','Part2','Part3','Part4']
)
df.show()
+-----+-----+-----+-----+
|Part1|Part2|Part3|Part4|
+-----+-----+-----+-----+
|  aaa|   up|   24|k-123|
|  bbb| down|   45| i-98|
|  ccc| down|   54| k-89|
|  fff|  int|   23| l-34|
|  xyz|   up|   22| o-89|
|  www|   up|   89| u-56|
+-----+-----+-----+-----+

第2步：首先，我们需要找到down的第一次出现，并删除它上面所有行。要做到这一点，我们创建一个列cumulative，具有值为1，如果Part2 == down，否则为0，并最终采取这一列的累计总和。

df = df.withColumn('Dummy',lit('dummy'))
df = df.withColumn('cumulative',when(col('Part2')=='down',1).otherwise(0))
df = df.selectExpr(
    'Part1','Part2','Part3','Part4','Dummy',
    'sum(cumulative) over (order by row_number() over (order by Dummy)) as cumulative'
 )
df.show()
+-----+-----+-----+-----+-----+----------+
|Part1|Part2|Part3|Part4|Dummy|cumulative|
+-----+-----+-----+-----+-----+----------+
|  aaa|   up|   24|k-123|dummy|         0|
|  bbb| down|   45| i-98|dummy|         1|
|  ccc| down|   54| k-89|dummy|         2|
|  fff|  int|   23| l-34|dummy|         2|
|  xyz|   up|   22| o-89|dummy|         2|
|  www|   up|   89| u-56|dummy|         2|
+-----+-----+-----+-----+-----+----------+

现在，删除所有行累计总和为0。这将删除所有行直到down出现的第一次。

df = df.where(col('cumulative')>=1)

第3步：执行相同按上述步骤2中，不同之处在于用于up做到这一点，并删除所有行，其中在列cumulative值小于或等于1。通过这种方式，我们将除去up的第一次出现下面的所有行。

df = df.withColumn('cumulative',when(col('Part2')=='up',1).otherwise(0))
df = df.selectExpr(
    'Part1','Part2','Part3','Part4','Dummy',
    'sum(cumulative) over (order by row_number() over (order by Dummy)) as cumulative'
 )
df = df.where(col('cumulative')<=1).drop('Dummy','cumulative')
df.show()
+-----+-----+-----+-----+
|Part1|Part2|Part3|Part4|
+-----+-----+-----+-----+
|  bbb| down|   45| i-98|
|  ccc| down|   54| k-89|
|  fff|  int|   23| l-34|
|  xyz|   up|   22| o-89|
+-----+-----+-----+-----+