Spark Scala中如何将sig列内嵌列名替换为对应列的实际值

AI 小北

2026-9-25

实现方案

这里提供PySpark和Scala两种版本的可运行实现，支持任意数量的非`sig`列，无需提前硬编码列名：

PySpark版本

1. 提取待替换列名

提取DataFrame中除sig外的所有列作为待匹配替换的列集合：

from pyspark.sql import functions as F
from pyspark.sql.types import StringType
# 替换为你自己的源DataFrame变量名
df = 你的源DataFrame
replace_cols = [col for col in df.columns if col != "sig"]

2. 注册替换逻辑UDF

def replace_sig(sig_value, *col_values):
    # 构建列名和当前行对应值的映射关系
    col_val_map = dict(zip(replace_cols, col_values))
    result = sig_value
    for col_name, col_val in col_val_map.items():
        # 匹配规则：列名包裹在`-`和`:`之间，避免部分匹配错误
        result = result.replace(f"-{col_name}:", f"-{str(col_val)}:")
    return result
replace_sig_udf = F.udf(replace_sig, StringType())

3. 调用UDF更新sig列

result_df = df.withColumn(
    "sig",
    replace_sig_udf(
        F.col("sig"),
        *[F.col(col) for col in replace_cols]
    )
)
# 输出验证结果
result_df.show(truncate=False)

Scala版本参考

import org.apache.spark.sql.functions._
import org.apache.spark.sql.Row

val df = 你的源DataFrame
val replaceCols = df.columns.filter(_ != "sig")
val replaceSigUdf = udf((sig: String, row: Row) => {
  replaceCols.foldLeft(sig){ (acc, colName) =>
    acc.replace(s"-$colName:", s"-${row.getAs[Any](colName).toString}:")
  }
})
val resultDf = df.withColumn("sig", replaceSigUdf(col("sig"), struct(replaceCols.map(col): _*)))
resultDf.show(false)