PySpark能否广播字典用作查找？Azure Databricks报错求助

原代码中匿名lambda UDF直接引用广播变量时，Spark在将UDF序列化到Executor节点的过程中，无法正确解析广播变量的引用，导致出现“广播变量未加载”错误。匿名lambda的闭包处理逻辑在分布式环境下存在序列化兼容性问题。

方案一：使用命名UDF捕获广播变量 #

通过定义命名函数作为UDF，确保广播变量能被正确捕获并在Executor端访问：

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

# 简化字典创建（用推导式替代循环）
erp_bu_dic = {row['erp_code']: row['bu'] for row in erp_bu if row['erp_code']}
broadcast_erp_bu_dic = sc.broadcast(erp_bu_dic)

# 定义命名UDF，使用get方法避免KeyError
def get_business_unit(erp_code):
    return broadcast_erp_bu_dic.value.get(erp_code, None)

# 注册UDF并应用
bu_udf = udf(get_business_unit, StringType())
j06 = j06.withColumn('business_unit', bu_udf('entity_source_id'))
display(j06)

方案二：改用Spark内置广播关联（推荐） #

放弃UDF，采用Spark原生的广播表关联方式，这是更符合分布式计算模型的做法，性能更优且无序列化问题：

from pyspark.sql.functions import broadcast

# 将erp_bu列表转为Spark DataFrame并过滤空值
erp_bu_df = spark.createDataFrame(erp_bu).filter("erp_code is not null").select("erp_code", "bu")

# 广播小表后执行左关联，避免笛卡尔积，提升性能
j06 = j06.join(
    broadcast(erp_bu_df),
    j06.entity_source_id == erp_bu_df.erp_code,
    "left"
).drop("erp_code").withColumnRenamed("bu", "business_unit")

display(j06)

• 方案二是优先推荐的方式：Spark会自动优化广播关联的执行逻辑，比UDF性能更高，且无需手动处理广播变量的生命周期。
• 若必须使用UDF，方案一的命名UDF能解决匿名lambda的序列化问题，同时get方法让查找逻辑更健壮，避免因未找到匹配值抛出异常。

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