Spark DataFrame连接后重复列取值问题：别名失效如何解决？

当执行Full Join后，DataFrame不会保留表别名的前缀，所以直接用s.Pk1这种方式访问会报错。除了给列加别名，还有以下几种可行方法：

1. 通过列的索引位置访问 #

重复列会按照Join的顺序排列（先source表的列，再target表的列），可以直接通过列的索引位置来获取对应的值。比如第一个Pk1来自source，第二个来自target：

// 获取source侧的Pk1（假设是第1列，索引从0开始）
val sourcePk1 = join.col(0)
// 获取target侧的Pk1（假设是第5列，根据实际列顺序调整）
val targetPk1 = join.col(4)

⚠️ 注意：这种方式依赖列的固定顺序，如果后续表结构发生变化（比如新增/删除列），索引会失效，需要谨慎使用。

2. 结合列名数组动态定位 #

可以通过columns数组获取所有列名，再过滤出重复列的位置，动态获取目标列：

// 找出所有名为Pk1的列的索引
val pk1Indexes = join.columns.zipWithIndex.filter(_._1 == "Pk1").map(_._2)
// 取第一个索引对应的source侧Pk1
val sourcePk1 = join.col(pk1Indexes(0))
// 取第二个索引对应的target侧Pk1
val targetPk1 = join.col(pk1Indexes(1))

也可以在select时直接用索引重命名，避免后续混淆：

val cleanedDF = join.select(
  join.columns.zipWithIndex.map { case (colName, idx) =>
    colName match {
      case "Pk1" if idx == pk1Indexes(0) => col(s"_$idx").alias("s_Pk1")
      case "Pk1" if idx == pk1Indexes(1) => col(s"_$idx").alias("t_Pk1")
      case "Col3" => 
        // 同理处理Col3的重复列
        val col3Indexes = join.columns.zipWithIndex.filter(_._1 == "Col3").map(_._2)
        if (idx == col3Indexes(0)) col(s"_$idx").alias("s_Col3") else col(s"_$idx").alias("t_Col3")
      case _ => col(colName)
    }
  }: _*
)

3. Join前预先重命名重复列 #

在执行Join操作之前，先把target表中的重复列重命名，从根源上避免列名冲突：

// 重命名target表的重复列
val renamedTarget = target
  .withColumnRenamed("Pk1", "t_Pk1")
  .withColumnRenamed("Col3", "t_Col3")

// 执行Full Join
val join = source.alias("s").join(
  renamedTarget.alias("t"),
  source("Pk1") === renamedTarget("t_Pk1") && source("Pk2") === renamedTarget("Pk3"),
  "full"
)

// 直接通过新列名访问
val targetPk1 = join("t_Pk1")
val targetCol3 = join("t_Col3")

4. 将表列打包为Struct #

把source和target的所有列分别打包成Struct类型，通过Struct的别名来区分不同表的列：

// 将source表的所有列打包成名为source的Struct
val sourceStructDF = source.select(struct(source.columns.map(col): _*).alias("source"))
// 将target表的所有列打包成名为target的Struct
val targetStructDF = target.select(struct(target.columns.map(col): _*).alias("target"))

// 基于Struct内的列执行Join
val join = sourceStructDF.join(
  targetStructDF,
  sourceStructDF("source.Pk1") === targetStructDF("target.Pk1") &&
  sourceStructDF("source.Pk2") === targetStructDF("target.Pk3"),
  "full"
)

// 通过Struct别名访问对应表的列
val sourcePk1 = join("source.Pk1")
val targetCol3 = join("target.Col3")

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