Spark Dataset执行多次join操作后列解析失败问题咨询

你遇到的列解析错误和是否使用强类型Dataset无关，核心问题是列引用的绑定逻辑：

• 当你调用left("office_id")时，得到的Column对象和你最开始定义的left: Dataset[Employee]的元数据绑定，对应了唯一的列id（报错信息里的office_id#42L）
• 第一次joinWith+map操作会生成全新的Dataset，这个新Dataset的所有列都会被分配新的唯一id（报错里的新office_id是office_id#116L）
• 第二次join你依然用原始left变量引用office_id，Spark按绑定的旧idoffice_id#42L去找，自然找不到对应列，就会抛出解析异常。

强类型Dataset的作用仅保障运行时对象序列化、类型转换符合case class定义，不会改变Column的绑定逻辑。

下面几个方案都可以解决该问题： #

方案1：用中间转换后的Dataset变量引用列 #

把第一次转换后的结果赋值给新变量，第二次join用新变量引用列，保证列id匹配：

val right: Dataset[Renumeration]  = ???
val left: Dataset[Employee] = ???

// 第一次转换后的结果赋值给新变量
val afterProjUpdate: Dataset[Employee] = left.joinWith(broadcast(right), left("proj_id") === right("id"), "leftouter")
  .map { case(emp, ren) => updateProj(emp, ren) }

// 第二次join用新变量afterProjUpdate引用列
val finalDs = afterProjUpdate.joinWith(broadcast(right), afterProjUpdate("office_id") === right("id"), "leftouter")
  .map { case(emp, ren) => updateOffice(emp, ren) }

def updateProj(emp: Employee, ren: Renumeration): Employee = ???
def updateOffice(emp: Employee, ren: Renumeration): Employee = ???

方案2：使用动态列名引用 #

第二次join时不用绑定了旧id的原始left变量引用列，改用按列名匹配的通用引用：

import org.apache.spark.sql.functions.col

left.joinWith(broadcast(right), left("proj_id") === right("id"), "leftouter")
  .map { case(emp, ren) => updateProj(emp, ren) }
  // 直接用col("office_id")按列名匹配当前Dataset的列，不会绑定旧id
  .joinWith(broadcast(right), col("office_id") === right("id"), "leftouter")
  .map { case(emp, ren) => updateOffice(emp, ren) }

方案3：先完成两次join再做更新 #

避免多次转换带来的列绑定问题，先把两次关联都做完再统一执行业务更新逻辑，性能也更好（减少一次shuffle和序列化/反序列化开销）：

import org.apache.spark.sql.functions.col

left.joinWith(broadcast(right.as("proj_ren")), col("proj_id") === col("proj_ren.id"), "leftouter")
  .joinWith(broadcast(right.as("office_ren")), col("_1.office_id") === col("office_ren.id"), "leftouter")
  .map { case ((emp, projRen), officeRen) =>
    // 先更新项目薪资再更新办公地薪资
    updateOffice(updateProj(emp, projRen), officeRen)
  }

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