如何使用映射批量转换Spark DataFrame多列中的指定ID值？

针对你遇到的多列ID翻译需求，不用逐个列做关联，推荐两种高效批量处理方案，基于Spark 3.3 + Scala实现：

方案1：广播映射 + 自定义UDF（灵活适配复杂ID格式） #

利用广播变量将小映射分发到集群节点，避免重复传输；通过UDF统一处理所有列的ID判断与翻译逻辑：

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

// 初始化SparkSession
val spark = SparkSession.builder().appName("IdTranslation").getOrCreate()
import spark.implicits._

// 示例DataFrame
val df = Seq(
  ("ValueX", "@id/bar@", "ABC"),
  ("@id/foo@", "ValueY", "DEF")
).toDF("Col1", "Col2", "Col3")

// ID-翻译映射，转为广播变量
val idMapping = Map("@id/foo@" -> "Foo", "@id/bar@" -> "Bar")
val broadcastMapping = spark.sparkContext.broadcast(idMapping)

// 定义UDF：判断是否为ID格式，是则翻译，否则返回原值
val translateIdUdf = udf((value: String) => {
  if (value != null && value.startsWith("@id/") && value.endsWith("@")) {
    broadcastMapping.value.getOrElse(value, value) // 找不到映射则保留原ID
  } else {
    value
  }
})

// 批量处理所有列：对每一列应用UDF，生成新列（或覆盖原列）
val translatedDf = df.select(df.columns.map(colName => translateIdUdf(col(colName)).alias(colName)): _*)

translatedDf.show()

方案2：内置函数组合（无需UDF，性能更优） #

用Spark内置的when+like+map函数实现，避免自定义UDF的序列化开销：

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

// 将映射转为Spark的Map类型
val idMap = create_map(
  lit("@id/foo@"), lit("Foo"),
  lit("@id/bar@"), lit("Bar")
)

// 批量处理所有列：匹配ID格式时从map取翻译值，否则保留原值
val translatedDf = df.select(df.columns.map(colName =>
  when(col(colName).like("@id/%@"), idMap.getItem(col(colName))).otherwise(col(colName)).alias(colName)
): _*)

translatedDf.show()

为什么比关联更高效？ #

• 关联操作（join）需要对每一列单独做表关联，多次Shuffle操作会大幅增加IO开销；
• 上述两种方案仅需遍历DataFrame一次，广播变量/内置map的查找都是O(1)操作，性能远优于多列join；
• 批量处理逻辑无需逐个指定列，适配任意数量的列，扩展性强。

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