Styjun

I have written one of the Spark data frame columns into Kafka in Avro format. Then I try to read the data from this topic and convert from Avro to the data frame column. The type of the data is a timestamp and instead of the timestamps from the database, I get some default values:

1970-01-01 00:00:00
1970-01-01 00:00:00
1970-01-01 00:00:00
1970-01-01 00:00:00
1970-01-01 00:00:00
1970-01-01 00:00:00
1970-01-01 00:00:00
1970-01-01 00:00:00
1970-01-01 00:00:00
1970-01-01 00:00:00

The same behavior can be noticed with columns of other data types, like String. Initial timestamp value looks like this and this the result I want to obtain:

2019-03-19 12:26:03.003
2019-03-19 12:26:09 
2019-03-19 12:27:04.003
2019-03-19 12:27:08.007
2019-03-19 12:28:01.013
2019-03-19 12:28:05.007
2019-03-19 12:28:09.023 
2019-03-19 12:29:04.003
2019-03-19 12:29:07.047
2019-03-19 12:30:00.003

And here is the same data after conversion to Avro:

00 F0 E1 9B BC B3 9C C2 05
00 80 E9 F7 C1 B3 9C C2 05
00 F0 86 B2 F6 B3 9C C2 05
00 B0 E9 9A FA B3 9C C2 05
00 90 A4 E1 AC B4 9C C2 05
00 B0 EA C8 B0 B4 9C C2 05
00 B0 88 B3 B4 B4 9C C2 05
00 F0 BE EA E8 B4 9C C2 05
00 B0 89 DE EB B4 9C C2 05
00 F0 B6 9E 9E B5 9C C2 05

What can I do to fix this conversion problem?

The code for writing Avro into Kafka, reading it and converting back to the data frame. I tried to use to_avro and from_avro Spark-avro methods:

import org.apache.spark.sql.avro._

 val castDF = testDataDF.select(to_avro(testDataDF.col("update_database_time")) as 'value)

 castDF
 .write
 .format("kafka")
 .option("kafka.bootstrap.servers", bootstrapServers)
 .option("topic", "app_state_test")
 .save()

 val cachedDf = spark
 .read
 .format("kafka")
 .option("kafka.bootstrap.servers", bootstrapServers)
 .option("subscribe", "app_state_test")
 .load()

 val jsonSchema = ""name": "update_database_time", "type": "long", "logicalType": "timestamp-millis", "default": "NONE""
 cachedDf.select(from_avro(cachedDf.col("value"), jsonSchema) as 'test)