首先，应该将秒传递到毫秒，然后转换为时间戳或日期

object ToTimestamp exteds App{
  val spark = SparkSession
    .builder()
    .appName("ToTimestamp")
    .master("local[*]")
    .config("spark.sql.shuffle.partitions","4") //Change to a more reasonable default number of partitions for our data
    .config("spark.app.id","ToTimestamp") // To silence Metrics warning
    .getOrCreate()

  val sc = spark.sparkContext

  import org.apache.spark.sql.functions._
  import spark.implicits._

  val data = sc.parallelize(List(1589509800768L,1589509802730L,1589509809092L,1589509810402L)).toDF("millis")

  val toTimestamp = data.withColumn("timestamp", from_unixtime(col("millis") / 1000))
  toTimestamp.show(truncate = false)
/*
+-------------+-------------------+
|millis       |timestamp          |
+-------------+-------------------+
|1589509800768|2020-05-15 04:30:00|
|1589509802730|2020-05-15 04:30:02|
|1589509809092|2020-05-15 04:30:09|
|1589509810402|2020-05-15 04:30:10|
+-------------+-------------------+

* /

  val toDate = toTimestamp.selectExpr("millis", "timestamp").withColumn("data", to_date(col("timestamp")))
  toDate.show(truncate = false)

/*
+-------------+-------------------+----------+
|millis       |timestamp          |data      |
+-------------+-------------------+----------+
|1589509800768|2020-05-15 04:30:00|2020-05-15|
|1589509802730|2020-05-15 04:30:02|2020-05-15|
|1589509809092|2020-05-15 04:30:09|2020-05-15|
|1589509810402|2020-05-15 04:30:10|2020-05-15|
+-------------+-------------------+----------+

* /

}

spark需要以秒为单位的epoch时间，而不是毫秒，因此可以将它除以1000。

scala> val values = List(1589509800768L)
values: List[Long] = List(1589509800768)

scala> val df = values.toDF()
df: org.apache.spark.sql.DataFrame = [value: bigint]

scala> df.show(false)
+-------------+
|value        |
+-------------+
|1589509800768|
+-------------+

scala> df.select((col("value") / 1000 ).cast(TimestampType).as("current_time")).show(false)
+-----------------------+
|current_time           |
+-----------------------+
|2020-05-14 19:30:00.768|
+-----------------------+

scala> df.select((col("value") / 1000 ).cast(TimestampType).as("current_time")).withColumn("time_utc",
     |   expr("""to_utc_timestamp(current_time, "PST")""")
     | ).show(false)
+-----------------------+-----------------------+
|current_time           |time_utc               |
+-----------------------+-----------------------+
|2020-05-14 19:30:00.768|2020-05-15 02:30:00.768|
+-----------------------+-----------------------+