scala - Spark `unionAll`的 `DataFrame`怎么了？

Question

使用Spark 1.5.0并给出以下代码，我希望unionAll根据其列名来合并DataFrame。在代码中，我使用了一些FunSuite来传递SparkContext sc:

object Entities {

  case class A (a: Int, b: Int)
  case class B (b: Int, a: Int)

  val as = Seq(
    A(1,3),
    A(2,4)
  )

  val bs = Seq(
    B(5,3),
    B(6,4)
  )
}

class UnsortedTestSuite extends SparkFunSuite {

  configuredUnitTest("The truth test.") { sc =>
    val sqlContext = new SQLContext(sc)
    import sqlContext.implicits._
    val aDF = sc.parallelize(Entities.as, 4).toDF
    val bDF = sc.parallelize(Entities.bs, 4).toDF
    aDF.show()
    bDF.show()
    aDF.unionAll(bDF).show
  }
}

输出:

+---+---+
|  a|  b|
+---+---+
|  1|  3|
|  2|  4|
+---+---+

+---+---+
|  b|  a|
+---+---+
|  5|  3|
|  6|  4|
+---+---+

+---+---+
|  a|  b|
+---+---+
|  1|  3|
|  2|  4|
|  5|  3|
|  6|  4|
+---+---+

为什么结果中包含 混合的“b”和“a” 列，而不是根据列名对齐列？听起来像是 严重的错误!？

Answer 1

它看起来根本不是一个bug。您将看到一个标准的SQL行为，并且每个主要RDMBS(包括PostgreSQL，MySQL，Oracle和MS SQL)的行为都完全相同。您会找到与名称链接的SQL Fiddle示例。

引用PostgreSQL manual:

In order to calculate the union, intersection, or difference of two queries, the two queries must be "union compatible", which means that they return the same number of columns and the corresponding columns have compatible data types

列名(不包括set操作中的第一个表)将被忽略。

此行为直接来自关系代数，其中基本构建块是元组。因为元组是有序的，所以两组元组的并集等同于此处得到的输出(忽略重复处理)。

如果要使用名称进行匹配，可以执行以下操作

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

def unionByName(a: DataFrame, b: DataFrame): DataFrame = {
  val columns = a.columns.toSet.intersect(b.columns.toSet).map(col).toSeq
  a.select(columns: _*).unionAll(b.select(columns: _*))
}

要检查名称和类型，只需将 columns替换为:

a.dtypes.toSet.intersect(b.dtypes.toSet).map{case (c, _) => col(c)}.toSeq