scala – “unionAll”的Spark“DataFrame”出了什么问题？

发布时间：2020-12-16 21:31:26 所属栏目：安全来源：网络整理

导读：使用Spark 1.5.0并给出以下代码,我希望unionAll根据它们的列名称联合DataFrames.在代码中,我使用一些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

使用Spark 1.5.0并给出以下代码,我希望unionAll根据它们的列名称联合DataFrames.在代码中,我使用一些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,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”列,而不是按列名称对齐列？听起来像一个严重的错误！

解决方法

它看起来不像一个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

列名称(不包括设置操作中的第一个表)将被忽略.

这种行为直接来自关系代数,其中基本构建块是元组.由于元组被命令,两组元组的联合是相当于(忽略重复处理)到您在这里得到的输出.

如果你想使用名称匹配,你可以这样做

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: _*))
}

要检查这两个名称和类型,应该足以用以下替换列：

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

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!