python - 如何在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')