如何在PySpark Delta表中按车辆号生成去重拼接字段并更新全行

需求说明 #

你的Delta表包含VehNum、Control_circuit、similar_partnumbers等字段，每个VehNum对应唯一的Control_circuit，且同一VehNum下有多行数据，similar_partnumbers存在重复值。需要实现：

• 按VehNum分组，提取所有similar_partnumbers并去重
• 将去重后的结果拼接为字符串
• 把该字符串更新到对应VehNum下所有行的similar_partnumbers字段

实现步骤与代码 #

1. 初始化SparkSession并读取Delta表

from pyspark.sql import SparkSession
from pyspark.sql.functions import collect_set, concat_ws, col
from delta.tables import DeltaTable

# 初始化SparkSession（若未初始化）
spark = SparkSession.builder.appName("DeltaUpdateTask").getOrCreate()

# 读取目标Delta表，替换为你的表路径
delta_df = spark.read.format("delta").load("/path/to/your/delta_table")

2. 分组聚合生成去重拼接后的字符串

利用collect_set实现去重收集，再用concat_ws拼接成指定格式的字符串：

# 按VehNum和Control_circuit分组（确保每个VehNum对应唯一Control_circuit）
aggregated_df = delta_df.groupBy("VehNum", "Control_circuit") \
    .agg(collect_set("similar_partnumbers").alias("unique_part_list")) \
    .withColumn("updated_similar_partnumbers", concat_ws(", ", col("unique_part_list"))) \
    .drop("unique_part_list")

3. 关联原表与聚合结果，准备更新数据集

将原表和聚合后的结果通过VehNum和Control_circuit关联，获取需要更新的完整行数据：

# 关联得到待更新的数据集
update_source_df = delta_df.join(
    aggregated_df,
    on=["VehNum", "Control_circuit"],
    how="inner"
).select(
    delta_df["*"],
    aggregated_df["updated_similar_partnumbers"]
)

4. 使用Delta Merge操作完成更新

通过Delta Lake的Merge API高效更新匹配的行，需确保匹配条件能唯一定位每行（示例中假设存在唯一标识字段id，若没有可替换为其他唯一键组合）：

# 获取Delta表对象
delta_table = DeltaTable.forPath(spark, "/path/to/your/delta_table")

# 执行Merge更新
delta_table.alias("target") \
    .merge(
        update_source_df.alias("source"),
        # 匹配条件：唯一定位每行，根据你的表结构调整
        "target.VehNum = source.VehNum AND target.Control_circuit = source.Control_circuit AND target.id = source.id"
    ) \
    .whenMatchedUpdate(set={
        "similar_partnumbers": col("source.updated_similar_partnumbers")
    }) \
    .execute()

5. 验证更新结果

重新读取Delta表，确认更新效果：

updated_df = spark.read.format("delta").load("/path/to/your/delta_table")
updated_df.show(truncate=False)

内容的提问来源于stack exchange，提问作者karthik