作为替代，您可以尝试使用别名，如下所示：

val aliases = dataDF.columns.map(columnName => $"${columnName}".as(map_of_codename.getOrElse(columnName, "None")))
dataDF.select(aliases: _*).show()

dataDF.select(aliases: _*).explain(true)

执行计划将由单个投影节点组成，例如，它可能有助于减少优化阶段：

== Analyzed Logical Plan ==
code_1: int, None: int, code_3: int, code_4: int, code_5: int
Project [col_1#16 AS code_1#77, col_2#17 AS None#78, col_3#18 AS code_3#79, col_4#19 AS code_4#80, col_5#20 AS code_5#81]
+- Project [_1#5 AS col_1#16, _2#6 AS col_2#17, _3#7 AS col_3#18, _4#8 AS col_4#19, _5#9 AS col_5#20]
   +- LocalRelation [_1#5, _2#6, _3#7, _4#8, _5#9]

也就是说，我不确定它是否能解决性能问题，因为在这两种情况下，您的 foreach 通过上述方案，可以将物理方案优化到单个节点 CollapseProject 规则。
仅供参考， withColumnRenamed 在引擎盖下使用类似的方法，不同的是它对每个列分别执行：

def withColumnRenamed(existingName: String, newName: String): DataFrame = {
    val resolver = sparkSession.sessionState.analyzer.resolver
    val output = queryExecution.analyzed.output
    val shouldRename = output.exists(f => resolver(f.name, existingName))
    if (shouldRename) {
      val columns = output.map { col =>
        if (resolver(col.name, existingName)) {
          Column(col).as(newName)
        } else {
          Column(col)
        }
      }
      select(columns : _*)
    } else {
      toDF()
    }
  }

你对观察到的性能问题有什么意见吗？有哪些措施可以帮助确定操作需要时间？也许这不一定与列重命名有关？你以后对这些重命名的列做什么？

一种方法是先获得不带spark的完整Map列列表，然后到for循环以重命名所有列，而不是call columns.foreach
下面是我的解决方案的一个例子（对不起，我不是scala方面的Maven，有些数据解析可能很难看）

var dataDF = Seq((10, 20, 30, 40, 50),(100, 200, 300, 400, 500),(10, 222, 333, 444, 555),(1123, 2123, 3123, 4123, 5123),(1321, 2321, 3321, 4321, 5321))
  .toDF("col_1", "col_2", "col_3", "col_4", "col_5")
dataDF.show(false)

val mapDF = Seq(("col_1", "code_1", "true"),("col_3", "code_3", "true"),("col_4", "code_4", "true"),("col_5", "code_5", "true"))
  .toDF("original_name", "code_name", "important")

val schema_mapping = mapDF.select("original_name", "code_name").collect()
//For mapping of None column (col 2)
val none_mapping = old_schema.map(x => if (!schema_mapping.map(x => x(0)).contains(x)) Array[String](x, "None")).filter(_ != ())

for(i <- 0 until schema_mapping.length){
    try {
        dataDF = dataDF.withColumnRenamed(schema_mapping(i)(0).toString, schema_mapping(i)(1).toString)
    }
    catch{
        case e : Throwable => println("cannot rename" +  schema_mapping(i)(0).toString + " to " + schema_mapping(i)(1).toString)
    }
}

for(i <- 0 until none_mapping.length){
    try {
        dataDF = dataDF.withColumnRenamed(none_mapping(i).asInstanceOf[Array[String]](0), none_mapping(i).asInstanceOf[Array[String]](1))
    }
    catch{
        case e : Throwable => println("cannot rename")
    }
}

dataDF.show(false)

在spark ui中，为每一列重命名它，使其成为一个stage，但是当我们看到dag可视化时，这些stage应该并行执行。