0

我有以下 Scala 函数，已在 Azure DataBricks（Spark 2.3.1 和 Scala 2.11）中注册为 UDF：

import org.joda.time.DateTime

val slot = (dt : DateTime) => { 
  var _s : Int = (dt.dayOfYear().get() - 1) * 24 + dt.hourOfDay().get()
  _s  
}

val compute_slot = (t: String, offset: Int) => {
  var dt:DateTime = DateTime.parse(t) 
  ((Set(slot(dt)) ++ Set(slot(dt.minusSeconds(offset)))) ++ Set(slot    (dt.plusSeconds(offset)))).toArray  
}

    spark.udf.register("get_slot", compute_slot)

如果直接调用该函数效果很好...

compute_slot("2018-11-01T05:04:33.827+0000", 1800)

...或作为 SQL 中的 UDF：

%sql
SELECT explode(get_slot(cast("2018-11-01T05:04:33.827+0000" as string), 1800))

在 SparkSql SQL 查询中使用 UDF 时……</p> 

%sql
SELECT * FROM tab LATERAL VIEW explode(get_slot(cast(timestamp as string), 1800)) my_view

… 或者

val ColUDF = udf(compute_slot)
df.withColumn("arr", ColUDF($"timestamp", lit(1800))).show()

它失败并显示以下错误消息。它提到了错误的日期格式，这让我感到困惑，因为该函数在我的表上的 SELECT 语句之外运行良好。

org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 177.0 failed 4 times, most recent failure: Lost task 0.3 in stage 177.0 (TID 33992, 10.139.64.4, executor 4): org.apache.spark.SparkException: Failed to execute user defined function($anonfun$2: (string, int) => array<int>)

at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage2.processNext(Unknown Source)
    at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
    at org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$10$$anon$1.hasNext(WholeStageCodegenExec.scala:620)
    at org.apache.spark.sql.execution.collect.UnsafeRowBatchUtils$.encodeUnsafeRows(UnsafeRowBatchUtils.scala:49)
    at org.apache.spark.sql.execution.collect.Collector$$anonfun$2.apply(Collector.scala:126)
    at org.apache.spark.sql.execution.collect.Collector$$anonfun$2.apply(Collector.scala:125)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
    at org.apache.spark.scheduler.Task.run(Task.scala:112)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:384)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.IllegalArgumentException: Invalid format: "2018-11-01 05:04:33.827" is malformed at " 05:04:33.827"
    at org.joda.time.format.DateTimeFormatter.parseDateTime(DateTimeFormatter.java:945)
    at org.joda.time.DateTime.parse(DateTime.java:160)
    at org.joda.time.DateTime.parse(DateTime.java:149)
    at line7c3ca3974ac14b88a9a351882d40a949342.$read$$iw$$iw$$iw$$iw$$iw$$iw$$anonfun$2.apply(command-1611582591913421:11)
    at line7c3ca3974ac14b88a9a351882d40a949342.$read$$iw$$iw$$iw$$iw$$iw$$iw$$anonfun$2.apply(command-1611582591913421:10)
    ... 12 more

任何帮助集中我做错的事情将不胜感激！

非常感谢！

保罗

4

2 回答 2

1

该列timestamp的格式不正确。它在错误中说：

引起：java.lang.IllegalArgumentException：无效格式：“2018-11-01 05:04:33.827”在“05:04:33.827”格式错误

此外，这失败了：

get_slot(cast(timestamp as string), 1800)

但这成功了：

get_slot(cast("2018-11-01T05:04:33.827+0000" as string), 1800)

因此，您的timestamp列的格式根本不正确。只需添加T部分，你会没事的。

于 2018-12-04T21:32:19.313 回答
1

默认情况下parse使用ISODateTimeFormat定义为的模式：

datetime          = time | date-opt-time
time              = 'T' time-element [offset]
date-opt-time     = date-element ['T' [time-element] [offset]]
date-element      = std-date-element | ord-date-element | week-date-element
std-date-element  = yyyy ['-' MM ['-' dd]]
ord-date-element  = yyyy ['-' DDD]
week-date-element = xxxx '-W' ww ['-' e]
time-element      = HH [minute-element] | [fraction]
minute-element    = ':' mm [second-element] | [fraction]
second-element    = ':' ss [fraction]
fraction          = ('.' | ',') digit+
offset            = 'Z' | (('+' | '-') HH [':' mm [':' ss [('.' | ',') SSS]]])

而您的日期不包含offset组件。为了匹配2018-11-01 05:04:33.827，您应该提供一种格式：

scala> import org.joda.time.format.DateTimeFormat
import org.joda.time.format.DateTimeFormat
scala> DateTime.parse("2018-11-01 05:04:33.827", DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss.SSS"));
res3: org.joda.time.DateTime = 2018-11-01T05:04:33.827+01:00
于 2018-12-04T21:32:08.193 回答