Spark SQL的Scala API - TimestampType - 找不到org.apache.spark.sql.types.TimestampType的编码器

Question

我在Databricks笔记本上使用Spark 2.1和Scala 2.11

什么是TimestampType？

我们知道,从SparkSQL的文档那是官方的时间戳类型是TimestampType,这显然是对的java.sql.Timestamp一个别名:

TimestampType可以在SparkSQL的Scala API中找到

使用模式和数据集API时,我们有所不同

从Databricks的Scala Structured Streaming示例解析时{"time":1469501297,"action":"Open"}

使用Json架构 - > OK(我更喜欢使用优雅的Dataset API):

val jsonSchema = new StructType().add("time", TimestampType).add("action", StringType)

val staticInputDF = 
  spark
    .read
    .schema(jsonSchema)
    .json(inputPath)

使用数据集API - > KO:找不到TimestampType的编码器

创建Event类

import org.apache.spark.sql.types._
case class Event(action: String, time: TimestampType)
--> defined class Event

在databricks上从DBFS读取事件时出错.

注意:java.sql.Timestamp作为"时间"类型使用时,我们不会收到错误

val path = "/databricks-datasets/structured-streaming/events/"
val events = spark.read.json(path).as[Event]

错误信息

java.lang.UnsupportedOperationException: No Encoder found for org.apache.spark.sql.types.TimestampType
- field (class: "org.apache.spark.sql.types.TimestampType", name: "time")
- root class: 

Answer 1

结合模式读取方法.schema(jsonSchema)和as[Type]包含该类型的方法java.sql.Timestamp将解决此问题.这个想法是在阅读Structured Streaming文档创建流式数据框架和流式数据集之后得出的

这些示例生成无类型的流式DataFrame ,这意味着在编译时不检查DataFrame的架构,仅在提交查询时在运行时检查.map,flatMap等一些操作需要在编译时知道类型.要执行这些操作,您可以使用与静态DataFrame相同的方法将这些无类型的流式DataFrame转换为类型化的流式数据集.

val path = "/databricks-datasets/structured-streaming/events/"

val jsonSchema = new StructType().add("time", TimestampType).add("action", StringType)

case class Event(action: String, time: java.sql.Timestamp)

val staticInputDS = 
  spark
    .read
    .schema(jsonSchema)
    .json(path)
    .as[Event]

staticInputDF.printSchema

将输出:

root
 |-- time: timestamp (nullable = true)
 |-- action: string (nullable = true)