PySpark按列值变化模式分组：合并通话中发言人的连续发言单词

核心思路 #

• 这是典型的「间隙与孤岛」（Gap and Island）场景，核心是为每个连续发言段生成唯一分组标识
• 仅需4步即可完成：定义排序窗口 -> 标记发言切换点 -> 生成连续分组ID -> 分组聚合单词

完整实现代码 #

1. 依赖导入与测试数据构造 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, lag, when, sum, collect_list
from pyspark.sql.window import Window

# 初始化SparkSession
spark = SparkSession.builder.appName("meeting_word_merge").getOrCreate()
# 构造测试输入数据
data = [
    (1, "Speaker_1", 1, "Hi"),
    (1, "Speaker_1", 2, "I"),
    (1, "Speaker_1", 3, "am"),
    (1, "Speaker_1", 4, "Pete"),
    (1, "Speaker_2", 5, "Hello"),
    (1, "Speaker_1", 6, "Sorry"),
    (1, "Speaker_1", 7, "Gotta"),
    (1, "Speaker_1", 8, "Leave"),
    (2, "Speaker_2", 1, "Hello"),
    (2, "Speaker_2", 2, "Luis"),
    (2, "Speaker_1", 3, "Hey")
]
df = spark.createDataFrame(data, schema=["Call_id", "Speaker", "WordNum", "Word"])

2. 核心逻辑实现 #

# 步骤1：定义基础窗口，按Call_id分区，按WordNum排序保证单词顺序正确
base_window = Window.partitionBy("Call_id").orderBy("WordNum")

# 步骤2：标记发言人切换点：和上一行发言人不同则标记为1，代表新发言段开始
df = df.withColumn("prev_speaker", lag("Speaker").over(base_window))
df = df.withColumn("change_flag", 
                   when(col("prev_speaker").isNull() | (col("prev_speaker") != col("Speaker")), 1)
                   .otherwise(0))

# 步骤3：生成分组ID：对切换标记做累加求和，相同连续段的累加值一致
group_window = base_window.rowsBetween(Window.unboundedPreceding, 0)
df = df.withColumn("group_id", sum("change_flag").over(group_window))

# 步骤4：分组聚合得到最终结果
result_df = df.groupBy("Call_id", "Speaker", "group_id")\
              .agg(collect_list("Word").alias("Sentence"))\
              .orderBy("Call_id", "group_id")\
              .select("Call_id", "Speaker", "Sentence")

3. 结果输出 #

调用result_df.show(truncate=False)即可得到预期输出：

注意事项 #

• 必须保证WordNum是连续且按发言顺序递增的，否则排序后会出现单词顺序错乱
• 该方案同时兼容2位及以上发言人的场景，不需要做特殊适配

内容的提问来源于stack exchange，提问作者Mow.Massimo