如何更新 Scala 中日期格式的列

Question

在 Spark Scala 中需要帮助来编写以下问题的代码。 我有一个包含如下记录的文件。

aaa|2019-07-11 02:15:50

bbb|2019-07-03 22:21:50

vvv|2019-07-03 19:30:40

bzx|2019-07-11 02:15:30

rrr|2019-06-24 01:29:10

mmm|2019-06-23 20:35:05

qqq|2019-07-12 08:10:15

eee|2019-07-11 01:49:30

iii|2019-06-23 22:31:45

我已拆分文件并取了第二列

val file = spark.read.format("csv").option("delimiter", """|""").load(pathOfDumpfile).toDF()  

现在我需要将“0000-00-00 00:00:05”添加到所有 文件的值（日期格式的第二列）并将其另存为文件，如下所示

aaa|2019-07-11 02:15:55

bbb|2019-07-03 22:21:55

vvv|2019-07-03 19:30:45

bzx|2019-07-11 02:15:35

rrr|2019-06-24 01:29:15

mmm|2019-06-23 20:35:10

qqq|2019-07-12 08:10:20

eee|2019-07-11 01:49:35

iii|2019-06-23 22:31:50

谁能建议我如何将 + 5 秒添加到文件/列中的所有记录。

真的很有帮助。在日期时间字段中添加后，秒或分钟应该改变，它不应该像它的 2019-07-11 23:59:59 那样影响日期，然后即使在秒上加 1 也会是 2019-07-12 00： 00:00。我想添加但不想更改日期，所以我怎么做

Answer 1

你可以使用 unix_timestamp 来做

scala>  var dfv = Seq(("aaa","2019-07-11 23:59:59"),("bbb","2019-07-03 22:21:50"),("vvv","2019-07-03 19:30:40"),("bzx","2019-07-11 02:15:30"),("rrr","2019-06-24 01:29:10"),("mmm","2019-06-23 20:35:05"),("qqq","2019-07-12 08:10:15"),("eee","2019-07-11 01:49:30"),("iii","2019-06-23 22:31:45")).toDF("value","_date")

scala> dfv.show
+-----+-------------------+
|value|              _date|
+-----+-------------------+
|  aaa|2019-07-11 23:59:59|
|  bbb|2019-07-03 22:21:50|
|  vvv|2019-07-03 19:30:40|
|  bzx|2019-07-11 02:15:30|
|  rrr|2019-06-24 01:29:10|
|  mmm|2019-06-23 20:35:05|
|  qqq|2019-07-12 08:10:15|
|  eee|2019-07-11 01:49:30|
|  iii|2019-06-23 22:31:45|
+-----+-------------------+

scala> dfv.withColumn("_date_v1",when(date_format(from_unixtime(unix_timestamp($"_date")),"HH:mm:ss ")>"23:59:55",$"_date").otherwise(from_unixtime(unix_timestamp($"_date")+5,"yyyy-MM-dd HH:mm:ss"))).show
+-----+-------------------+-------------------+
|value|              _date|           _date_v1|
+-----+-------------------+-------------------+
|  aaa|2019-07-11 23:59:59|2019-07-11 23:59:59|
|  bbb|2019-07-03 22:21:50|2019-07-03 22:21:55|
|  vvv|2019-07-03 19:30:40|2019-07-03 19:30:45|
|  bzx|2019-07-11 02:15:30|2019-07-11 02:15:35|
|  rrr|2019-06-24 01:29:10|2019-06-24 01:29:15|
|  mmm|2019-06-23 20:35:05|2019-06-23 20:35:10|
|  qqq|2019-07-12 08:10:15|2019-07-12 08:10:20|
|  eee|2019-07-11 01:49:30|2019-07-11 01:49:35|
|  iii|2019-06-23 22:31:45|2019-06-23 22:31:50|
+-----+-------------------+-------------------+

如果您有任何与此相关的问题，请告诉我。

Answer 2

您可以在自定义 udf 的帮助下执行此操作，例如：

import org.apache.spark.sql.functions.{col, udf}

val file = spark.read.format("csv").option("delimiter", """|""").load(pathOfDumpfile).toDF("a", "b")

val timeUDF = udf((x: java.sql.Timestamp) => new java.sql.Timestamp(time.getTime + 5000)) //getTime returns ms

file.select(col("a"), timeUDF(col("b")))
.write(...)

Answer 3

您可以在 spark 中使用 INTERVAL 语法。

使用数据框：

val foo = spark.sql(""" select current_timestamp as ts """)
foo.select($"ts", $"ts" + expr("INTERVAL 5 SECONDS") as "ts_plus").show(false)
+-----------------------+-----------------------+
|ts                     |ts_plus                |
+-----------------------+-----------------------+
|2019-09-16 10:33:17.626|2019-09-16 10:33:22.626|
+-----------------------+-----------------------+

使用 Spark SQL：

foo.createOrReplaceTempView("fooView")
spark.sql(""" select ts, ts + INTERVAL 5 seconds from fooView""").show(false)
+-----------------------+------------------------------------------+
|ts                     |CAST(ts + interval 5 seconds AS TIMESTAMP)|
+-----------------------+------------------------------------------+
|2019-09-16 10:35:12.847|2019-09-16 10:35:17.847                   |
+-----------------------+------------------------------------------+