创建 Spark 数据集时不支持 TypeTag[java.sql.Timestamp] 类型的架构

Question

我有以下代码尝试从 Seq[T] 创建 DataFrame

case class CaseConvert[T: TypeTag](a: T)

def createDf[T: TypeTag](data: Seq[T]): DataFrame = {
   spark.createDataFrame(data.map(CaseConvert[T])
}

当通过传递类型say Seq[java.sql.Timestamp]执行上述createDf方法时，它失败并显示以下错误

UnsupportedOperaionException：不支持 TypeTag[java.sql.Timestamp] 类型的架构

我想我必须为 CaseConvert 类创建编码器，但不确定如何使用 Scala 的复杂泛型来实现。我是 Spark 和 Scala 的新手。

Answer 1

像这样添加Encoder：

import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
import org.apache.spark.sql._
import scala.reflect.runtime.universe._

val spark = SparkSession.builder.getOrCreate()
import spark.implicits._

case class CaseConvert[T](a: T)

def createDf[T:  TypeTag](data: Seq[T])(implicit e: Encoder[T]): DataFrame = {
   spark.createDataFrame(data.map(CaseConvert(_)))
}

createDf(Seq(new java.sql.Timestamp(1)))
//org.apache.spark.sql.DataFrame = [a: timestamp] 

到这里就够了。 Products 的隐式解析将完成剩下的工作。