如何基于映射表存储的列名实现Spark两表关联？

我尝试通过一张映射表生成的列名列表关联两张Spark表，但没找到可行方法。基础数据定义如下：

data = [('B_YEAR, B_ORG', 'YEAR, ORG', 1), ('B_YEAR', 'YEAR', 2) ]
test = spark.createDataFrame(data, ['key_orig', 'key_map', 'Fil_Number']) 
fc_year = 2024

我尝试的关联代码存在问题，无法实现预期关联：

for i in range (1,x):
test_1= test.filter(col("Fil_Number")==1)
list_key_orig = test_1.select("key_orig").collect()
list_key_map = test_1.select("key_map").collect()

df_calc_values =spark.read.table("hive_metastore.reporting_datalake.df_calc_fc_values")

df_calc_values = df_calc_values.filter((col("GJ")==fc_year))
display(df_calc_values)

df_new= df_orig.join(df_calc_values, on = (key_orig1.key_orig == key_map1.key_map) ,how = "left")

核心问题分析 #

• 原代码中collect()返回的是Row对象列表，不是纯列名字符串，无法直接用于关联条件
• 未正确解析映射表中逗号分隔的多列映射关系
• 关联条件的写法错误，没有动态生成多列匹配的逻辑

实现步骤 #

1. 预处理映射表：将逗号分隔的列名字符串拆分为数组，方便后续遍历
2. 获取目标映射关系：根据Fil_Number提取对应的原表列和目标表列映射
3. 动态生成关联条件：遍历列对生成匹配条件，再组合为最终关联逻辑
4. 执行表关联：使用动态生成的条件完成left join

完整代码示例 #

from pyspark.sql.functions import split, col
import functools

# 1. 预处理映射表，拆分列名字符串为数组
test = test.withColumn("key_orig_arr", split(col("key_orig"), ", ")) \
           .withColumn("key_map_arr", split(col("key_map"), ", "))

# 2. 获取指定Fil_Number的映射关系（这里以Fil_Number=1为例）
mapping_row = test.filter(col("Fil_Number") == 1).select("key_orig_arr", "key_map_arr").first()
orig_col_list = mapping_row.key_orig_arr
map_col_list = mapping_row.key_map_arr

# 3. 动态生成关联条件
join_conditions = []
for orig_col, map_col in zip(orig_col_list, map_col_list):
    # 生成每一组列的匹配条件
    join_conditions.append(col(orig_col) == col(map_col))
# 组合所有条件（多列关联时需同时满足）
final_join_condition = functools.reduce(lambda cond1, cond2: cond1 & cond2, join_conditions)

# 4. 读取并过滤目标表
df_calc_values = spark.read.table("hive_metastore.reporting_datalake.df_calc_fc_values") \
                       .filter(col("GJ") == fc_year)

# 5. 执行关联
df_new = df_orig.join(df_calc_values, on=final_join_condition, how="left")
display(df_new)

扩展说明 #

• 如果需要循环处理不同Fil_Number的映射，可遍历test中唯一的Fil_Number值，重复上述逻辑
• 若两张表存在列名冲突，可给表添加别名，比如df_orig.alias("orig")，然后用orig.B_YEAR == calc.YEAR的形式写条件

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