你现在遇到的是需要把DataFrame的多列和一个查找表做映射的需求，其实完全不需要多次做Join（多次Join不仅代码繁琐，还会导致不必要的笛卡尔积，性能很差），我们可以用映射表广播+内置函数的方式来高效实现这个需求，下面给你两种实用的方案：

方案一：使用Spark内置函数（推荐，性能更优） #

利用Spark SQL的map_from_entries函数把查找表df2转换成一个Key-Value的映射Map，然后通过交叉连接（因为映射表只有一行，不会产生笛卡尔积）关联到df1，最后遍历每一列生成对应的映射列：

from pyspark.sql import functions as F

# 将df2转换为包含全局映射的单行DataFrame
mapping_df = df2.agg(
    F.map_from_entries(F.collect_list(F.struct("key", "val"))).alias("value_map")
)

# 广播映射表（小表广播能大幅提升性能），并和df1做交叉连接
df_with_mapping = df1.crossJoin(F.broadcast(mapping_df))

# 遍历df1的每一列，生成对应的lu列
for col in df1.columns:
    # 提取列名的数字后缀（比如c1→1），生成lu1这样的列名
    lu_col_name = f"lu{col[-1]}"
    df_with_mapping = df_with_mapping.withColumn(lu_col_name, F.col("value_map")[col])

# 筛选出需要的列（原df1列 + 新生成的lu列）
final_df = df_with_mapping.select(df1.columns + [f"lu{col[-1]}" for col in df1.columns])
final_df.show()

方案二：使用广播字典+UDF（灵活度更高） #

如果需要更灵活的逻辑处理，可以把df2转换成Python字典并广播，然后用UDF来实现映射：

from pyspark.sql import functions as F

# 将df2转换为Python字典，并广播（避免每个Executor重复加载）
mapping_dict = dict(df2.collect())
broadcast_mapping = sc.broadcast(mapping_dict)

# 创建自定义UDF，从广播字典中查找对应值
lookup_udf = F.udf(lambda key: broadcast_mapping.value.get(key))

# 逐个处理df1的列，生成映射列
final_df = df1
for col in df1.columns:
    lu_col_name = f"lu{col[-1]}"
    final_df = final_df.withColumn(lu_col_name, lookup_udf(F.col(col)))

final_df.show()

为什么不推荐多次Join？ #

如果对每一列单独做Join，比如先Join c1，再Join c2，会导致DataFrame不断做关联操作，当数据量较大时会产生笛卡尔积，不仅性能极差，还会占用大量资源。而上面两种方案都是基于映射表的一次性加载，效率要高得多。

运行上面的代码后，你就能得到想要的结果：

+---+---+---+---+---+---+---+---+---+---+
| c1| c2| c3| c4| c5|lu1|lu2|lu3|lu4|lu5|
+---+---+---+---+---+---+---+---+---+---+
|  0|  1|  2|  3|  4|  A|  B|  C|  D|  E|
|  5|  6|  7|  8|  9|  F|  G|  H|  I|  J|
+---+---+---+---+---+---+---+---+---+---+

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