如何使用源的pk1的计划输出：

val source = sourceSeq.toDS()
    val target = targetSeq.toDS()
    val joined = source.join(target, source("Pk1") === target("Pk1") && source("Pk2") === target("Pk3"), "full")

    val found =
      for {
        sourceFound <- joined.queryExecution.analyzed.output.find { at => at.name == "Pk1" }
        joinedFound <- joined.queryExecution.analyzed.output.zipWithIndex.find { case (at, _) => at == sourceFound }
        ind = joinedFound._2
      } yield ind

    val sourcePk1 = joined.collect().head.getAs[Any](found.get)
    println(sourcePk1)

字符串
这样做是因为AttributeReferences在计划中是唯一的，它们只是在输出中被简单的名称丢弃。alias函数只在使用sql连接时有帮助，在代码中，您直接使用源数据集来访问列，因此使用它是正确的。
至于这是不是个好主意。queryExecution的内容和可访问性在不同版本之间可能会有所不同，它并不完全是一个用户界面。您最好直接重命名字段：

def prefixFields[T](ds: Dataset[T], prefix: String): DataFrame = {
      val newNames = ds.schema.fieldNames.map(o => col(o).as(prefix + o))
      ds.select(newNames : _*)
    }

    // prefixing with s. _may_ work, but .'s have specific meaning
    val source = prefixFields(sourceSeq.toDS(), "source_")
    val target = targetSeq.toDS()
    val joined = source.join(target, col("source_Pk1") === target("Pk1") && col("source_Pk2") === target("Pk3"), "full")

    joined.show()
    val sourcePk1 = joined.collect().head.getAs[Any]("source_Pk1")
    println(sourcePk1)

型
这是较少的内部魔法和一个有点清晰的意图。