Parsing epoch milliseconds from json with Spark 2

Has anyone parsed a millisecond timestamp using from_json in Spark 2+? How's it done?

So Spark changed the TimestampType to parse epoch numerical values as being in seconds instead of millis in v2.

My input is a hive table that has a json formatted string in a column which I'm trying to parse like this:

val spark = SparkSession
  .builder
  .appName("Problematic Timestamps")
  .enableHiveSupport()
  .getOrCreate()
import spark.implicits._
val schema = StructType(
  StructField("categoryId", LongType) ::
  StructField("cleared", BooleanType) ::
  StructField("dataVersion", LongType) ::
  StructField("details", DataTypes.createArrayType(StringType)) ::
  …
  StructField("timestamp", TimestampType) ::
  StructField("version", StringType) :: Nil
)
val item_parsed =
    spark.sql("select * FROM source.jsonStrInOrc")
    .select('itemid, 'locale,
            from_json('internalitem, schema)
                as 'internalitem,
            'version, 'createdat, 'modifiedat)
val item_flattened = item_parsed
    .select('itemid, 'locale,
            $"internalitem.*",
            'version as'outer_version, 'createdat, 'modifiedat)

This can parse a row with a column containing:

{"timestamp": 1494790299549, "cleared": false, "version": "V1", "dataVersion": 2, "categoryId": 2641, "details": [], …}

And that gives me timestamp fields like 49338-01-08 00:39:09.0 from a value 1494790299549 which I'd rather read as: 2017-05-14 19:31:39.549

Now I could set the schema for timestamp to be a long, then divide the value by 1000 and cast to a timestamp, but then I'd have 2017-05-14 19:31:39.000 not 2017-05-14 19:31:39.549. I'm having trouble figuring out how I could either:

• Tell from_json to parse a millisecond timestamp (maybe by subclassing the TimestampType in some way to use in the schema)
• Use a LongType in the schema and cast that to a Timestamp which preserves the milliseconds.

Addendum on UDFs

I found that trying to do the division in the select and then casting didn't look clean to me, though it's a perfectly valid method. I opted for a UDF that used a java.sql.timestamp which is actually specified in epoch milliseconds.

import java.sql.Timestamp

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{explode, from_json, udf}
import org.apache.spark.sql.types.
{BooleanType, DataTypes, IntegerType, LongType,
StringType, StructField, StructType, TimestampType}

val tsmillis = udf { t: Long => new Timestamp (t) }

val spark = SparkSession
  .builder
  .appName("Problematic Timestamps")
  .enableHiveSupport()
  .getOrCreate()
import spark.implicits._
val schema = StructType(
  StructField("categoryId", LongType) ::
  StructField("cleared", BooleanType) ::
  StructField("dataVersion", LongType) ::
  StructField("details", DataTypes.createArrayType(StringType)) ::
  …
  StructField("timestamp", LongType) ::
  StructField("version", StringType) :: Nil
)
val item_parsed =
    spark.sql("select * FROM source.jsonStrInOrc")
    .select('itemid, 'locale,
            from_json('internalitem, schema)
                as 'internalitem,
            'version, 'createdat, 'modifiedat)
val item_flattened = item_parsed
    .select('itemid, 'locale,
            $"internalitem.categoryId", $"internalitem.cleared",
            $"internalitem.dataVersion", $"internalitem.details",
            tsmillis($"internalitem.timestamp"),
            $"internalitem.version",
            'version as'outer_version, 'createdat, 'modifiedat)

See how that's in the select. I think it would be worthwhile to do a performance test to see if using withcolumn division and casting is faster than the udf.