INOLTRO_PSAP2为'S'时DataFrame匹配静态表替换值的高效方案问询

你现有写法存在两个核心问题：

1. 逻辑错误：join后取head()[0]只会返回第一条匹配的描述值，所有INOLTRO_PSAP2为'S'的行都会被替换成同一个值，无法对应每行的id匹配结果
2. 性能极差：每行触发一次join操作，数据量稍大就会出现严重的性能问题甚至任务失败

方案1：广播小表左连（推荐，通用性强） #

如果df_ente是小表，直接用广播join一次完成匹配，全程无shuffle，性能远高于原写法：

from pyspark.sql import functions as F
from pyspark.sql.functions import broadcast

# 1. 广播小表左连，一次完成所有id和描述的匹配
df_joined = df.join(broadcast(df_ente), on="id_ente_competenza_tipo", how="left")

# 2. 按条件替换字段值
df_result = df_joined.withColumn(
    "id_ente_competenza_tipo",
    F.when(
        F.col("INOLTRO_PSAP2") == "S",
        F.col("des_ente_competenza_tipo")
    ).otherwise(F.col("id_ente_competenza_tipo").cast("string")) # 注意类型统一：数字id转字符串和描述字段类型对齐
).drop("des_ente_competenza_tipo") # 删掉多余的辅助字段

方案2：字典映射（适合映射表记录极少的场景） #

如果df_ente的记录数非常少（比如仅几十条），可以直接转成字典存到Spark字面量中，连join都不需要，性能最高：

from pyspark.sql import functions as F

# 先把映射表转成Python字典
id_to_des = {row["id_ente_competenza_tipo"]: row["des_ente_competenza_tipo"] for row in df_ente.collect()}

# 构造map映射字段直接查询
map_expr = F.create_map([F.lit(x) for x in sum(id_to_des.items(), ())])

df_result = df.withColumn(
    "id_ente_competenza_tipo",
    F.when(
        F.col("INOLTRO_PSAP2") == "S",
        map_expr[F.col("id_ente_competenza_tipo")]
    ).otherwise(F.col("id_ente_competenza_tipo").cast("string"))
)

两种方案输出都和预期结果完全一致，性能是原写法的几十到上千倍不等，数据量越大优势越明显。
注意代码里的字段大小写要和实际DataFrame的字段名保持一致，避免匹配不到字段的报错。

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