scala - 如何在数据集中使用 java.time.LocalDate(失败并显示 java.lang.UnsupportedOperationException : No Encoder found)?

Question

这个问题已经有答案了:

How to store custom objects in Dataset? (9 个回答)

已关闭 6 年前。

• Spark 2.1.1
• Scala 2.11.8
• Java 8
• Linux Ubuntu 16.04 LTS

我想将 RDD 转换为数据集。为此，我使用 implicits方法toDS()这给了我以下错误:

Exception in thread "main" java.lang.UnsupportedOperationException: No Encoder found for java.time.LocalDate
- field (class: "java.time.LocalDate", name: "date")
- root class: "observatory.TemperatureRow"
    at org.apache.spark.sql.catalyst.ScalaReflection$.org$apache$spark$sql$catalyst$ScalaReflection$$serializerFor(ScalaReflection.scala:602)
    at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$9.apply(ScalaReflection.scala:596)
    at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$9.apply(ScalaReflection.scala:587)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
    at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
    at scala.collection.immutable.List.foreach(List.scala:381)
    at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:241)
    at scala.collection.immutable.List.flatMap(List.scala:344)

就我而言，我必须使用类型 java.time.LocalDate ，我无法使用java.sql.data 。我读到我需要通知 Spark 如何将 Java 类型转换为 Sql 类型，我在这个方向上构建了下面的 2 个隐式函数:

implicit def toSerialized(t: TemperatureRow): EncodedTemperatureRow = EncodedTemperatureRow(t.date.toString, t.location, t.temperature)
implicit def fromSerialized(t: EncodedTemperatureRow): TemperatureRow = TemperatureRow(LocalDate.parse(t.date), t.location, t.temperature)

<小时/>

下面是一些关于我的应用程序的代码:

case class Location(lat: Double, lon: Double)

case class TemperatureRow(
                             date: LocalDate,
                             location: Location,
                             temperature: Double
                         )

case class EncodedTemperatureRow(
                             date: String,
                             location: Location,
                             temperature: Double

val s = Seq[TemperatureRow](
                    TemperatureRow(LocalDate.parse("2017-01-01"), Location(1.4,5.1), 4.9),
                    TemperatureRow(LocalDate.parse("2014-04-05"), Location(1.5,2.5), 5.5)
                )

import spark.implicits._
val temps: RDD[TemperatureRow] = sc.parallelize(s)
val tempsDS = temps.toDS

我不知道为什么 Spark 在编码器中搜索 java.time.LocalDate ，我为 TemperatureRow 提供隐式转换和EncodedTemperatureRow ...

Answer 1

Spark 2.2 之前不支持

java.time.LocalDate (并且我一直在尝试为该类型编写 Encoder 一段时间，并且 failed )。

您必须将 java.time.LocalDate 转换为其他受支持的类型(例如 java.sql.Timestamp 或 java.sql.Date >)，或字符串中的纪元或日期时间。