如何基于Schema为DataFrame选择不同列名并合并数据？

你的问题核心是两个数据源列名采用了不同的命名规范（下划线/驼峰），导致select时找不到对应列，且union要求Schema完全一致。可以通过统一列名映射+列对齐+使用unionByName来实现Schema感知的合并，具体步骤如下：

1. 定义列名转换规则 #

先实现一个函数，将不同风格的列名统一为目标格式（比如把驼峰式转成下划线式，或者反过来），处理列名的细微差异：

// 驼峰转下划线（比如customerId → customer_id）
def camelToUnderscore(name: String): String = {
  name.replaceAll("([a-z])([A-Z])", "$1_$2").toLowerCase()
}

// 可选：下划线转驼峰（比如customer_id → customerId）
def underscoreToCamel(name: String): String = {
  name.split("_").map(word => word.head.toUpper + word.tail).mkString.replaceFirst("^.", _.toLowerCase())
}

2. 实现列对齐函数 #

针对每个输入DataFrame，将其列对齐到目标Schema：存在的列重命名为目标列名，不存在的列填充null（需匹配目标列的数据类型）：

import org.apache.spark.sql.{DataFrame, Column}
import org.apache.spark.sql.functions.{col, lit}
import org.apache.spark.sql.types.StringType

def alignToTargetSchema(df: DataFrame, targetColumns: Seq[String]): DataFrame = {
  // 建立「转换后的列名 → 原列名」的映射
  val columnMapping = df.columns.map(colName => camelToUnderscore(colName) -> colName).toMap
  
  // 遍历目标列，逐个处理
  val alignedDF = targetColumns.foldLeft(df) { (acc, targetCol) =>
    columnMapping.get(targetCol) match {
      // 找到匹配列：保留原列值，重命名为目标列名
      case Some(sourceCol) => acc.withColumn(targetCol, col(sourceCol))
      // 未找到匹配列：填充null，类型与目标列一致（这里假设是StringType，可根据实际调整）
      case None => acc.withColumn(targetCol, lit(null).cast(StringType))
    }
  }
  
  // 按目标列顺序选择，确保Schema完全一致
  alignedDF.select(targetColumns.map(col): _*)
}

3. 修改合并逻辑 #

使用unionByName代替union（unionByName会按列名匹配，忽略列的位置），先对每个DataFrame做列对齐再合并：

val targetColumns = Seq("id", "customer_id")
val dataframe = input.map(df => alignToTargetSchema(df, targetColumns)).reduce(_.unionByName(_))

关键说明 #

• 列名转换函数可以根据你的实际命名差异调整，比如还可以处理大小写差异（比如ID和id），只需在转换时统一为小写/大写即可。
• 填充null时要注意数据类型匹配，如果目标列是数值型，就把StringType改成对应类型（比如IntegerType）。
• unionByName在Spark 2.3+版本可用，相比union更适合Schema列顺序不一致的场景。

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