No Encoder found for java.time.Instant

[Lihe-Ma]

I have some troubles when upgrade my code to the latest sdk version 0.4.2 , but it was fine when using 0.2.1 before, here is my code:

    val events = spark.read.textFile(mySource)
      .map(line => Event.parse(line))
      .filter(_.isValid)
      .flatMap(_.toOption)    
      .map(event => event.toJson(true).noSpaces)
    val dataframe = spark.read.json(events)
    dataframe.show(100, false)

then I got error below :

Exception in thread "main" java.lang.UnsupportedOperationException: No Encoder found for java.time.Instant
option value class: "java.time.Instant"
field (class: "scala.Option", name: "etl_tstamp")
root class: "com.snowplowanalytics.snowplow.analytics.scalasdk.Event"
	at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$org$apache$spark$sql$catalyst$ScalaReflection$$serializerFor$1.apply(ScalaReflection.scala:650)
	at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$org$apache$spark$sql$catalyst$ScalaReflection$$serializerFor$1.apply(ScalaReflection.scala:452)
	at scala.reflect.internal.tpe.TypeConstraints$UndoLog.undo(TypeConstraints.scala:56)
	at org.apache.spark.sql.catalyst.ScalaReflection$class.cleanUpReflectionObjects(ScalaReflection.scala:906)
	at org.apache.spark.sql.catalyst.ScalaReflection$.cleanUpReflectionObjects(ScalaReflection.scala:46)
	at org.apache.spark.sql.catalyst.ScalaReflection$.org$apache$spark$sql$catalyst$ScalaReflection$$serializerFor(ScalaReflection.scala:452)
	at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$org$apache$spark$sql$catalyst$ScalaReflection$$serializerFor$1.apply(ScalaReflection.scala:494)
	at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$org$apache$spark$sql$catalyst$ScalaReflection$$serializerFor$1.apply(ScalaReflection.scala:452)
	at scala.reflect.internal.tpe.TypeConstraints$UndoLog.undo(TypeConstraints.scala:56)
	at org.apache.spark.sql.catalyst.ScalaReflection$class.cleanUpReflectionObjects(ScalaReflection.scala:906)
	at org.apache.spark.sql.catalyst.ScalaReflection$.cleanUpReflectionObjects(ScalaReflection.scala:46)
	at org.apache.spark.sql.catalyst.ScalaReflection$.org$apache$spark$sql$catalyst$ScalaReflection$$serializerFor(ScalaReflection.scala:452)
	at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$org$apache$spark$sql$catalyst$ScalaReflection$$serializerFor$1$$anonfun$8.apply(ScalaReflection.scala:644)
	at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$org$apache$spark$sql$catalyst$ScalaReflection$$serializerFor$1$$anonfun$8.apply(ScalaReflection.scala:632)
	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:392)
	at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:241)
	at scala.collection.immutable.List.flatMap(List.scala:355)
	at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$org$apache$spark$sql$catalyst$ScalaReflection$$serializerFor$1.apply(ScalaReflection.scala:632)
	at org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$org$apache$spark$sql$catalyst$ScalaReflection$$serializerFor$1.apply(ScalaReflection.scala:452)
	at scala.reflect.internal.tpe.TypeConstraints$UndoLog.undo(TypeConstraints.scala:56)
	at org.apache.spark.sql.catalyst.ScalaReflection$class.cleanUpReflectionObjects(ScalaReflection.scala:906)
	at org.apache.spark.sql.catalyst.ScalaReflection$.cleanUpReflectionObjects(ScalaReflection.scala:46)
	at org.apache.spark.sql.catalyst.ScalaReflection$.org$apache$spark$sql$catalyst$ScalaReflection$$serializerFor(ScalaReflection.scala:452)
	at org.apache.spark.sql.catalyst.ScalaReflection$.serializerFor(ScalaReflection.scala:441)
	at org.apache.spark.sql.catalyst.encoders.ExpressionEncoder$.apply(ExpressionEncoder.scala:71)
	at org.apache.spark.sql.Encoders$.product(Encoders.scala:275)
	at org.apache.spark.sql.LowPrioritySQLImplicits$class.newProductEncoder(SQLImplicits.scala:248)
	at org.apache.spark.sql.SQLImplicits.newProductEncoder(SQLImplicits.scala:34)
	at org.mlh.hello.myTest$.main(myTest.scala:16)
	at org.mlh.hello.myTest.main(myTest.scala)

line 16 is .flatMap(_.toOption) , it seems that the Instant type in Event class cannnot be dealt correctly.
Is there any ideas ?

[chuwy]

Hello @take-it-out,

What version of Scala/Spark are you using? The exception is quite strange because it tells us Spark cannot find its own encoder for Instant whereas Instant is encoded into string by Analytics SDK and circe (the JSON library) themselves, so no Spark serialization should be involved.

(Just in case - you don't need to use .isValid because .toOption will throw away invalid values anyway)

[Lihe-Ma]

thanks for you reply, I am using scala 2.12.8 and spark 2.4.0 .

[chuwy]

Thanks, we'll try out these versions (don't think we did). Meanwhile, could you please try this code:

val events = spark.read.textFile(mySource)
  .flatMap(line => Event.parse(line).toOption.map(_.toJson(true).noSpaces))

Because I was wrong saying that "Spark cannot find its own encoder for Instant whereas Instant is encoded into string by Analytics SDK" - in your code Instant exists between RDD.map invocations and hence has to be serialized.

[Lihe-Ma]

@chuwy thx, it works.

[chuwy]

Hey @take-it-out, hope you don't mind if it stays open for a bit. My concern is that our production systems use Analytics SDK in a way you shown with Spark 2.3, but I'm wondering why it doesn't work on 2.4.