java - Spark sql 的 'no viable alternative at input' 是什么？

Question

我有一个 DF，其中包含包含纪元时间戳的 startTimeUnix 列(Mongo 中的 Number 类型)。我想查询此列上的 DF，但我想传递 EST 日期时间。我经历了多次在 Spark-Shell 上测试以下内容:

val df = Seq(("1", "1523937600000"), ("2", "1523941200000"),("3","1524024000000")).toDF("id", "unix")

df.filter($"unix" > java.time.ZonedDateTime.parse("04/17/2018 01:00:00", java.time.format.DateTimeFormatter.ofPattern ("MM/dd/yyyy HH:mm:ss").withZone ( java.time.ZoneId.of("America/New_York"))).toEpochSecond()*1000).collect()

输出:

= Array([3,1524024000000])

由于 java.time 函数正在运行，我将其传递给 spark-submit ，在从 Mongo 检索数据时，过滤器查询如下所示:

startTimeUnix < (java.time.ZonedDateTime.parse(${LT}, java.time.format.DateTimeFormatter.ofPattern('MM/dd/yyyyHHmmss').withZone(java.time) .ZoneId.of('America/New_York'))).toEpochSecond()*1000) AND startTimeUnix > (java.time.ZonedDateTime.parse(${GT}, java.time.format. DateTimeFormatter.ofPattern('MM/dd/yyyyHHmmss').withZone(java.time.ZoneId.of('America/New_York'))).toEpochSecond()*1000)`

但是，我不断收到以下错误:

Caused by: org.apache.spark.sql.catalyst.parser.ParseException: no viable alternative at input '(java.time.ZonedDateTime.parse(04/18/2018000000, java.time.format.DateTimeFormatter.ofPattern('MM/dd/yyyyHHmmss').withZone('(line 1, pos 138) == SQL == startTimeUnix < (java.time.ZonedDateTime.parse(04/18/2018000000, java.time.format.DateTimeFormatter.ofPattern('MM/dd/yyyyHHmmss').withZone(java.time.ZoneId.of('America/New_York'))).toEpochSecond()*1000).toString() AND startTimeUnix > (java.time.ZonedDateTime.parse(04/17/2018000000, java.time.format.DateTimeFormatter.ofPattern('MM/dd/yyyyHHmmss').withZone(java.time.ZoneId.of('America/New_York'))).toEpochSecond()*1000).toString() at org.apache.spark.sql.catalyst.parser.ParseException.withCommand(ParseDriver.scala:217) at org.apache.spark.sql.catalyst.parser.AbstractSqlParser.parse(ParseDriver.scala:114) at org.apache.spark.sql.execution.SparkSqlParser.parse(SparkSqlParser.scala:48) at org.apache.spark.sql.catalyst.parser.AbstractSqlParser.parseExpression(ParseDriver.scala:43) at org.apache.spark.sql.Dataset.filter(Dataset.scala:1315)

某处说错误意味着数据类型不匹配。我尝试将 toString 应用于日期转换的输出，但没有成功。

Answer 1

您可以使用 Spark 数据框函数。

scala> val df = Seq(("1", "1523937600000"), ("2", "1523941200000"),("3","1524024000000")).toDF("id", "unix")
df: org.apache.spark.sql.DataFrame = [id: string, unix: string]

scala> df.filter($"unix" > unix_timestamp()*1000).collect()
res5: Array[org.apache.spark.sql.Row] = Array([3,1524024000000])
scala> df.withColumn("unixinEST"
                        ,from_utc_timestamp(
                            from_unixtime(unix_timestamp()),
                             "EST"))
         .show()
+---+-------------+-------------------+
| id|         unix|          unixinEST|
+---+-------------+-------------------+
|  1|1523937600000|2018-04-18 06:13:19|
|  2|1523941200000|2018-04-18 06:13:19|
|  3|1524024000000|2018-04-18 06:13:19|
+---+-------------+-------------------+