Я пытаюсь сохранить данные из фрейма данных Spark в HDFS, используя схему Avro, хранящуюся в реестре схемы. Тем не менее, я получаю сообщение об ошибке при записи данных:
Caused by: org.apache.avro.AvroRuntimeException: Not a union: {"type":"long","logicalType":"timestamp-millis"}
at org.apache.avro.Schema.getTypes(Schema.java:299)
at org.apache.spark.sql.avro.AvroSerializer.org$apache$spark$sql$avro$AvroSerializer$$resolveNullableType(AvroSerializer.scala:229)
at org.apache.spark.sql.avro.AvroSerializer$$anonfun$3.apply(AvroSerializer.scala:209)
at org.apache.spark.sql.avro.AvroSerializer$$anonfun$3.apply(AvroSerializer.scala:208)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.immutable.List.foreach(List.scala:392)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
at scala.collection.immutable.List.map(List.scala:296)
at org.apache.spark.sql.avro.AvroSerializer.newStructConverter(AvroSerializer.scala:208)
at org.apache.spark.sql.avro.AvroSerializer.<init>(AvroSerializer.scala:51)
at org.apache.spark.sql.avro.AvroOutputWriter.serializer$lzycompute(AvroOutputWriter.scala:42)
at org.apache.spark.sql.avro.AvroOutputWriter.serializer(AvroOutputWriter.scala:42)
at org.apache.spark.sql.avro.AvroOutputWriter.write(AvroOutputWriter.scala:64)
at org.apache.spark.sql.execution.datasources.SingleDirectoryDataWriter.write(FileFormatDataWriter.scala:137)
В чем может быть причина?
Поля в схеме Avro выглядят так:
{"name":"CreateDate","type":["null",{"type":"long","logicalType":"timestamp-millis"}],"default":null}
Вот пример формата даты:
1900-01-01 00:00:00
Тип данных этого поля в кадре данных Spark:
|-- CreateDate: timestamp (nullable = true)
Вот как я пишу данные:
dataDF.write
.mode("append")
.format("avro")
.option(
"avroSchema",
SchemaRegistry.getSchema(
schemaRegistryConfig.url,
schemaRegistryConfig.dataSchemaSubject,
schemaRegistryConfig.dataSchemaVersion))
.save(hdfsURL)