Spark SQL - 通用数据集阅读器答案

【问题标题】：Spark SQL - generic Dataset readerSpark SQL - 通用数据集阅读器
【发布时间】：2018-02-23 21:35:34
【问题描述】：

我正在尝试创建通用 DataSet[T] 阅读器，以避免每个阅读器调用的 dataframe.as[..]。支持原始类型和案例类，所以我在考虑类似的东西：

def read[T <: Product](sql : String): Dataset[T] = {
  import sparkSession.implicits._
  val sqlContext = sparkSession.sqlContext
  val df: DataFrame = sqlContext.read.option("query", sql).load()
  df.as[T]
}

但我收到“无法找到存储在数据集中的类型的编码器”错误。有没有可能做这样的事情？

第二个周期：

def read[T <: Product](sql : String) : Dataset[T] = {
  import sparkSession.implicits._
  innerRead(sql)
}

private def innerRead[T <: Product : Encoder](sql : String): Dataset[T] = {
  val sqlContext = sparkSession.sqlContext
  val df: DataFrame = sqlContext.read.option("query", sql).load()
  df.as[T]
}

以类型不匹配结尾（找到 Encoder[Nothing] ，需要 Encoder[T]）。

我试图仅导入 newProductEncoder，但结果相同。

【问题讨论】：

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

【解决方案1】：

要将DataFrame 转换为Dataset，您需要有一个Encoder。你可以通过简单地为T添加一个上下文绑定和Encoder：

def read[T <: Product : Encoder](sql : String): Dataset[T] = {
  import sparkSession.implicits._
  val sqlContext = sparkSession.sqlContext
  val df: DataFrame = sqlContext.read.option("query", sql).load()
  df.as[T]
}

上下文绑定是以下内容的语法糖：

def read[T <: Product](sql : String)(implicit $ev: Encoder[T]): Dataset[T]

这意味着您需要在隐式上下文中拥有一个（并且只有一个）Encoder[T] 实例。

这是必需的，因为as 方法本身需要此上下文绑定。

Spark 本身可以通过导入（如您所做的）您的 SparkSession 的隐式来为您提供您可能需要的大部分 Encoders（原语、Strings 和 case classes）。但是，这些必须在调用站点的隐式范围内可用，这意味着您想要拥有的可能更像以下内容：

def read[T <: Product : Encoder](spark: SparkSession, sql: String): Dataset[T] = {
  import spark.implicits._
  val df: DataFrame = spark.sqlContext.read.option("query", sql).load()
  df.as[T]
}

val spark: SparkSession = ??? // your SparkSession object
import spark.implicits._
val ds: Dataset[YourType] = read[YourType](spark, "select something from a_table")

【讨论】：

谢谢，我用新的细节扩展了原来的问题。
read 没有指定上下文绑定，这意味着innerRead 它看不到它。上下文绑定必须在您的调用中传播。但是，为了使隐式在正确的时刻可用，您必须在调用站点而不是在定义方法时导入它们。

【解决方案2】：

在您的第二个周期中，也许您需要将类型参数提供给您的 innerRead 调用：

innerRead[T](sql)

【讨论】：