Databricks中如何拼接ADF传入的列名列表生成新列

问题根源 #

你从ADF传入的Primary_Key_Columns是字符串格式（比如输入"id,name"），dbutils.widgets.get()返回的是原始字符串，直接传给f.concat()时，Spark会把这个整个字符串当作单个列名去查找，自然找不到对应列，导致报错。

解决方案 #

需要先将字符串格式的列名转换成Spark列对象的列表，再传入拼接函数：

1. 解析列名字符串：将逗号分隔的字符串分割成列名列表，同时清理每个列名的空格（避免输入时带空格导致列名不匹配）
2. 转换为Spark列对象：用f.col()将每个列名转为列对象
3. 执行拼接：如果需要无分隔符拼接用f.concat()，需要分隔符（比如_、|）推荐用f.concat_ws()（更灵活）

完整代码示例 #

import pyspark.sql.functions as f

# 获取ADF传入的主键列名字符串
dbutils.widgets.text('Primary_Key_Columns','')
primary_key_str = dbutils.widgets.get('Primary_Key_Columns')

# 处理空输入的情况
if primary_key_str.strip():
    # 分割列名并清理空格
    pk_columns = [col.strip() for col in primary_key_str.split(',')]
    # 转换为Spark列对象
    pk_col_objects = [f.col(col) for col in pk_columns]
    
    # 方式1：无分隔符拼接
    df = df.withColumn('PK_Col', f.concat(*pk_col_objects))
    
    # 方式2：带分隔符拼接（推荐，可读性更强，比如用下划线分隔）
    # df = df.withColumn('PK_Col', f.concat_ws('_', *pk_col_objects))
else:
    # 处理未传入主键列的情况，比如赋值默认值或抛出提示
    df = df.withColumn('PK_Col', f.lit(None))

关键说明 #

• 使用*解包列对象列表：f.concat(*pk_col_objects)是把列表里的每个列对象作为单独参数传给concat，而不是传整个列表
• concat_ws()更适合多列拼接：第一个参数是分隔符，后面跟列对象，避免拼接后内容连在一起难以区分
• 一定要清理列名的空格：如果ADF传入的是"id , name"，分割后不清理空格会变成['id ', ' name']，同样会找不到列

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