如何在PySpark中不使用UDF实现广播变量生成新列？

我正在探索PySpark中Broadcast变量的实现，样本数据集包含State_Code列，希望通过Broadcast变量实现类似{'CA':'California', 'NJ':'New Jersey'}的映射。我已通过以下代码实现新列生成：

val = {"CA": "California", "NY": "New York", "NJ": "New Jersey"}

broad = sc.broadcast(val)

def broad_function(a):
    return broad.value[a]

broad_udf = udf(broad_function)

df.withColumn('State_Name',broad_udf('State_code')).show()

该代码可生成州名称新列，但UDF无法利用Spark优化。使用Broadcast变量的核心目的是优化性能，如何在不使用UDF且不转换为RDD的前提下，利用Broadcast变量在DataFrame中生成新列？我尝试过when、col方法，但无法利用Broadcast变量，期望得到可行方案。

可以通过PySpark内置的MapType相关函数结合Broadcast变量实现需求，完全无需UDF，同时能让Spark执行引擎进行全量优化。

方法1：使用create_map生成映射列 #

将Broadcast变量的键值对转换为create_map所需的参数列表，再用State_Code列作为键匹配映射值：

from pyspark.sql.functions import create_map, lit, col

val = {"CA": "California", "NY": "New York", "NJ": "New Jersey"}
broad = sc.broadcast(val)

# 把广播变量的键值对拆解为create_map的参数：[lit(key), lit(value), ...]
map_expr = create_map(*[lit(item) for sublist in broad.value.items() for item in sublist])

# 生成新列
df.withColumn("State_Name", map_expr[col("State_Code")]).show()

方法2：使用map_from_entries（Spark 2.3+） #

这种方式更简洁，直接将广播变量的键值对转换为struct数组，再生成MapType列：

from pyspark.sql.functions import map_from_entries, array, struct, col

val = {"CA": "California", "NY": "New York", "NJ": "New Jersey"}
broad = sc.broadcast(val)

# 将键值对转为struct结构，再组合成数组生成map
map_expr = map_from_entries(array(*[struct(lit(k), lit(v)) for k, v in broad.value.items()]))

# 生成新列
df.withColumn("State_Name", map_expr[col("State_Code")]).show()

性能优化说明 #

• 以上方案均使用PySpark内置函数，而非自定义UDF。Spark对内置函数支持完整的逻辑优化（如谓词下推）和执行计划优化（如代码生成），能规避UDF带来的性能损耗。
• 方案依然保留了Broadcast变量的核心优势：映射字典会被一次性分发到各个Executor节点的内存中，避免任务重复传输小数据集，最大化利用节点本地缓存。

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