如何在pyspark中合并连续的重复行

Question

我有一个以下格式的数据框

Col-1Col-2
a   d1
a   d2
x   d3
a   d4
f   d5
a   d6
a   d7

我想通过查看 col1 中的连续重复项来合并 col-2 中的值。我们可以看到 a 出现了两次连续重复项。它应该分别合并 d1+d2 和 d6+d7。这些列的数据类型是字符串，d1+d2 表示将字符串 d1 与 d2 连接

最终的输出应该如下图所示

Col-1Col-2
a   d1+d2
x   d3
a   d4
f   d5
a   d6+d7

Answer 1

您将需要一个定义 DataFrame 顺序的列。如果一个尚不存在，您可以使用pyspark.sql.functions.monotonically_increasing_id 创建一个。

import pyspark.sql.functions as f
df = df.withColumn("id", f.monotonically_increasing_id())

接下来，您可以使用this post 中描述的技术为每组连续重复创建段：

import sys
import pyspark.sql.Window

globalWindow = Window.orderBy("id")
upToThisRowWindow = globalWindow.rowsBetween(-sys.maxsize-1, 0)

df = df.withColumn(
    "segment",
    f.sum(
        f.when(
            f.lag("Col-2", 1).over(globalWindow) != f.col("Col-2"),
            1
        ).otherwise(0)
    ).over(upToThisRowWindow)+1
)

现在您可以按段分组并使用pyspark.sql.functions.collect_list 将值收集到一个列表中并使用pyspark.sql.functions.concat() 来连接字符串：

df = df.groupBy('segment').agg(f.concat(f.collect_list('Col-2'))).drop('segment')