如何使用 Scala 将 DataSet(s) 传递给接受 DataFrame(s) 作为 Apache Spark 中的参数的函数？答案

【问题标题】：How to pass DataSet(s) to a function that accepts DataFrame(s) as arguments in Apache Spark using Scala?如何使用 Scala 将 DataSet(s) 传递给接受 DataFrame(s) 作为 Apache Spark 中的参数的函数？
【发布时间】：2021-01-20 03:59:16
【问题描述】：

我在 Scala 中有一个用于 Spark 的库，其中包含许多函数。一个示例是以下函数，用于合并具有不同列的两个数据框：

def appendDF(df2: DataFrame): DataFrame = {

  val cols1 = df.columns.toSeq
  val cols2 = df2.columns.toSeq

  def expr(sourceCols: Seq[String], targetCols: Seq[String]): Seq[Column] = {
    targetCols.map({
      case x if sourceCols.contains(x) => col(x)
      case y                           => lit(null).as(y)
    })
  }

  // both df's need to pass through `expr` to guarantee the same order, as needed for correct unions.
  df.select(expr(cols1, cols1): _*).union(df2.select(expr(cols2, cols1): _*))

}

我想将此功能（以及更多功能）用于Dataset[CleanRow]，而不是 DataFrames。 CleanRow 是一个简单的类，它定义了列的名称和类型。我有根据的猜测是使用.toDF() 方法将数据集转换为数据帧。但是，我想知道是否有更好的方法。

据我了解，Dataset 和 Dataframe 之间应该没有太多区别，因为 Dataset 只是 Dataframe[Row]。另外，我认为从 Spark 2.x 开始，DF 和 DS 的 API 已经统一，所以我想我可以互换地通过它们中的任何一个，但事实并非如此。

【问题讨论】：

如果方法的签名不能改变（例如接受泛型），我猜你必须做Dataset.toDF()，否则如果你能改变签名，你能做到吗def appendDF(ds: DataSet[A]) 可以带Dataset[Row] 和Dataset[T]?
我明白了。所以这是除了更改方法签名之外的唯一选择。它是否被认为是好的做法（例如：对于生产级代码）？此外，关于您更改签名的建议，如果我将其更改为 Dataset[A] 那么它也可以作为参数somedata.toDF() 对吗？这只是出于好奇。
是的，我在回答中发布了。

标签： scala apache-spark apache-spark-sql apache-spark-dataset

【解决方案1】：

如果可以更改签名：

import spark.implicits._
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.Dataset

def f[T](d: Dataset[T]): Dataset[T] = {d}

// You are able to pass a dataframe:
f(Seq(0,1).toDF()).show
// res1: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [value: int]

// You are also able to pass a dataset:
f(spark.createDataset(Seq(0,1)))
// res2: org.apache.spark.sql.Dataset[Int] = [value: int]

【讨论】：