Myujth

Schema has the dataType declared as Timestamp but spark job is not converting it in the right format.

Dataset<Row> stream = sparkSession.readStream()
 .format("kafka")
 .option("kafka.bootstrap.servers", kafkaBootstrapServersString)
 .option("subscribe", topic)
// .option("maxOffsetsPerTrigger", 10000)
 .load();

 Dataset<Row> rawStream = stream
 .selectExpr("CAST(value AS STRING)")
 .select(from_json(col("value"), eventSpecificStructType).as("eventData"))
 .select("eventData.*")
 .filter(col("eventType").equalTo("Test"));

Timestamp coming in as 1542126896113 gets converted to 50838-01-28 18:49:111.0.

Is there a way to cast milliseconds to datetime format?

You will have to create a UDF in Java.

import java.sql.Timestamp;
import java.text.SimpleDateFormat;

SimpleDateFormat dateFormat = new SimpleDateFormat("....Date time pattern...");
spark.udf().register("timestamp", new UDF1<String, Timestamp>() 
 private static final long serialVersionUID = 1335972766810808134L;
 @Override
 public Timestamp call(String source)
 
 try
 return new Timestamp(dateFormat.parse(source).getTime());
 catch (ParseException e) 
 e.printStackTrace();
 
 
 return null;
 
 }, DataTypes.TimestampType);

Finally:

stream = stream.withColumn("col", callUDF("timestamp", dataframe.col("col")));

How about dividing the millisecond value by 1000. Is below meets your expectation?

val df = Seq(("1542126896113"),("1542126896116")).toDF("unixtime")
 df.withColumn("times",from_unixtime('unixtime.cast("long")/1000)).show(false)

Output

+-------------+-------------------+
|unixtime |times |
+-------------+-------------------+
|1542126896113|2018-11-13 22:04:56|
|1542126896116|2018-11-13 22:04:56|
+-------------+-------------------+